i2ptunnel: Close input stream when HTTP client decompressor terminates (ticket #1506)

streaming: Minor cleanups, log tweaks

apps/i2ptunnel/java/src/net/i2p/i2ptunnel/HTTPResponseOutputStream.java

@@ -309,7 +309,7 @@ class HTTPResponseOutputStream extends FilterOutputStream {
             } catch (OutOfMemoryError oom) {
                 _log.error("OOM in HTTP Decompressor", oom);
             } finally {
-                if (_log.shouldLog(Log.INFO) && (_in != null))
+                if (_log.shouldInfo())
                     _log.info("After decompression, written=" + written + 
                                 " read=" + _in.getTotalRead() 
                                 + ", expanded=" + _in.getTotalExpanded() + ", remaining=" + _in.getRemaining() 
@@ -319,9 +319,11 @@ class HTTPResponseOutputStream extends FilterOutputStream {
                 if (_out != null) try { 
                     _out.close(); 
                 } catch (IOException ioe) {}
+                try { 
+                    _in.close(); 
+                } catch (IOException ioe) {}
             }
 
-            if (_in != null) {
             double compressed = _in.getTotalRead();
             double expanded = _in.getTotalExpanded();
             ReusableGZIPInputStream.release(_in);
@@ -334,7 +336,6 @@ class HTTPResponseOutputStream extends FilterOutputStream {
             }
         }
     }
-    }
 
 /*******
     public static void main(String args[]) {

apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnel.java

@@ -1517,7 +1517,7 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
         if (tasks.isEmpty()) {
             System.exit(0);
         }
-        l.log("There are running tasks. Try 'list'.");
+        l.log("There are running tasks. Try 'list' or 'close all'.");
         //notifyEvent("quitResult", "error");
     }
 

apps/streaming/java/src/net/i2p/client/streaming/impl/I2PSocketFull.java

@@ -52,7 +52,7 @@ class I2PSocketFull implements I2PSocket {
         Connection c = _connection;
         if (c == null) return;
         if (log.shouldLog(Log.INFO))
-            log.info("close() called, connected? " + c.getIsConnected() + " : " + c);
+            log.info("close() called, connected? " + c.getIsConnected() + " : " + c, new Exception());
         if (c.getIsConnected()) {
             MessageInputStream in = c.getInputStream();
             in.close();

apps/streaming/java/src/net/i2p/client/streaming/impl/MessageInputStream.java

@@ -62,6 +62,9 @@ class MessageInputStream extends InputStream {
     private final byte[] _oneByte = new byte[1];
     private final Object _dataLock;
 
+    /** only in _notYetReadyBlocks, never in _readyDataBlocks */
+    private static final ByteArray DUMMY_BA = new ByteArray(null);
+    
     private static final int MIN_READY_BUFFERS = 16;
 
 
@@ -320,9 +323,9 @@ class MessageInputStream extends InputStream {
                 _highestReadyBlockId = messageId;
                 long cur = _highestReadyBlockId + 1;
                 // now pull in any previously pending blocks
-                while (_notYetReadyBlocks.containsKey(Long.valueOf(cur))) {
-                    ByteArray ba = _notYetReadyBlocks.remove(Long.valueOf(cur));
-                    if ( (ba != null) && (ba.getData() != null) && (ba.getValid() > 0) ) {
+                ByteArray ba;
+                while ((ba = _notYetReadyBlocks.remove(Long.valueOf(cur))) != null) {
+                    if (ba.getData() != null && ba.getValid() > 0) {
                         _readyDataBlocks.add(ba);
                     }
                     
@@ -336,14 +339,18 @@ class MessageInputStream extends InputStream {
                                         // Java throws a SocketTimeoutException.
                                         // We do neither.
             } else {
-                if (_log.shouldLog(Log.INFO))
+                // _notYetReadyBlocks size is limited in canAccept()
+                if (_locallyClosed) {
+                    if (_log.shouldInfo())
+                        _log.info("Message received on closed stream: " + messageId);
+                    // dont need the payload, just the msgId in order
+                    _notYetReadyBlocks.put(Long.valueOf(messageId), DUMMY_BA);
+                } else {
+                    if (_log.shouldInfo())
                         _log.info("Message is out of order: " + messageId);
-                // _notYetReadyBlocks size is limited in ConnectionPacketHandler.
-                if (_locallyClosed) // dont need the payload, just the msgId in order
-                    _notYetReadyBlocks.put(Long.valueOf(messageId), new ByteArray(null));
-                else
                     _notYetReadyBlocks.put(Long.valueOf(messageId), payload);
                 }
+            }
             _dataLock.notifyAll();
         }
         return true;