echelon/i2p.i2p
1
0
Fork 0
forked from I2P_Developers/i2p.i2p
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

* Streaming:

Browse Source 
- Set ports on many packets that were missing them
   - Use connection throttling methods on pings too (ticket #1142)
   - Add methods to set ports on pings
   - Argument checking on ping methods
This commit is contained in:
zzz
2014-02-02 16:47:29 +00:00
parent b7207fd29f
commit 3622501471
6 changed files with 130 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java
View File

@ -307,6 +307,8 @@ class Connection {
reply.setSendStreamId(_sendStreamId);
reply.setReceiveStreamId(_receiveStreamId);
reply.setOptionalFrom(_connectionManager.getSession().getMyDestination());
reply.setLocalPort(_localPort);
reply.setRemotePort(_remotePort);
// this just sends the packet - no retries or whatnot
if (_outboundQueue.enqueue(reply)) {
_unackedPacketsReceived = 0;

4
apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionDataReceiver.java
View File

@ -206,9 +206,9 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
packet.setFlag(Packet.FLAG_SYNCHRONIZE);
packet.setOptionalFrom(con.getSession().getMyDestination());
packet.setOptionalMaxSize(con.getOptions().getMaxMessageSize());
packet.setLocalPort(con.getLocalPort());
packet.setRemotePort(con.getPort());
}
packet.setLocalPort(con.getLocalPort());
packet.setRemotePort(con.getPort());
if (con.getSendStreamId() == Packet.STREAM_ID_UNKNOWN) {
packet.setFlag(Packet.FLAG_NO_ACK);
}

81
apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionManager.java
View File

@ -262,6 +262,8 @@ class ConnectionManager {
reply.setSendStreamId(synPacket.getReceiveStreamId());
reply.setReceiveStreamId(0);
reply.setOptionalFrom(_session.getMyDestination());
reply.setLocalPort(synPacket.getLocalPort());
reply.setRemotePort(synPacket.getRemotePort());
// this just sends the packet - no retries or whatnot
_outboundQueue.enqueue(reply);
return null;
@ -284,6 +286,44 @@ class ConnectionManager {
return con;
}
/**
* Process a ping by checking for throttling, etc., then sending a pong.
*
* @param con null if unknown
* @param ping Ping packet to process, must have From and Sig fields,
* with signature already verified, only if answerPings() returned true
* @return true if we sent a pong
* @since 0.9.12 from PacketHandler.receivePing()
*/
public boolean receivePing(Connection con, Packet ping) {
Destination dest = ping.getOptionalFrom();
if (dest == null)
return false;
if (con == null) {
// Use the same throttling as for connections
String why = shouldRejectConnection(ping);
if (why != null) {
if ((!_defaultOptions.getDisableRejectLogging()) || _log.shouldLog(Log.WARN))
_log.logAlways(Log.WARN, "Dropping ping since peer is " + why + ": " + dest.calculateHash());
return false;
}
} else {
// in-connection ping to a 3rd party ???
if (!dest.equals(con.getRemotePeer())) {
_log.logAlways(Log.WARN, "Dropping ping from " + con.getRemotePeer().calculateHash() +
" to " + dest.calculateHash());
return false;
}
}
PacketLocal pong = new PacketLocal(_context, dest);
pong.setFlag(Packet.FLAG_ECHO | Packet.FLAG_NO_ACK);
pong.setReceiveStreamId(ping.getSendStreamId());
pong.setLocalPort(ping.getLocalPort());
pong.setRemotePort(ping.getRemotePort());
_outboundQueue.enqueue(pong);
return true;
}
private static final long DEFAULT_STREAM_DELAY_MAX = 10*1000;
/**
@ -572,24 +612,34 @@ class ConnectionManager {
return new HashSet<Connection>(_connectionByInboundId.values());
}
/** blocking */
public boolean ping(Destination peer, long timeoutMs) {
return ping(peer, timeoutMs, true, null);
}
public boolean ping(Destination peer, long timeoutMs, boolean blocking) {
return ping(peer, timeoutMs, blocking, null);
/**
* blocking
*
* @param timeoutMs greater than zero
* @return true if pong received
* @since 0.9.12 added port args
*/
public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs) {
return ping(peer, fromPort, toPort, timeoutMs, true, null);
}
/**
* @deprecated I2PSession ignores tags, use non-tag variant
* @param keyToUse ignored
* @param tagsToSend ignored
* @param timeoutMs greater than zero
* @return true if blocking and pong received
* @since 0.9.12 added port args
*/
public boolean ping(Destination peer, long timeoutMs, boolean blocking, SessionKey keyToUse, Set<?> tagsToSend, PingNotifier notifier) {
return ping(peer, timeoutMs, blocking, notifier);
public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs, boolean blocking) {
return ping(peer, fromPort, toPort, timeoutMs, blocking, null);
}
public boolean ping(Destination peer, long timeoutMs, boolean blocking, PingNotifier notifier) {
/**
* @param timeoutMs greater than zero
* @param notifier may be null
* @return true if blocking and pong received
* @since 0.9.12 added port args
*/
public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs,
boolean blocking, PingNotifier notifier) {
Long id = Long.valueOf(_context.random().nextLong(Packet.MAX_STREAM_ID-1)+1);
PacketLocal packet = new PacketLocal(_context, peer);
packet.setSendStreamId(id.longValue());
@ -597,13 +647,15 @@ class ConnectionManager {
Packet.FLAG_NO_ACK |
Packet.FLAG_SIGNATURE_INCLUDED);
packet.setOptionalFrom(_session.getMyDestination());
packet.setLocalPort(fromPort);
packet.setRemotePort(toPort);
//if ( (keyToUse != null) && (tagsToSend != null) ) {
// packet.setKeyUsed(keyToUse);
// packet.setTagsSent(tagsToSend);
//}
if (_log.shouldLog(Log.INFO)) {
_log.info(String.format("about to ping %s timeout=%d blocking=%b",
peer,timeoutMs,blocking));
_log.info(String.format("about to ping %s port %d from port %d timeout=%d blocking=%b",
peer.calculateHash().toString(), toPort, fromPort, timeoutMs, blocking));
}
@ -658,6 +710,7 @@ class ConnectionManager {
private boolean _ponged;
private final PingNotifier _notifier;
/** @param notifier may be null */
public PingRequest(PingNotifier notifier) {
_notifier = notifier;
}

32
apps/streaming/java/src/net/i2p/client/streaming/impl/I2PSocketManagerFull.java
View File

@ -148,13 +148,41 @@ public class I2PSocketManagerFull implements I2PSocketManager {
* Ping the specified peer, returning true if they replied to the ping within
* the timeout specified, false otherwise. This call blocks.
*
* Uses the ports from the default options.
*
* @param peer
* @param timeoutMs
* @param timeoutMs timeout in ms, greater than zero
* @return true on success, false on failure
* @throws IllegalArgumentException
*/
public boolean ping(Destination peer, long timeoutMs) {
return _connectionManager.ping(peer, timeoutMs);
if (timeoutMs <= 0)
throw new IllegalArgumentException("bad timeout");
return _connectionManager.ping(peer, _defaultOptions.getLocalPort(),
_defaultOptions.getPort(), timeoutMs);
}
/**
* Ping the specified peer, returning true if they replied to the ping within
* the timeout specified, false otherwise. This call blocks.
*
* Uses the ports specified.
*
* @param peer Destination to ping
* @param localPort 0 - 65535
* @param remotePort 0 - 65535
* @param timeoutMs timeout in ms, greater than zero
* @return success or failure
* @throws IllegalArgumentException
* @since 0.9.12
*/
public boolean ping(Destination peer, int localPort, int remotePort, long timeoutMs) {
if (localPort < 0 || localPort > 65535 ||
remotePort < 0 || remotePort > 65535)
throw new IllegalArgumentException("bad port");
if (timeoutMs <= 0)
throw new IllegalArgumentException("bad timeout");
return _connectionManager.ping(peer, localPort, remotePort, timeoutMs);
}
/**

16
apps/streaming/java/src/net/i2p/client/streaming/impl/PacketHandler.java
View File

@ -136,7 +136,7 @@ class PacketHandler {
if (packet.isFlagSet(Packet.FLAG_ECHO)) {
if (packet.getSendStreamId() > 0) {
if (con.getOptions().getAnswerPings())
receivePing(packet);
receivePing(con, packet);
else if (_log.shouldLog(Log.WARN))
_log.warn("Dropping Echo packet on existing con: " + packet);
} else if (packet.getReceiveStreamId() > 0) {
@ -247,6 +247,8 @@ class PacketHandler {
reply.setSendStreamId(packet.getReceiveStreamId());
reply.setReceiveStreamId(packet.getSendStreamId());
reply.setOptionalFrom(_manager.getSession().getMyDestination());
reply.setLocalPort(packet.getLocalPort());
reply.setRemotePort(packet.getRemotePort());
// this just sends the packet - no retries or whatnot
_manager.getPacketQueue().enqueue(reply);
}
@ -255,7 +257,7 @@ class PacketHandler {
if (packet.isFlagSet(Packet.FLAG_ECHO)) {
if (packet.getSendStreamId() > 0) {
if (_manager.answerPings())
receivePing(packet);
receivePing(null, packet);
else if (_log.shouldLog(Log.WARN))
_log.warn("Dropping Echo packet on unknown con: " + packet);
} else if (packet.getReceiveStreamId() > 0) {
@ -335,7 +337,10 @@ class PacketHandler {
}
}
private void receivePing(Packet packet) {
/**
* @param con null if unknown
*/
private void receivePing(Connection con, Packet packet) {
boolean ok = packet.verifySignature(_context, packet.getOptionalFrom(), null);
if (!ok) {
if (_log.shouldLog(Log.WARN)) {
@ -348,10 +353,7 @@ class PacketHandler {
+ " sig=" + packet.getOptionalSignature().toBase64() + ")");
}
} else {
PacketLocal pong = new PacketLocal(_context, packet.getOptionalFrom());
pong.setFlag(Packet.FLAG_ECHO | Packet.FLAG_NO_ACK);
pong.setReceiveStreamId(packet.getSendStreamId());
_manager.getPacketQueue().enqueue(pong);
_manager.receivePing(con, packet);
}
}