NTCP: Retain pending messages when replacing connection

history.txt

@@ -1,6 +1,20 @@
+2020-04-08 zzz
+ * i2psnark: Give peers preference to get first pieces (ticket #2473)
+ * NetDB: Remove class M from auto-floodfill
+ * NTCP: Retain pending messages when replacing connection
+ * Ratchet: TagSet cleanups
+
+2020-04-07 zzz
+ * Console: Fix disabling sidebar refresh
+ * Graphs:
+   - Reduce rrd4j sync thread pool size
+   - Disable pool if not persisting
+   - Stop pool on shutdown
+
 2020-04-06 zzz
  * Ratchet:
    - Finish Next Key impl.
+   - Simplify OB Session
    - Performance improvements and cleanups
    - Debug page fixes
 

router/java/src/net/i2p/router/RouterVersion.java

@@ -18,7 +18,7 @@ public class RouterVersion {
     /** deprecated */
     public final static String ID = "Monotone";
     public final static String VERSION = CoreVersion.VERSION;
-    public final static long BUILD = 6;
+    public final static long BUILD = 7;
 
     /** for example "-test" */
     public final static String EXTRA = "";

router/java/src/net/i2p/router/transport/ntcp/NTCPConnection.java

@@ -342,8 +342,9 @@ public class NTCPConnection implements Closeable {
     void finishInboundEstablishment(SessionKey key, long clockSkew, byte prevWriteEnd[], byte prevReadEnd[]) {
         NTCPConnection toClose = locked_finishInboundEstablishment(key, clockSkew, prevWriteEnd, prevReadEnd);
         if (toClose != null) {
-            if (_log.shouldLog(Log.DEBUG))
-                _log.debug("Old connection closed: " + toClose + " replaced by " + this);
+            int drained = toClose.drainOutboundTo(_outbound);
+            if (_log.shouldWarn())
+                _log.warn("Old connection closed: " + toClose + " replaced by " + this + "; drained " + drained);
             _context.statManager().addRateData("ntcp.inboundEstablishedDuplicate", toClose.getUptime());
             toClose.close();
         }
@@ -412,6 +413,24 @@ public class NTCPConnection implements Closeable {
             return ! _currentOutbound.isEmpty();
         }
     }
+
+    /**
+     *  Drain any pending outbound messages to a new queue
+     *  @return number drained
+     *  @since 0.9.46
+     */
+    private int drainOutboundTo(Queue<OutNetMessage> to) {
+        int rv = 0;
+        synchronized (_currentOutbound) {
+            rv = _currentOutbound.size();
+            if (rv > 0) {
+                to.addAll(_currentOutbound);
+                _currentOutbound.clear();
+            }
+            rv += _outbound.drainTo(to);
+        }
+        return rv;
+    }
     
     /** @return milliseconds */
     public long getTimeSinceSend() { return _context.clock().now()-_lastSendTime; }
@@ -1860,8 +1879,9 @@ public class NTCPConnection implements Closeable {
         }
         NTCPConnection toClose = _transport.inboundEstablished(this);
         if (toClose != null && toClose != this) {
+            int drained = toClose.drainOutboundTo(_outbound);
             if (_log.shouldWarn())
-                _log.warn("Old connection closed: " + toClose + " replaced by " + this);
+                _log.warn("Old connection closed: " + toClose + " replaced by " + this + "; drained " + drained);
             _context.statManager().addRateData("ntcp.inboundEstablishedDuplicate", toClose.getUptime());
             toClose.close();
         }