blob: ad2e96c79a32f22978b53cec88bc63b941c0208f (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
package org.openslx.filetransfer.util;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.openslx.bwlp.thrift.iface.TransferInformation;
import org.openslx.filetransfer.Uploader;
public abstract class OutgoingTransferBase extends AbstractTransfer
{
/*
* Constants
*/
private static final Logger LOGGER = LogManager.getLogger( OutgoingTransferBase.class );
private static final long INACTIVITY_TIMEOUT = TimeUnit.MINUTES.toMillis( 5 );
/*
* Overridable constants
*/
protected static int MAX_CONNECTIONS_PER_TRANSFER = 2;
/*
* Class members
*/
/**
* Remote peer is downloading, so we have Uploaders
*/
private final List<Uploader> uploads = new ArrayList<>();
/**
* File being uploaded
*/
private final File sourceFile;
private final TransferInformation transferInformation;
public OutgoingTransferBase( String transferId, File sourceFile, int plainPort, int sslPort )
{
super( transferId );
this.sourceFile = sourceFile;
this.transferInformation = new TransferInformation( transferId, plainPort, sslPort );
}
/**
* Add another connection for this file transfer.
*
* @param connection
* @return true if the connection is accepted, false if it should be
* discarded
*/
public synchronized boolean addConnection( final Uploader connection, ExecutorService pool )
{
synchronized ( uploads ) {
if ( uploads.size() >= MAX_CONNECTIONS_PER_TRANSFER )
return false;
uploads.add( connection );
}
return runConnectionInternal( connection, pool );
}
protected boolean runConnectionInternal( final Uploader connection, ExecutorService pool )
{
try {
pool.execute( new Runnable() {
@Override
public void run()
{
boolean ret = false;
try {
ret = connection.upload( sourceFile.getAbsolutePath() );
} finally {
synchronized ( uploads ) {
uploads.remove( connection );
}
}
if ( ret ) {
connectFails.set( 0 );
}
if ( ret && uploads.isEmpty() && potentialFinishTime.get() == 0 ) {
potentialFinishTime.set( System.currentTimeMillis() );
}
if ( !ret && uploads.isEmpty() ) {
connectFails.incrementAndGet();
}
lastActivityTime.set( System.currentTimeMillis() );
}
} );
} catch ( Exception e ) {
LOGGER.warn( "threadpool rejected the incoming file transfer", e );
synchronized ( uploads ) {
uploads.remove( connection );
}
return false;
}
return true;
}
@Override
public TransferInformation getTransferInfo()
{
return transferInformation;
}
@Override
public final boolean isActive()
{
return uploads.size() > 0 || lastActivityTime.get() + INACTIVITY_TIMEOUT > System.currentTimeMillis();
}
@Override
public void cancel()
{
// Void
}
@Override
public final int getActiveConnectionCount()
{
return uploads.size();
}
}
|
