forked from I2P_Developers/i2p.i2p
Added {get,set}SOTimeout() to the ServerSocket API,
and fixed all the broken mainstream applications depending on it. Fixed a grave bug in SimpleTimer. Fixed Steraming Timer to be public. Fixed a pile of JavaDoc comments, and reformatted the files I touched.
This commit is contained in:
sponge
@@ -1,5 +1,6 @@
|
||||
package net.i2p.client.streaming;
|
||||
|
||||
import java.net.SocketTimeoutException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
|
||||
@@ -21,393 +21,459 @@ import net.i2p.util.SimpleTimer;
|
||||
*
|
||||
*/
|
||||
public class ConnectionManager {
|
||||
private I2PAppContext _context;
|
||||
private Log _log;
|
||||
private I2PSession _session;
|
||||
private MessageHandler _messageHandler;
|
||||
private PacketHandler _packetHandler;
|
||||
private ConnectionHandler _connectionHandler;
|
||||
private PacketQueue _outboundQueue;
|
||||
private SchedulerChooser _schedulerChooser;
|
||||
private ConnectionPacketHandler _conPacketHandler;
|
||||
/** Inbound stream ID (Long) to Connection map */
|
||||
private Map _connectionByInboundId;
|
||||
/** Ping ID (Long) to PingRequest */
|
||||
private Map _pendingPings;
|
||||
private boolean _allowIncoming;
|
||||
private int _maxConcurrentStreams;
|
||||
private ConnectionOptions _defaultOptions;
|
||||
private volatile int _numWaiting;
|
||||
private Object _connectionLock;
|
||||
|
||||
public ConnectionManager(I2PAppContext context, I2PSession session, int maxConcurrent, ConnectionOptions defaultOptions) {
|
||||
_context = context;
|
||||
_log = context.logManager().getLog(ConnectionManager.class);
|
||||
_connectionByInboundId = new HashMap(32);
|
||||
_pendingPings = new HashMap(4);
|
||||
_connectionLock = new Object();
|
||||
_messageHandler = new MessageHandler(context, this);
|
||||
_packetHandler = new PacketHandler(context, this);
|
||||
_connectionHandler = new ConnectionHandler(context, this);
|
||||
_schedulerChooser = new SchedulerChooser(context);
|
||||
_conPacketHandler = new ConnectionPacketHandler(context);
|
||||
_session = session;
|
||||
session.setSessionListener(_messageHandler);
|
||||
_outboundQueue = new PacketQueue(context, session, this);
|
||||
_allowIncoming = false;
|
||||
_maxConcurrentStreams = maxConcurrent;
|
||||
_defaultOptions = defaultOptions;
|
||||
_numWaiting = 0;
|
||||
_context.statManager().createRateStat("stream.con.lifetimeMessagesSent", "How many messages do we send on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeMessagesReceived", "How many messages do we receive on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeBytesSent", "How many bytes do we send on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeBytesReceived", "How many bytes do we receive on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeDupMessagesSent", "How many duplicate messages do we send on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeDupMessagesReceived", "How many duplicate messages do we receive on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeRTT", "What is the final RTT when a stream closes?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeCongestionSeenAt", "When was the last congestion seen at when a stream closes?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.con.lifetimeSendWindowSize", "What is the final send window size when a stream closes?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
_context.statManager().createRateStat("stream.receiveActive", "How many streams are active when a new one is received (period being not yet dropped)", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
|
||||
}
|
||||
|
||||
Connection getConnectionByInboundId(long id) {
|
||||
synchronized (_connectionLock) {
|
||||
return (Connection)_connectionByInboundId.get(new Long(id));
|
||||
}
|
||||
}
|
||||
/**
|
||||
* not guaranteed to be unique, but in case we receive more than one packet
|
||||
* on an inbound connection that we havent ack'ed yet...
|
||||
*/
|
||||
Connection getConnectionByOutboundId(long id) {
|
||||
synchronized (_connectionLock) {
|
||||
for (Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext(); ) {
|
||||
Connection con = (Connection)iter.next();
|
||||
if (DataHelper.eq(con.getSendStreamId(), id))
|
||||
return con;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public void setAllowIncomingConnections(boolean allow) {
|
||||
_connectionHandler.setActive(allow);
|
||||
}
|
||||
/** should we acceot connections, or just reject everyone? */
|
||||
public boolean getAllowIncomingConnections() {
|
||||
return _connectionHandler.getActive();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new connection based on the SYN packet we received.
|
||||
*
|
||||
* @return created Connection with the packet's data already delivered to
|
||||
* it, or null if the syn's streamId was already taken
|
||||
*/
|
||||
public Connection receiveConnection(Packet synPacket) {
|
||||
Connection con = new Connection(_context, this, _schedulerChooser, _outboundQueue, _conPacketHandler, new ConnectionOptions(_defaultOptions));
|
||||
long receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID-1)+1;
|
||||
boolean reject = false;
|
||||
int active = 0;
|
||||
int total = 0;
|
||||
synchronized (_connectionLock) {
|
||||
total = _connectionByInboundId.size();
|
||||
for (Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext(); ) {
|
||||
if ( ((Connection)iter.next()).getIsConnected() )
|
||||
active++;
|
||||
}
|
||||
if (locked_tooManyStreams()) {
|
||||
reject = true;
|
||||
} else {
|
||||
while (true) {
|
||||
Connection oldCon = (Connection)_connectionByInboundId.put(new Long(receiveId), con);
|
||||
if (oldCon == null) {
|
||||
break;
|
||||
} else {
|
||||
_connectionByInboundId.put(new Long(receiveId), oldCon);
|
||||
// receiveId already taken, try another
|
||||
receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID-1)+1;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_context.statManager().addRateData("stream.receiveActive", active, total);
|
||||
|
||||
if (reject) {
|
||||
if (_log.shouldLog(Log.WARN))
|
||||
_log.warn("Refusing connection since we have exceeded our max of "
|
||||
+ _maxConcurrentStreams + " connections");
|
||||
PacketLocal reply = new PacketLocal(_context, synPacket.getOptionalFrom());
|
||||
reply.setFlag(Packet.FLAG_RESET);
|
||||
reply.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
|
||||
reply.setAckThrough(synPacket.getSequenceNum());
|
||||
reply.setSendStreamId(synPacket.getReceiveStreamId());
|
||||
reply.setReceiveStreamId(0);
|
||||
reply.setOptionalFrom(_session.getMyDestination());
|
||||
// this just sends the packet - no retries or whatnot
|
||||
_outboundQueue.enqueue(reply);
|
||||
return null;
|
||||
}
|
||||
|
||||
con.setReceiveStreamId(receiveId);
|
||||
try {
|
||||
con.getPacketHandler().receivePacket(synPacket, con);
|
||||
} catch (I2PException ie) {
|
||||
synchronized (_connectionLock) {
|
||||
_connectionByInboundId.remove(new Long(receiveId));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
_context.statManager().addRateData("stream.connectionReceived", 1, 0);
|
||||
return con;
|
||||
}
|
||||
|
||||
private static final long DEFAULT_STREAM_DELAY_MAX = 10*1000;
|
||||
|
||||
/**
|
||||
* Build a new connection to the given peer. This blocks if there is no
|
||||
* connection delay, otherwise it returns immediately.
|
||||
*
|
||||
* @return new connection, or null if we have exceeded our limit
|
||||
*/
|
||||
public Connection connect(Destination peer, ConnectionOptions opts) {
|
||||
Connection con = null;
|
||||
long receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID-1)+1;
|
||||
long expiration = _context.clock().now() + opts.getConnectTimeout();
|
||||
if (opts.getConnectTimeout() <= 0)
|
||||
expiration = _context.clock().now() + DEFAULT_STREAM_DELAY_MAX;
|
||||
_numWaiting++;
|
||||
while (true) {
|
||||
long remaining = expiration - _context.clock().now();
|
||||
if (remaining <= 0) {
|
||||
if (_log.shouldLog(Log.WARN))
|
||||
_log.warn("Refusing to connect since we have exceeded our max of "
|
||||
+ _maxConcurrentStreams + " connections");
|
||||
_numWaiting--;
|
||||
return null;
|
||||
}
|
||||
boolean reject = false;
|
||||
synchronized (_connectionLock) {
|
||||
if (locked_tooManyStreams()) {
|
||||
// allow a full buffer of pending/waiting streams
|
||||
if (_numWaiting > _maxConcurrentStreams) {
|
||||
if (_log.shouldLog(Log.WARN))
|
||||
_log.warn("Refusing connection since we have exceeded our max of "
|
||||
+ _maxConcurrentStreams + " and there are " + _numWaiting
|
||||
+ " waiting already");
|
||||
_numWaiting--;
|
||||
return null;
|
||||
}
|
||||
|
||||
// no remaining streams, lets wait a bit
|
||||
try { _connectionLock.wait(remaining); } catch (InterruptedException ie) {}
|
||||
} else {
|
||||
con = new Connection(_context, this, _schedulerChooser, _outboundQueue, _conPacketHandler, opts);
|
||||
con.setRemotePeer(peer);
|
||||
|
||||
while (_connectionByInboundId.containsKey(new Long(receiveId))) {
|
||||
receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID-1)+1;
|
||||
}
|
||||
_connectionByInboundId.put(new Long(receiveId), con);
|
||||
break; // stop looping as a psuedo-wait
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ok we're in...
|
||||
con.setReceiveStreamId(receiveId);
|
||||
con.eventOccurred();
|
||||
|
||||
_log.debug("Connect() conDelay = " + opts.getConnectDelay());
|
||||
if (opts.getConnectDelay() <= 0) {
|
||||
con.waitForConnect();
|
||||
}
|
||||
if (_numWaiting > 0)
|
||||
_numWaiting--;
|
||||
|
||||
_context.statManager().addRateData("stream.connectionCreated", 1, 0);
|
||||
return con;
|
||||
}
|
||||
private I2PAppContext _context;
|
||||
private Log _log;
|
||||
private I2PSession _session;
|
||||
private MessageHandler _messageHandler;
|
||||
private PacketHandler _packetHandler;
|
||||
private ConnectionHandler _connectionHandler;
|
||||
private PacketQueue _outboundQueue;
|
||||
private SchedulerChooser _schedulerChooser;
|
||||
private ConnectionPacketHandler _conPacketHandler;
|
||||
/** Inbound stream ID (Long) to Connection map */
|
||||
private Map _connectionByInboundId;
|
||||
/** Ping ID (Long) to PingRequest */
|
||||
private Map _pendingPings;
|
||||
private boolean _allowIncoming;
|
||||
private int _maxConcurrentStreams;
|
||||
private ConnectionOptions _defaultOptions;
|
||||
private volatile int _numWaiting;
|
||||
private Object _connectionLock;
|
||||
private long SoTimeout;
|
||||
|
||||
private boolean locked_tooManyStreams() {
|
||||
if (_maxConcurrentStreams <= 0) return false;
|
||||
if (_connectionByInboundId.size() < _maxConcurrentStreams) return false;
|
||||
int active = 0;
|
||||
for (Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext(); ) {
|
||||
Connection con = (Connection)iter.next();
|
||||
if (con.getIsConnected())
|
||||
active++;
|
||||
}
|
||||
|
||||
if ( (_connectionByInboundId.size() > 100) && (_log.shouldLog(Log.INFO)) )
|
||||
_log.info("More than 100 connections! " + active
|
||||
+ " total: " + _connectionByInboundId.size());
|
||||
public ConnectionManager(I2PAppContext context, I2PSession session, int maxConcurrent, ConnectionOptions defaultOptions) {
|
||||
_context = context;
|
||||
_log = context.logManager().getLog(ConnectionManager.class);
|
||||
_connectionByInboundId = new HashMap(32);
|
||||
_pendingPings = new HashMap(4);
|
||||
_connectionLock = new Object();
|
||||
_messageHandler = new MessageHandler(context, this);
|
||||
_packetHandler = new PacketHandler(context, this);
|
||||
_connectionHandler = new ConnectionHandler(context, this);
|
||||
_schedulerChooser = new SchedulerChooser(context);
|
||||
_conPacketHandler = new ConnectionPacketHandler(context);
|
||||
_session = session;
|
||||
session.setSessionListener(_messageHandler);
|
||||
_outboundQueue = new PacketQueue(context, session, this);
|
||||
_allowIncoming = false;
|
||||
_maxConcurrentStreams = maxConcurrent;
|
||||
_defaultOptions = defaultOptions;
|
||||
_numWaiting = 0;
|
||||
/** Socket timeout for accept() */
|
||||
SoTimeout = -1;
|
||||
|
||||
return (active >= _maxConcurrentStreams);
|
||||
}
|
||||
|
||||
public MessageHandler getMessageHandler() { return _messageHandler; }
|
||||
public PacketHandler getPacketHandler() { return _packetHandler; }
|
||||
public ConnectionHandler getConnectionHandler() { return _connectionHandler; }
|
||||
public I2PSession getSession() { return _session; }
|
||||
public PacketQueue getPacketQueue() { return _outboundQueue; }
|
||||
|
||||
/**
|
||||
* Something b0rked hard, so kill all of our connections without mercy.
|
||||
* Don't bother sending close packets.
|
||||
*
|
||||
*/
|
||||
public void disconnectAllHard() {
|
||||
synchronized (_connectionLock) {
|
||||
for (Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext(); ) {
|
||||
Connection con = (Connection)iter.next();
|
||||
con.disconnect(false, false);
|
||||
}
|
||||
_connectionByInboundId.clear();
|
||||
_connectionLock.notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop the (already closed) connection on the floor.
|
||||
*
|
||||
*/
|
||||
public void removeConnection(Connection con) {
|
||||
boolean removed = false;
|
||||
synchronized (_connectionLock) {
|
||||
Object o = _connectionByInboundId.remove(new Long(con.getReceiveStreamId()));
|
||||
removed = (o == con);
|
||||
if (_log.shouldLog(Log.DEBUG))
|
||||
_log.debug("Connection removed? " + removed + " remaining: "
|
||||
+ _connectionByInboundId.size() + ": " + con);
|
||||
if (!removed && _log.shouldLog(Log.DEBUG))
|
||||
_log.debug("Failed to remove " + con +"\n" + _connectionByInboundId.values());
|
||||
_connectionLock.notifyAll();
|
||||
}
|
||||
if (removed) {
|
||||
_context.statManager().addRateData("stream.con.lifetimeMessagesSent", con.getLastSendId(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeMessagesReceived", con.getHighestAckedThrough(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeBytesSent", con.getLifetimeBytesSent(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeBytesReceived", con.getLifetimeBytesReceived(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeDupMessagesSent", con.getLifetimeDupMessagesSent(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeDupMessagesReceived", con.getLifetimeDupMessagesReceived(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeRTT", con.getOptions().getRTT(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeCongestionSeenAt", con.getLastCongestionSeenAt(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeSendWindowSize", con.getOptions().getWindowSize(), con.getLifetime());
|
||||
}
|
||||
}
|
||||
|
||||
/** return a set of Connection objects */
|
||||
public Set listConnections() {
|
||||
synchronized (_connectionLock) {
|
||||
return new HashSet(_connectionByInboundId.values());
|
||||
}
|
||||
}
|
||||
|
||||
public boolean ping(Destination peer, long timeoutMs) {
|
||||
return ping(peer, timeoutMs, true);
|
||||
}
|
||||
public boolean ping(Destination peer, long timeoutMs, boolean blocking) {
|
||||
return ping(peer, timeoutMs, blocking, null, null, null);
|
||||
}
|
||||
public boolean ping(Destination peer, long timeoutMs, boolean blocking, SessionKey keyToUse, Set tagsToSend, PingNotifier notifier) {
|
||||
Long id = new Long(_context.random().nextLong(Packet.MAX_STREAM_ID-1)+1);
|
||||
PacketLocal packet = new PacketLocal(_context, peer);
|
||||
packet.setSendStreamId(id.longValue());
|
||||
packet.setFlag(Packet.FLAG_ECHO);
|
||||
packet.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
|
||||
packet.setOptionalFrom(_session.getMyDestination());
|
||||
if ( (keyToUse != null) && (tagsToSend != null) ) {
|
||||
packet.setKeyUsed(keyToUse);
|
||||
packet.setTagsSent(tagsToSend);
|
||||
}
|
||||
|
||||
PingRequest req = new PingRequest(peer, packet, notifier);
|
||||
|
||||
synchronized (_pendingPings) {
|
||||
_pendingPings.put(id, req);
|
||||
}
|
||||
|
||||
_outboundQueue.enqueue(packet);
|
||||
packet.releasePayload();
|
||||
|
||||
if (blocking) {
|
||||
synchronized (req) {
|
||||
if (!req.pongReceived())
|
||||
try { req.wait(timeoutMs); } catch (InterruptedException ie) {}
|
||||
}
|
||||
|
||||
synchronized (_pendingPings) {
|
||||
_pendingPings.remove(id);
|
||||
}
|
||||
} else {
|
||||
SimpleTimer.getInstance().addEvent(new PingFailed(id, notifier), timeoutMs);
|
||||
}
|
||||
|
||||
boolean ok = req.pongReceived();
|
||||
return ok;
|
||||
}
|
||||
_context.statManager().createRateStat("stream.con.lifetimeMessagesSent", "How many messages do we send on a stream?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeMessagesReceived", "How many messages do we receive on a stream?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeBytesSent", "How many bytes do we send on a stream?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeBytesReceived", "How many bytes do we receive on a stream?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeDupMessagesSent", "How many duplicate messages do we send on a stream?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeDupMessagesReceived", "How many duplicate messages do we receive on a stream?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeRTT", "What is the final RTT when a stream closes?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeCongestionSeenAt", "When was the last congestion seen at when a stream closes?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.con.lifetimeSendWindowSize", "What is the final send window size when a stream closes?", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
_context.statManager().createRateStat("stream.receiveActive", "How many streams are active when a new one is received (period being not yet dropped)", "Stream", new long[] {60 * 60 * 1000, 24 * 60 * 60 * 1000});
|
||||
}
|
||||
|
||||
interface PingNotifier {
|
||||
public void pingComplete(boolean ok);
|
||||
}
|
||||
|
||||
private class PingFailed implements SimpleTimer.TimedEvent {
|
||||
private Long _id;
|
||||
private PingNotifier _notifier;
|
||||
public PingFailed(Long id, PingNotifier notifier) {
|
||||
_id = id;
|
||||
_notifier = notifier;
|
||||
}
|
||||
|
||||
public void timeReached() {
|
||||
boolean removed = false;
|
||||
synchronized (_pendingPings) {
|
||||
Object o = _pendingPings.remove(_id);
|
||||
if (o != null)
|
||||
removed = true;
|
||||
}
|
||||
if (removed) {
|
||||
if (_notifier != null)
|
||||
_notifier.pingComplete(false);
|
||||
if (_log.shouldLog(Log.INFO))
|
||||
_log.info("Ping failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class PingRequest {
|
||||
private boolean _ponged;
|
||||
private Destination _peer;
|
||||
private PacketLocal _packet;
|
||||
private PingNotifier _notifier;
|
||||
public PingRequest(Destination peer, PacketLocal packet, PingNotifier notifier) {
|
||||
_ponged = false;
|
||||
_peer = peer;
|
||||
_packet = packet;
|
||||
_notifier = notifier;
|
||||
}
|
||||
public void pong() {
|
||||
_log.debug("Ping successful");
|
||||
_context.sessionKeyManager().tagsDelivered(_peer.getPublicKey(), _packet.getKeyUsed(), _packet.getTagsSent());
|
||||
synchronized (ConnectionManager.PingRequest.this) {
|
||||
_ponged = true;
|
||||
ConnectionManager.PingRequest.this.notifyAll();
|
||||
}
|
||||
if (_notifier != null)
|
||||
_notifier.pingComplete(true);
|
||||
}
|
||||
public boolean pongReceived() { return _ponged; }
|
||||
}
|
||||
|
||||
void receivePong(long pingId) {
|
||||
PingRequest req = null;
|
||||
synchronized (_pendingPings) {
|
||||
req = (PingRequest)_pendingPings.remove(new Long(pingId));
|
||||
}
|
||||
if (req != null)
|
||||
req.pong();
|
||||
}
|
||||
Connection getConnectionByInboundId(long id) {
|
||||
synchronized(_connectionLock) {
|
||||
return (Connection)_connectionByInboundId.get(new Long(id));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* not guaranteed to be unique, but in case we receive more than one packet
|
||||
* on an inbound connection that we havent ack'ed yet...
|
||||
*/
|
||||
Connection getConnectionByOutboundId(long id) {
|
||||
synchronized(_connectionLock) {
|
||||
for(Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext();) {
|
||||
Connection con = (Connection)iter.next();
|
||||
if(DataHelper.eq(con.getSendStreamId(), id)) {
|
||||
return con;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the socket accept() timeout.
|
||||
* @param x
|
||||
*/
|
||||
public void MsetSoTimeout(long x) {
|
||||
SoTimeout = x;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the socket accept() timeout.
|
||||
* @return
|
||||
*/
|
||||
public long MgetSoTimeout() {
|
||||
return SoTimeout;
|
||||
}
|
||||
|
||||
public void setAllowIncomingConnections(boolean allow) {
|
||||
_connectionHandler.setActive(allow);
|
||||
}
|
||||
|
||||
/** should we acceot connections, or just reject everyone? */
|
||||
public boolean getAllowIncomingConnections() {
|
||||
return _connectionHandler.getActive();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new connection based on the SYN packet we received.
|
||||
*
|
||||
* @return created Connection with the packet's data already delivered to
|
||||
* it, or null if the syn's streamId was already taken
|
||||
*/
|
||||
public Connection receiveConnection(Packet synPacket) {
|
||||
Connection con = new Connection(_context, this, _schedulerChooser, _outboundQueue, _conPacketHandler, new ConnectionOptions(_defaultOptions));
|
||||
long receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID - 1) + 1;
|
||||
boolean reject = false;
|
||||
int active = 0;
|
||||
int total = 0;
|
||||
synchronized(_connectionLock) {
|
||||
total = _connectionByInboundId.size();
|
||||
for(Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext();) {
|
||||
if(((Connection)iter.next()).getIsConnected()) {
|
||||
active++;
|
||||
}
|
||||
}
|
||||
if(locked_tooManyStreams()) {
|
||||
reject = true;
|
||||
} else {
|
||||
while(true) {
|
||||
Connection oldCon = (Connection)_connectionByInboundId.put(new Long(receiveId), con);
|
||||
if(oldCon == null) {
|
||||
break;
|
||||
} else {
|
||||
_connectionByInboundId.put(new Long(receiveId), oldCon);
|
||||
// receiveId already taken, try another
|
||||
receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID - 1) + 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_context.statManager().addRateData("stream.receiveActive", active, total);
|
||||
|
||||
if(reject) {
|
||||
if(_log.shouldLog(Log.WARN)) {
|
||||
_log.warn("Refusing connection since we have exceeded our max of " + _maxConcurrentStreams + " connections");
|
||||
}
|
||||
PacketLocal reply = new PacketLocal(_context, synPacket.getOptionalFrom());
|
||||
reply.setFlag(Packet.FLAG_RESET);
|
||||
reply.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
|
||||
reply.setAckThrough(synPacket.getSequenceNum());
|
||||
reply.setSendStreamId(synPacket.getReceiveStreamId());
|
||||
reply.setReceiveStreamId(0);
|
||||
reply.setOptionalFrom(_session.getMyDestination());
|
||||
// this just sends the packet - no retries or whatnot
|
||||
_outboundQueue.enqueue(reply);
|
||||
return null;
|
||||
}
|
||||
|
||||
con.setReceiveStreamId(receiveId);
|
||||
try {
|
||||
con.getPacketHandler().receivePacket(synPacket, con);
|
||||
} catch(I2PException ie) {
|
||||
synchronized(_connectionLock) {
|
||||
_connectionByInboundId.remove(new Long(receiveId));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
_context.statManager().addRateData("stream.connectionReceived", 1, 0);
|
||||
return con;
|
||||
}
|
||||
private static final long DEFAULT_STREAM_DELAY_MAX = 10 * 1000;
|
||||
|
||||
/**
|
||||
* Build a new connection to the given peer. This blocks if there is no
|
||||
* connection delay, otherwise it returns immediately.
|
||||
*
|
||||
* @return new connection, or null if we have exceeded our limit
|
||||
*/
|
||||
public Connection connect(Destination peer, ConnectionOptions opts) {
|
||||
Connection con = null;
|
||||
long receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID - 1) + 1;
|
||||
long expiration = _context.clock().now() + opts.getConnectTimeout();
|
||||
if(opts.getConnectTimeout() <= 0) {
|
||||
expiration = _context.clock().now() + DEFAULT_STREAM_DELAY_MAX;
|
||||
}
|
||||
_numWaiting++;
|
||||
while(true) {
|
||||
long remaining = expiration - _context.clock().now();
|
||||
if(remaining <= 0) {
|
||||
if(_log.shouldLog(Log.WARN)) {
|
||||
_log.warn("Refusing to connect since we have exceeded our max of " + _maxConcurrentStreams + " connections");
|
||||
}
|
||||
_numWaiting--;
|
||||
return null;
|
||||
}
|
||||
boolean reject = false;
|
||||
synchronized(_connectionLock) {
|
||||
if(locked_tooManyStreams()) {
|
||||
// allow a full buffer of pending/waiting streams
|
||||
if(_numWaiting > _maxConcurrentStreams) {
|
||||
if(_log.shouldLog(Log.WARN)) {
|
||||
_log.warn("Refusing connection since we have exceeded our max of " + _maxConcurrentStreams + " and there are " + _numWaiting + " waiting already");
|
||||
}
|
||||
_numWaiting--;
|
||||
return null;
|
||||
}
|
||||
|
||||
// no remaining streams, lets wait a bit
|
||||
try {
|
||||
_connectionLock.wait(remaining);
|
||||
} catch(InterruptedException ie) {
|
||||
}
|
||||
} else {
|
||||
con = new Connection(_context, this, _schedulerChooser, _outboundQueue, _conPacketHandler, opts);
|
||||
con.setRemotePeer(peer);
|
||||
|
||||
while(_connectionByInboundId.containsKey(new Long(receiveId))) {
|
||||
receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID - 1) + 1;
|
||||
}
|
||||
_connectionByInboundId.put(new Long(receiveId), con);
|
||||
break; // stop looping as a psuedo-wait
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ok we're in...
|
||||
con.setReceiveStreamId(receiveId);
|
||||
con.eventOccurred();
|
||||
|
||||
_log.debug("Connect() conDelay = " + opts.getConnectDelay());
|
||||
if(opts.getConnectDelay() <= 0) {
|
||||
con.waitForConnect();
|
||||
}
|
||||
if(_numWaiting > 0) {
|
||||
_numWaiting--;
|
||||
}
|
||||
_context.statManager().addRateData("stream.connectionCreated", 1, 0);
|
||||
return con;
|
||||
}
|
||||
|
||||
private boolean locked_tooManyStreams() {
|
||||
if(_maxConcurrentStreams <= 0) {
|
||||
return false;
|
||||
}
|
||||
if(_connectionByInboundId.size() < _maxConcurrentStreams) {
|
||||
return false;
|
||||
}
|
||||
int active = 0;
|
||||
for(Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext();) {
|
||||
Connection con = (Connection)iter.next();
|
||||
if(con.getIsConnected()) {
|
||||
active++;
|
||||
}
|
||||
}
|
||||
|
||||
if((_connectionByInboundId.size() > 100) && (_log.shouldLog(Log.INFO))) {
|
||||
_log.info("More than 100 connections! " + active + " total: " + _connectionByInboundId.size());
|
||||
}
|
||||
return (active >= _maxConcurrentStreams);
|
||||
}
|
||||
|
||||
public MessageHandler getMessageHandler() {
|
||||
return _messageHandler;
|
||||
}
|
||||
|
||||
public PacketHandler getPacketHandler() {
|
||||
return _packetHandler;
|
||||
}
|
||||
|
||||
public ConnectionHandler getConnectionHandler() {
|
||||
return _connectionHandler;
|
||||
}
|
||||
|
||||
public I2PSession getSession() {
|
||||
return _session;
|
||||
}
|
||||
|
||||
public PacketQueue getPacketQueue() {
|
||||
return _outboundQueue;
|
||||
}
|
||||
|
||||
/**
|
||||
* Something b0rked hard, so kill all of our connections without mercy.
|
||||
* Don't bother sending close packets.
|
||||
*
|
||||
*/
|
||||
public void disconnectAllHard() {
|
||||
synchronized(_connectionLock) {
|
||||
for(Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext();) {
|
||||
Connection con = (Connection)iter.next();
|
||||
con.disconnect(false, false);
|
||||
}
|
||||
_connectionByInboundId.clear();
|
||||
_connectionLock.notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop the (already closed) connection on the floor.
|
||||
*
|
||||
*/
|
||||
public void removeConnection(Connection con) {
|
||||
boolean removed = false;
|
||||
synchronized(_connectionLock) {
|
||||
Object o = _connectionByInboundId.remove(new Long(con.getReceiveStreamId()));
|
||||
removed = (o == con);
|
||||
if(_log.shouldLog(Log.DEBUG)) {
|
||||
_log.debug("Connection removed? " + removed + " remaining: " + _connectionByInboundId.size() + ": " + con);
|
||||
}
|
||||
if(!removed && _log.shouldLog(Log.DEBUG)) {
|
||||
_log.debug("Failed to remove " + con + "\n" + _connectionByInboundId.values());
|
||||
}
|
||||
_connectionLock.notifyAll();
|
||||
}
|
||||
if(removed) {
|
||||
_context.statManager().addRateData("stream.con.lifetimeMessagesSent", con.getLastSendId(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeMessagesReceived", con.getHighestAckedThrough(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeBytesSent", con.getLifetimeBytesSent(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeBytesReceived", con.getLifetimeBytesReceived(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeDupMessagesSent", con.getLifetimeDupMessagesSent(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeDupMessagesReceived", con.getLifetimeDupMessagesReceived(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeRTT", con.getOptions().getRTT(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeCongestionSeenAt", con.getLastCongestionSeenAt(), con.getLifetime());
|
||||
_context.statManager().addRateData("stream.con.lifetimeSendWindowSize", con.getOptions().getWindowSize(), con.getLifetime());
|
||||
}
|
||||
}
|
||||
|
||||
/** return a set of Connection objects */
|
||||
public Set listConnections() {
|
||||
synchronized(_connectionLock) {
|
||||
return new HashSet(_connectionByInboundId.values());
|
||||
}
|
||||
}
|
||||
|
||||
public boolean ping(Destination peer, long timeoutMs) {
|
||||
return ping(peer, timeoutMs, true);
|
||||
}
|
||||
|
||||
public boolean ping(Destination peer, long timeoutMs, boolean blocking) {
|
||||
return ping(peer, timeoutMs, blocking, null, null, null);
|
||||
}
|
||||
|
||||
public boolean ping(Destination peer, long timeoutMs, boolean blocking, SessionKey keyToUse, Set tagsToSend, PingNotifier notifier) {
|
||||
Long id = new Long(_context.random().nextLong(Packet.MAX_STREAM_ID - 1) + 1);
|
||||
PacketLocal packet = new PacketLocal(_context, peer);
|
||||
packet.setSendStreamId(id.longValue());
|
||||
packet.setFlag(Packet.FLAG_ECHO);
|
||||
packet.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
|
||||
packet.setOptionalFrom(_session.getMyDestination());
|
||||
if((keyToUse != null) && (tagsToSend != null)) {
|
||||
packet.setKeyUsed(keyToUse);
|
||||
packet.setTagsSent(tagsToSend);
|
||||
}
|
||||
|
||||
PingRequest req = new PingRequest(peer, packet, notifier);
|
||||
|
||||
synchronized(_pendingPings) {
|
||||
_pendingPings.put(id, req);
|
||||
}
|
||||
|
||||
_outboundQueue.enqueue(packet);
|
||||
packet.releasePayload();
|
||||
|
||||
if(blocking) {
|
||||
synchronized(req) {
|
||||
if(!req.pongReceived()) {
|
||||
try {
|
||||
req.wait(timeoutMs);
|
||||
} catch(InterruptedException ie) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
synchronized(_pendingPings) {
|
||||
_pendingPings.remove(id);
|
||||
}
|
||||
} else {
|
||||
SimpleTimer.getInstance().addEvent(new PingFailed(id, notifier), timeoutMs);
|
||||
}
|
||||
|
||||
boolean ok = req.pongReceived();
|
||||
return ok;
|
||||
}
|
||||
|
||||
interface PingNotifier {
|
||||
|
||||
public void pingComplete(boolean ok);
|
||||
}
|
||||
|
||||
private class PingFailed implements SimpleTimer.TimedEvent {
|
||||
|
||||
private Long _id;
|
||||
private PingNotifier _notifier;
|
||||
|
||||
public PingFailed(Long id, PingNotifier notifier) {
|
||||
_id = id;
|
||||
_notifier = notifier;
|
||||
}
|
||||
|
||||
public void timeReached() {
|
||||
boolean removed = false;
|
||||
synchronized(_pendingPings) {
|
||||
Object o = _pendingPings.remove(_id);
|
||||
if(o != null) {
|
||||
removed = true;
|
||||
}
|
||||
}
|
||||
if(removed) {
|
||||
if(_notifier != null) {
|
||||
_notifier.pingComplete(false);
|
||||
}
|
||||
if(_log.shouldLog(Log.INFO)) {
|
||||
_log.info("Ping failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class PingRequest {
|
||||
|
||||
private boolean _ponged;
|
||||
private Destination _peer;
|
||||
private PacketLocal _packet;
|
||||
private PingNotifier _notifier;
|
||||
|
||||
public PingRequest(Destination peer, PacketLocal packet, PingNotifier notifier) {
|
||||
_ponged = false;
|
||||
_peer = peer;
|
||||
_packet = packet;
|
||||
_notifier = notifier;
|
||||
}
|
||||
|
||||
public void pong() {
|
||||
_log.debug("Ping successful");
|
||||
_context.sessionKeyManager().tagsDelivered(_peer.getPublicKey(), _packet.getKeyUsed(), _packet.getTagsSent());
|
||||
synchronized(ConnectionManager.PingRequest.this) {
|
||||
_ponged = true;
|
||||
ConnectionManager.PingRequest.this.notifyAll();
|
||||
}
|
||||
if(_notifier != null) {
|
||||
_notifier.pingComplete(true);
|
||||
}
|
||||
}
|
||||
|
||||
public boolean pongReceived() {
|
||||
return _ponged;
|
||||
}
|
||||
}
|
||||
|
||||
void receivePong(long pingId) {
|
||||
PingRequest req = null;
|
||||
synchronized(_pendingPings) {
|
||||
req = (PingRequest)_pendingPings.remove(new Long(pingId));
|
||||
}
|
||||
if(req != null) {
|
||||
req.pong();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
package net.i2p.client.streaming;
|
||||
|
||||
import java.net.SocketTimeoutException;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
import net.i2p.I2PException;
|
||||
|
||||
/**
|
||||
@@ -7,17 +10,46 @@ import net.i2p.I2PException;
|
||||
*
|
||||
*/
|
||||
public class I2PServerSocketFull implements I2PServerSocket {
|
||||
private I2PSocketManagerFull _socketManager;
|
||||
|
||||
public I2PServerSocketFull(I2PSocketManagerFull mgr) {
|
||||
_socketManager = mgr;
|
||||
}
|
||||
|
||||
public I2PSocket accept() throws I2PException {
|
||||
return _socketManager.receiveSocket();
|
||||
}
|
||||
|
||||
public void close() { _socketManager.getConnectionManager().setAllowIncomingConnections(false); }
|
||||
|
||||
public I2PSocketManager getManager() { return _socketManager; }
|
||||
|
||||
private I2PSocketManagerFull _socketManager;
|
||||
|
||||
/**
|
||||
*
|
||||
* @param mgr
|
||||
*/
|
||||
public I2PServerSocketFull(I2PSocketManagerFull mgr) {
|
||||
_socketManager = mgr;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
* @throws net.i2p.I2PException
|
||||
* @throws SocketTimeoutException
|
||||
*/
|
||||
public I2PSocket accept() throws I2PException, SocketTimeoutException {
|
||||
return _socketManager.receiveSocket();
|
||||
}
|
||||
|
||||
public long getSoTimeout() {
|
||||
return _socketManager.getConnectionManager().MgetSoTimeout();
|
||||
}
|
||||
|
||||
public void setSoTimeout(long x) {
|
||||
_socketManager.getConnectionManager().MsetSoTimeout(x);
|
||||
}
|
||||
/**
|
||||
* Close the connection.
|
||||
*/
|
||||
public void close() {
|
||||
_socketManager.getConnectionManager().setAllowIncomingConnections(false);
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @return _socketManager
|
||||
*/
|
||||
public I2PSocketManager getManager() {
|
||||
return _socketManager;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,119 +11,139 @@ import net.i2p.data.Destination;
|
||||
*
|
||||
*/
|
||||
public class I2PSocketFull implements I2PSocket {
|
||||
private Connection _connection;
|
||||
private I2PSocket.SocketErrorListener _listener;
|
||||
private Destination _remotePeer;
|
||||
private Destination _localPeer;
|
||||
|
||||
public I2PSocketFull(Connection con) {
|
||||
_connection = con;
|
||||
if (con != null) {
|
||||
_remotePeer = con.getRemotePeer();
|
||||
_localPeer = con.getSession().getMyDestination();
|
||||
}
|
||||
}
|
||||
|
||||
public void close() throws IOException {
|
||||
Connection c = _connection;
|
||||
if (c == null) return;
|
||||
if (c.getIsConnected()) {
|
||||
OutputStream out = c.getOutputStream();
|
||||
if (out != null) {
|
||||
try {
|
||||
out.close();
|
||||
} catch (IOException ioe) {
|
||||
// ignore any write error, as we want to keep on and kill the
|
||||
// con (thanks Complication!)
|
||||
}
|
||||
}
|
||||
c.disconnect(true);
|
||||
} else {
|
||||
//throw new IOException("Not connected");
|
||||
}
|
||||
destroy();
|
||||
}
|
||||
|
||||
Connection getConnection() { return _connection; }
|
||||
|
||||
public InputStream getInputStream() {
|
||||
Connection c = _connection;
|
||||
if (c != null)
|
||||
return c.getInputStream();
|
||||
else
|
||||
return null;
|
||||
}
|
||||
|
||||
public I2PSocketOptions getOptions() {
|
||||
Connection c = _connection;
|
||||
if (c != null)
|
||||
return c.getOptions();
|
||||
else
|
||||
return null;
|
||||
}
|
||||
|
||||
public OutputStream getOutputStream() throws IOException {
|
||||
Connection c = _connection;
|
||||
if (c != null)
|
||||
return c.getOutputStream();
|
||||
else
|
||||
return null;
|
||||
}
|
||||
|
||||
public Destination getPeerDestination() { return _remotePeer; }
|
||||
|
||||
public long getReadTimeout() {
|
||||
I2PSocketOptions opts = getOptions();
|
||||
if (opts != null)
|
||||
return opts.getReadTimeout();
|
||||
else
|
||||
return -1;
|
||||
}
|
||||
|
||||
public Destination getThisDestination() { return _localPeer; }
|
||||
|
||||
public void setOptions(I2PSocketOptions options) {
|
||||
Connection c = _connection;
|
||||
if (c == null) return;
|
||||
|
||||
if (options instanceof ConnectionOptions)
|
||||
c.setOptions((ConnectionOptions)options);
|
||||
else
|
||||
c.setOptions(new ConnectionOptions(options));
|
||||
}
|
||||
|
||||
public void setReadTimeout(long ms) {
|
||||
Connection c = _connection;
|
||||
if (c == null) return;
|
||||
|
||||
c.getInputStream().setReadTimeout((int)ms);
|
||||
c.getOptions().setReadTimeout(ms);
|
||||
}
|
||||
|
||||
public void setSocketErrorListener(I2PSocket.SocketErrorListener lsnr) {
|
||||
_listener = lsnr;
|
||||
}
|
||||
|
||||
public boolean isClosed() {
|
||||
Connection c = _connection;
|
||||
return ((c == null) ||
|
||||
(!c.getIsConnected()) ||
|
||||
(c.getResetReceived()) ||
|
||||
(c.getResetSent()));
|
||||
}
|
||||
|
||||
void destroy() {
|
||||
Connection c = _connection;
|
||||
_connection = null;
|
||||
_listener = null;
|
||||
if (c != null)
|
||||
c.disconnectComplete();
|
||||
}
|
||||
public String toString() {
|
||||
Connection c = _connection;
|
||||
if (c == null)
|
||||
return super.toString();
|
||||
else
|
||||
return c.toString();
|
||||
}
|
||||
|
||||
private Connection _connection;
|
||||
private I2PSocket.SocketErrorListener _listener;
|
||||
private Destination _remotePeer;
|
||||
private Destination _localPeer;
|
||||
|
||||
public I2PSocketFull(Connection con) {
|
||||
_connection = con;
|
||||
if(con != null) {
|
||||
_remotePeer = con.getRemotePeer();
|
||||
_localPeer = con.getSession().getMyDestination();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public void close() throws IOException {
|
||||
Connection c = _connection;
|
||||
if(c == null) {
|
||||
return;
|
||||
}
|
||||
if(c.getIsConnected()) {
|
||||
OutputStream out = c.getOutputStream();
|
||||
if(out != null) {
|
||||
try {
|
||||
out.close();
|
||||
} catch(IOException ioe) {
|
||||
// ignore any write error, as we want to keep on and kill the
|
||||
// con (thanks Complication!)
|
||||
}
|
||||
}
|
||||
c.disconnect(true);
|
||||
} else {
|
||||
//throw new IOException("Not connected");
|
||||
}
|
||||
destroy();
|
||||
}
|
||||
|
||||
Connection getConnection() {
|
||||
return _connection;
|
||||
}
|
||||
|
||||
public InputStream getInputStream() {
|
||||
Connection c = _connection;
|
||||
if(c != null) {
|
||||
return c.getInputStream();
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
public I2PSocketOptions getOptions() {
|
||||
Connection c = _connection;
|
||||
if(c != null) {
|
||||
return c.getOptions();
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
public OutputStream getOutputStream() throws IOException {
|
||||
Connection c = _connection;
|
||||
if(c != null) {
|
||||
return c.getOutputStream();
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
public Destination getPeerDestination() {
|
||||
return _remotePeer;
|
||||
}
|
||||
|
||||
public long getReadTimeout() {
|
||||
I2PSocketOptions opts = getOptions();
|
||||
if(opts != null) {
|
||||
return opts.getReadTimeout();
|
||||
} else {
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
public Destination getThisDestination() {
|
||||
return _localPeer;
|
||||
}
|
||||
|
||||
public void setOptions(I2PSocketOptions options) {
|
||||
Connection c = _connection;
|
||||
if(c == null) {
|
||||
return;
|
||||
}
|
||||
if(options instanceof ConnectionOptions) {
|
||||
c.setOptions((ConnectionOptions)options);
|
||||
} else {
|
||||
c.setOptions(new ConnectionOptions(options));
|
||||
}
|
||||
}
|
||||
|
||||
public void setReadTimeout(long ms) {
|
||||
Connection c = _connection;
|
||||
if(c == null) {
|
||||
return;
|
||||
}
|
||||
c.getInputStream().setReadTimeout((int)ms);
|
||||
c.getOptions().setReadTimeout(ms);
|
||||
}
|
||||
|
||||
public void setSocketErrorListener(I2PSocket.SocketErrorListener lsnr) {
|
||||
_listener = lsnr;
|
||||
}
|
||||
|
||||
public boolean isClosed() {
|
||||
Connection c = _connection;
|
||||
return ((c == null) ||
|
||||
(!c.getIsConnected()) ||
|
||||
(c.getResetReceived()) ||
|
||||
(c.getResetSent()));
|
||||
}
|
||||
|
||||
void destroy() {
|
||||
Connection c = _connection;
|
||||
_connection = null;
|
||||
_listener = null;
|
||||
if(c != null) {
|
||||
c.disconnectComplete();
|
||||
}
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
Connection c = _connection;
|
||||
if(c == null) {
|
||||
return super.toString();
|
||||
} else {
|
||||
return c.toString();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package net.i2p.client.streaming;
|
||||
|
||||
import java.net.NoRouteToHostException;
|
||||
import java.net.SocketTimeoutException;
|
||||
import java.util.HashSet;
|
||||
import java.util.Iterator;
|
||||
import java.util.Properties;
|
||||
@@ -13,7 +14,6 @@ import net.i2p.client.I2PSessionException;
|
||||
import net.i2p.data.Destination;
|
||||
import net.i2p.util.Log;
|
||||
|
||||
|
||||
/**
|
||||
* Centralize the coordination and multiplexing of the local client's streaming.
|
||||
* There should be one I2PSocketManager for each I2PSession, and if an application
|
||||
@@ -23,219 +23,317 @@ import net.i2p.util.Log;
|
||||
*
|
||||
*/
|
||||
public class I2PSocketManagerFull implements I2PSocketManager {
|
||||
private I2PAppContext _context;
|
||||
private Log _log;
|
||||
private I2PSession _session;
|
||||
private I2PServerSocketFull _serverSocket;
|
||||
private ConnectionOptions _defaultOptions;
|
||||
private long _acceptTimeout;
|
||||
private String _name;
|
||||
private int _maxStreams;
|
||||
private static int __managerId = 0;
|
||||
private ConnectionManager _connectionManager;
|
||||
|
||||
/**
|
||||
* How long to wait for the client app to accept() before sending back CLOSE?
|
||||
* This includes the time waiting in the queue. Currently set to 5 seconds.
|
||||
*/
|
||||
private static final long ACCEPT_TIMEOUT_DEFAULT = 5*1000;
|
||||
|
||||
public I2PSocketManagerFull() {
|
||||
_context = null;
|
||||
_session = null;
|
||||
}
|
||||
public I2PSocketManagerFull(I2PAppContext context, I2PSession session, Properties opts, String name) {
|
||||
this();
|
||||
init(context, session, opts, name);
|
||||
}
|
||||
|
||||
/** how many streams will we allow at once? */
|
||||
public static final String PROP_MAX_STREAMS = "i2p.streaming.maxConcurrentStreams";
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public void init(I2PAppContext context, I2PSession session, Properties opts, String name) {
|
||||
_context = context;
|
||||
_session = session;
|
||||
_log = _context.logManager().getLog(I2PSocketManagerFull.class);
|
||||
|
||||
_maxStreams = -1;
|
||||
try {
|
||||
String num = (opts != null ? opts.getProperty(PROP_MAX_STREAMS, "-1") : "-1");
|
||||
_maxStreams = Integer.parseInt(num);
|
||||
} catch (NumberFormatException nfe) {
|
||||
if (_log.shouldLog(Log.WARN))
|
||||
_log.warn("Invalid max # of concurrent streams, defaulting to unlimited", nfe);
|
||||
_maxStreams = -1;
|
||||
}
|
||||
_name = name + " " + (++__managerId);
|
||||
_acceptTimeout = ACCEPT_TIMEOUT_DEFAULT;
|
||||
_defaultOptions = new ConnectionOptions(opts);
|
||||
_connectionManager = new ConnectionManager(_context, _session, _maxStreams, _defaultOptions);
|
||||
_serverSocket = new I2PServerSocketFull(this);
|
||||
|
||||
if (_log.shouldLog(Log.INFO)) {
|
||||
_log.info("Socket manager created. \ndefault options: " + _defaultOptions
|
||||
+ "\noriginal properties: " + opts);
|
||||
}
|
||||
}
|
||||
|
||||
public I2PSocketOptions buildOptions() { return buildOptions(null); }
|
||||
public I2PSocketOptions buildOptions(Properties opts) {
|
||||
ConnectionOptions curOpts = new ConnectionOptions(_defaultOptions);
|
||||
curOpts.setProperties(opts);
|
||||
return curOpts;
|
||||
}
|
||||
|
||||
public I2PSession getSession() {
|
||||
return _session;
|
||||
}
|
||||
|
||||
public ConnectionManager getConnectionManager() {
|
||||
return _connectionManager;
|
||||
}
|
||||
private I2PAppContext _context;
|
||||
private Log _log;
|
||||
private I2PSession _session;
|
||||
private I2PServerSocketFull _serverSocket;
|
||||
private ConnectionOptions _defaultOptions;
|
||||
private long _acceptTimeout;
|
||||
private String _name;
|
||||
private int _maxStreams;
|
||||
private static int __managerId = 0;
|
||||
private ConnectionManager _connectionManager;
|
||||
/**
|
||||
* How long to wait for the client app to accept() before sending back CLOSE?
|
||||
* This includes the time waiting in the queue. Currently set to 5 seconds.
|
||||
*/
|
||||
private static final long ACCEPT_TIMEOUT_DEFAULT = 5 * 1000;
|
||||
|
||||
public I2PSocket receiveSocket() throws I2PException {
|
||||
verifySession();
|
||||
Connection con = _connectionManager.getConnectionHandler().accept(-1);
|
||||
if (_log.shouldLog(Log.DEBUG))
|
||||
_log.debug("receiveSocket() called: " + con);
|
||||
if (con != null) {
|
||||
I2PSocketFull sock = new I2PSocketFull(con);
|
||||
con.setSocket(sock);
|
||||
return sock;
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ping the specified peer, returning true if they replied to the ping within
|
||||
* the timeout specified, false otherwise. This call blocks.
|
||||
*
|
||||
*/
|
||||
public boolean ping(Destination peer, long timeoutMs) {
|
||||
return _connectionManager.ping(peer, timeoutMs);
|
||||
}
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public I2PSocketManagerFull() {
|
||||
_context = null;
|
||||
_session = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* How long should we wait for the client to .accept() a socket before
|
||||
* sending back a NACK/Close?
|
||||
*
|
||||
* @param ms milliseconds to wait, maximum
|
||||
*/
|
||||
public void setAcceptTimeout(long ms) { _acceptTimeout = ms; }
|
||||
public long getAcceptTimeout() { return _acceptTimeout; }
|
||||
/**
|
||||
*
|
||||
* @param context
|
||||
* @param session
|
||||
* @param opts
|
||||
* @param name
|
||||
*/
|
||||
public I2PSocketManagerFull(I2PAppContext context, I2PSession session, Properties opts, String name) {
|
||||
this();
|
||||
init(context, session, opts, name);
|
||||
}
|
||||
/** how many streams will we allow at once? */
|
||||
public static final String PROP_MAX_STREAMS = "i2p.streaming.maxConcurrentStreams";
|
||||
|
||||
public void setDefaultOptions(I2PSocketOptions options) {
|
||||
_defaultOptions = new ConnectionOptions((ConnectionOptions) options);
|
||||
}
|
||||
/**
|
||||
*
|
||||
*
|
||||
* @param context
|
||||
* @param session
|
||||
* @param opts
|
||||
* @param name
|
||||
*/
|
||||
public void init(I2PAppContext context, I2PSession session, Properties opts, String name) {
|
||||
_context = context;
|
||||
_session = session;
|
||||
_log = _context.logManager().getLog(I2PSocketManagerFull.class);
|
||||
|
||||
public I2PSocketOptions getDefaultOptions() {
|
||||
return _defaultOptions;
|
||||
}
|
||||
_maxStreams = -1;
|
||||
try {
|
||||
String num = (opts != null ? opts.getProperty(PROP_MAX_STREAMS, "-1") : "-1");
|
||||
_maxStreams = Integer.parseInt(num);
|
||||
} catch(NumberFormatException nfe) {
|
||||
if(_log.shouldLog(Log.WARN)) {
|
||||
_log.warn("Invalid max # of concurrent streams, defaulting to unlimited", nfe);
|
||||
}
|
||||
_maxStreams = -1;
|
||||
}
|
||||
_name = name + " " + (++__managerId);
|
||||
_acceptTimeout = ACCEPT_TIMEOUT_DEFAULT;
|
||||
_defaultOptions = new ConnectionOptions(opts);
|
||||
_connectionManager = new ConnectionManager(_context, _session, _maxStreams, _defaultOptions);
|
||||
_serverSocket = new I2PServerSocketFull(this);
|
||||
|
||||
public I2PServerSocket getServerSocket() {
|
||||
_connectionManager.setAllowIncomingConnections(true);
|
||||
return _serverSocket;
|
||||
}
|
||||
if(_log.shouldLog(Log.INFO)) {
|
||||
_log.info("Socket manager created. \ndefault options: " + _defaultOptions + "\noriginal properties: " + opts);
|
||||
}
|
||||
}
|
||||
|
||||
private void verifySession() throws I2PException {
|
||||
if (!_connectionManager.getSession().isClosed())
|
||||
return;
|
||||
_connectionManager.getSession().connect();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new connected socket (block until the socket is created)
|
||||
*
|
||||
* @param peer Destination to connect to
|
||||
* @param options I2P socket options to be used for connecting
|
||||
*
|
||||
* @throws NoRouteToHostException if the peer is not found or not reachable
|
||||
* @throws I2PException if there is some other I2P-related problem
|
||||
*/
|
||||
public I2PSocket connect(Destination peer, I2PSocketOptions options)
|
||||
throws I2PException, NoRouteToHostException {
|
||||
verifySession();
|
||||
if (options == null)
|
||||
options = _defaultOptions;
|
||||
ConnectionOptions opts = null;
|
||||
if (options instanceof ConnectionOptions)
|
||||
opts = new ConnectionOptions((ConnectionOptions)options);
|
||||
else
|
||||
opts = new ConnectionOptions(options);
|
||||
|
||||
if (_log.shouldLog(Log.INFO))
|
||||
_log.info("Connecting to " + peer.calculateHash().toBase64().substring(0,6)
|
||||
+ " with options: " + opts);
|
||||
Connection con = _connectionManager.connect(peer, opts);
|
||||
if (con == null)
|
||||
throw new TooManyStreamsException("Too many streams (max " + _maxStreams + ")");
|
||||
I2PSocketFull socket = new I2PSocketFull(con);
|
||||
con.setSocket(socket);
|
||||
if (con.getConnectionError() != null) {
|
||||
con.disconnect(false);
|
||||
throw new NoRouteToHostException(con.getConnectionError());
|
||||
}
|
||||
return socket;
|
||||
}
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public I2PSocketOptions buildOptions() {
|
||||
return buildOptions(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new connected socket (block until the socket is created)
|
||||
*
|
||||
* @param peer Destination to connect to
|
||||
*
|
||||
* @throws NoRouteToHostException if the peer is not found or not reachable
|
||||
* @throws I2PException if there is some other I2P-related problem
|
||||
*/
|
||||
public I2PSocket connect(Destination peer) throws I2PException, NoRouteToHostException {
|
||||
return connect(peer, _defaultOptions);
|
||||
}
|
||||
/**
|
||||
*
|
||||
* @param opts
|
||||
* @return
|
||||
*/
|
||||
public I2PSocketOptions buildOptions(Properties opts) {
|
||||
ConnectionOptions curOpts = new ConnectionOptions(_defaultOptions);
|
||||
curOpts.setProperties(opts);
|
||||
return curOpts;
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroy the socket manager, freeing all the associated resources. This
|
||||
* method will block untill all the managed sockets are closed.
|
||||
*
|
||||
*/
|
||||
public void destroySocketManager() {
|
||||
_connectionManager.disconnectAllHard();
|
||||
_connectionManager.setAllowIncomingConnections(false);
|
||||
// should we destroy the _session too?
|
||||
// yes, since the old lib did (and SAM wants it to, and i dont know why not)
|
||||
if ( (_session != null) && (!_session.isClosed()) ) {
|
||||
try {
|
||||
_session.destroySession();
|
||||
} catch (I2PSessionException ise) {
|
||||
_log.warn("Unable to destroy the session", ise);
|
||||
}
|
||||
}
|
||||
}
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public I2PSession getSession() {
|
||||
return _session;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve a set of currently connected I2PSockets, either initiated locally or remotely.
|
||||
*
|
||||
*/
|
||||
public Set listSockets() {
|
||||
Set connections = _connectionManager.listConnections();
|
||||
Set rv = new HashSet(connections.size());
|
||||
for (Iterator iter = connections.iterator(); iter.hasNext(); ) {
|
||||
Connection con = (Connection)iter.next();
|
||||
if (con.getSocket() != null)
|
||||
rv.add(con.getSocket());
|
||||
}
|
||||
return rv;
|
||||
}
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public ConnectionManager getConnectionManager() {
|
||||
return _connectionManager;
|
||||
}
|
||||
|
||||
public String getName() { return _name; }
|
||||
public void setName(String name) { _name = name; }
|
||||
|
||||
|
||||
public void addDisconnectListener(I2PSocketManager.DisconnectListener lsnr) {
|
||||
_connectionManager.getMessageHandler().addDisconnectListener(lsnr);
|
||||
}
|
||||
public void removeDisconnectListener(I2PSocketManager.DisconnectListener lsnr) {
|
||||
_connectionManager.getMessageHandler().removeDisconnectListener(lsnr);
|
||||
}
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
* @throws net.i2p.I2PException
|
||||
* @throws java.net.SocketTimeoutException
|
||||
*/
|
||||
public I2PSocket receiveSocket() throws I2PException, SocketTimeoutException {
|
||||
verifySession();
|
||||
Connection con = _connectionManager.getConnectionHandler().accept(_connectionManager.MgetSoTimeout());
|
||||
if(_log.shouldLog(Log.DEBUG)) {
|
||||
_log.debug("receiveSocket() called: " + con);
|
||||
}
|
||||
if(con != null) {
|
||||
I2PSocketFull sock = new I2PSocketFull(con);
|
||||
con.setSocket(sock);
|
||||
return sock;
|
||||
} else {
|
||||
if(_connectionManager.MgetSoTimeout() == -1) {
|
||||
return null;
|
||||
}
|
||||
throw new SocketTimeoutException("I2PSocket timed out");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ping the specified peer, returning true if they replied to the ping within
|
||||
* the timeout specified, false otherwise. This call blocks.
|
||||
*
|
||||
*
|
||||
* @param peer
|
||||
* @param timeoutMs
|
||||
* @return
|
||||
*/
|
||||
public boolean ping(Destination peer, long timeoutMs) {
|
||||
return _connectionManager.ping(peer, timeoutMs);
|
||||
}
|
||||
|
||||
/**
|
||||
* How long should we wait for the client to .accept() a socket before
|
||||
* sending back a NACK/Close?
|
||||
*
|
||||
* @param ms milliseconds to wait, maximum
|
||||
*/
|
||||
public void setAcceptTimeout(long ms) {
|
||||
_acceptTimeout = ms;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public long getAcceptTimeout() {
|
||||
return _acceptTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param options
|
||||
*/
|
||||
public void setDefaultOptions(I2PSocketOptions options) {
|
||||
_defaultOptions = new ConnectionOptions((ConnectionOptions)options);
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public I2PSocketOptions getDefaultOptions() {
|
||||
return _defaultOptions;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public I2PServerSocket getServerSocket() {
|
||||
_connectionManager.setAllowIncomingConnections(true);
|
||||
return _serverSocket;
|
||||
}
|
||||
|
||||
private void verifySession() throws I2PException {
|
||||
if(!_connectionManager.getSession().isClosed()) {
|
||||
return;
|
||||
}
|
||||
_connectionManager.getSession().connect();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new connected socket (block until the socket is created)
|
||||
*
|
||||
* @param peer Destination to connect to
|
||||
* @param options I2P socket options to be used for connecting
|
||||
*
|
||||
* @throws NoRouteToHostException if the peer is not found or not reachable
|
||||
* @throws I2PException if there is some other I2P-related problem
|
||||
*/
|
||||
public I2PSocket connect(Destination peer, I2PSocketOptions options)
|
||||
throws I2PException, NoRouteToHostException {
|
||||
verifySession();
|
||||
if(options == null) {
|
||||
options = _defaultOptions;
|
||||
}
|
||||
ConnectionOptions opts = null;
|
||||
if(options instanceof ConnectionOptions) {
|
||||
opts = new ConnectionOptions((ConnectionOptions)options);
|
||||
} else {
|
||||
opts = new ConnectionOptions(options);
|
||||
}
|
||||
if(_log.shouldLog(Log.INFO)) {
|
||||
_log.info("Connecting to " + peer.calculateHash().toBase64().substring(0, 6) + " with options: " + opts);
|
||||
}
|
||||
Connection con = _connectionManager.connect(peer, opts);
|
||||
if(con == null) {
|
||||
throw new TooManyStreamsException("Too many streams (max " + _maxStreams + ")");
|
||||
}
|
||||
I2PSocketFull socket = new I2PSocketFull(con);
|
||||
con.setSocket(socket);
|
||||
if(con.getConnectionError() != null) {
|
||||
con.disconnect(false);
|
||||
throw new NoRouteToHostException(con.getConnectionError());
|
||||
}
|
||||
return socket;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new connected socket (block until the socket is created)
|
||||
*
|
||||
* @param peer Destination to connect to
|
||||
*
|
||||
* @return
|
||||
* @throws NoRouteToHostException if the peer is not found or not reachable
|
||||
* @throws I2PException if there is some other I2P-related problem
|
||||
*/
|
||||
public I2PSocket connect(Destination peer) throws I2PException, NoRouteToHostException {
|
||||
return connect(peer, _defaultOptions);
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroy the socket manager, freeing all the associated resources. This
|
||||
* method will block untill all the managed sockets are closed.
|
||||
*
|
||||
*/
|
||||
public void destroySocketManager() {
|
||||
_connectionManager.disconnectAllHard();
|
||||
_connectionManager.setAllowIncomingConnections(false);
|
||||
// should we destroy the _session too?
|
||||
// yes, since the old lib did (and SAM wants it to, and i dont know why not)
|
||||
if((_session != null) && (!_session.isClosed())) {
|
||||
try {
|
||||
_session.destroySession();
|
||||
} catch(I2PSessionException ise) {
|
||||
_log.warn("Unable to destroy the session", ise);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve a set of currently connected I2PSockets, either initiated locally or remotely.
|
||||
*
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public Set listSockets() {
|
||||
Set connections = _connectionManager.listConnections();
|
||||
Set rv = new HashSet(connections.size());
|
||||
for(Iterator iter = connections.iterator(); iter.hasNext();) {
|
||||
Connection con = (Connection)iter.next();
|
||||
if(con.getSocket() != null) {
|
||||
rv.add(con.getSocket());
|
||||
}
|
||||
}
|
||||
return rv;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public String getName() {
|
||||
return _name;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param name
|
||||
*/
|
||||
public void setName(String name) {
|
||||
_name = name;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param lsnr
|
||||
*/
|
||||
public void addDisconnectListener(I2PSocketManager.DisconnectListener lsnr) {
|
||||
_connectionManager.getMessageHandler().addDisconnectListener(lsnr);
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param lsnr
|
||||
*/
|
||||
public void removeDisconnectListener(I2PSocketManager.DisconnectListener lsnr) {
|
||||
_connectionManager.getMessageHandler().removeDisconnectListener(lsnr);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@ import net.i2p.util.SimpleTimer;
|
||||
/**
|
||||
*
|
||||
*/
|
||||
class RetransmissionTimer extends SimpleTimer {
|
||||
public class RetransmissionTimer extends SimpleTimer {
|
||||
private static final RetransmissionTimer _instance = new RetransmissionTimer();
|
||||
public static final SimpleTimer getInstance() { return _instance; }
|
||||
protected RetransmissionTimer() { super("StreamingTimer"); }
|
||||
|
||||
Reference in New Issue
Block a user