blob: 18296c5d1ca70cbbe5b611a85023f16271ce7ff6 (
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
|
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.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 = Logger.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 = connection.upload( sourceFile.getAbsolutePath() );
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();
}
}
|
