Util: Support setSoTimeout() for InternalSockets

to keep susimail server connections from hanging too long

core/java/src/net/i2p/util/InternalServerSocket.java

@@ -100,10 +100,10 @@ public class InternalServerSocket extends ServerSocket {
         InternalServerSocket iss = _sockets.get(Integer.valueOf(port));
         if (iss == null)
              throw new IOException("No server for port: " + port);
-        PipedInputStream cis = new PipedInputStream(64*1024);
+        TimeoutPipedInputStream cis = new TimeoutPipedInputStream(64*1024);
-        PipedInputStream sis = new PipedInputStream(64*1024);
+        TimeoutPipedInputStream sis = new TimeoutPipedInputStream(64*1024);
-        PipedOutputStream cos = new PipedOutputStream(sis);
+        PipedOutputStream cos = new TimeoutPipedOutputStream(sis);
-        PipedOutputStream sos = new PipedOutputStream(cis);
+        PipedOutputStream sos = new TimeoutPipedOutputStream(cis);
         clientSock.setInputStream(cis);
         clientSock.setOutputStream(cos);
         iss.queueConnection(new InternalSocket(sis, sos));

core/java/src/net/i2p/util/InternalSocket.java

@@ -98,11 +98,22 @@ public class InternalSocket extends Socket {
         return ("Internal socket");
     }
 
-    // ignored stuff
+    /**
-    /** warning - unsupported */
+     *  Supported as of 0.9.34, if constructed with TimeoutPipedInputStream
+     *  and TimeoutPipedOutputStream. Otherwise, does nothing.
+     *  @see TimeoutPipedInputStream
+     */
     @Override
-    public void setSoTimeout(int timeout) {}
+    public synchronized void setSoTimeout(int timeout) {
+        if (_is != null && _is instanceof TimeoutPipedInputStream)
+            ((TimeoutPipedInputStream) _is).setReadTimeout(timeout);
+    }
 
+    // ignored stuff
+
+    /**
+     *  Always returns 0, even if setSoTimeout() was called.
+     */
     @Override
     public int getSoTimeout () {
         return 0;

core/java/src/net/i2p/util/TimeoutPipedInputStream.java

@@ -0,0 +1,115 @@
+package net.i2p.util;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.net.SocketTimeoutException;
+
+/**
+ *  Adds setReadTimeout().
+ *  Must be used with a TimeoutPipedOutputStream.
+ *
+ *  To support InternalSocket.setSoTimeout().
+ *  Package private, not a part of the public API, not for general use.
+ *
+ *  @since 0.9.34
+ */
+class TimeoutPipedInputStream extends PipedInputStream {
+
+    private int timeout;
+    // local version of pkg private in super
+    private boolean _closedByWriter;
+    // local version of pkg private in super
+    private volatile boolean _closedByReader;
+
+    public TimeoutPipedInputStream(int pipeSize) {
+        super(pipeSize);
+    }
+
+    /**
+     * @throws SocketTimeoutException if timeout is reached
+     */
+    @Override
+    public synchronized int read() throws IOException {
+        // This is similar to what is done in super, but with a timeout.
+        // We use local copies of closedByReader and closedByWriter as
+        // those are package private in super.
+        // This doesn't add any substantial runtime overhead,
+        // as we are just doing the 1-second wait and loop here
+        // instead of in super.
+        // If a timeout is set, we will always wait here instead of in super.
+        if (in < 0 && timeout > 0 && !_closedByReader) {
+            long now = System.currentTimeMillis();
+            long end = now + timeout;
+            while (true) {
+                if (_closedByWriter)
+                    return -1;
+                try {
+                    wait(Math.max(1L, Math.min(1000L, end - now)));
+                } catch (InterruptedException ex) {
+                    throw new InterruptedIOException();
+                }
+                if (in >= 0 || _closedByReader)
+                    break;
+                now = System.currentTimeMillis();
+                if (now >= end)
+                    throw new SocketTimeoutException();
+            }
+        }
+        return super.read();
+    }
+
+    /**
+     *  Must be called before blocking read call.
+     *  @param ms less than or equal to zero means forever
+     */
+    public void setReadTimeout(int ms) {
+        timeout = Math.max(0, ms);
+    }
+
+    /**
+     *  To save state.
+     *  We have to do this because can't get to closedByWriter in super.
+     */
+    synchronized void x_receivedLast() {
+        _closedByWriter = true;
+        notifyAll();
+    }
+
+    /**
+     *  Overridden to save state.
+     *  We have to do this because can't get to closedByReader in super
+     */
+    @Override
+    public void close() throws IOException {
+        _closedByReader = true;
+        super.close();
+    }
+
+/****
+    public static void main(String[] args) throws IOException {
+        TimeoutPipedInputStream in = new TimeoutPipedInputStream(1024);
+        TimeoutPipedOutputStream out = new TimeoutPipedOutputStream(in);
+        out.write('a');
+        in.setReadTimeout(5555);
+        long start = System.currentTimeMillis();
+        try {
+            int a = in.read();
+            if (a == 'a')
+                System.out.println("got 1 (pass)");
+            else
+                System.out.println("bad data (fail)");
+            in.read();
+            System.out.println("got 2 (fail)");
+        } catch (IOException ioe) {
+            ioe.printStackTrace();
+            System.out.println("got ioe (pass)");
+        }
+        System.out.println("took " + (System.currentTimeMillis() - start));
+        in.setReadTimeout(0);
+        System.out.println("wait forever");
+        in.read();
+    }
+****/
+}

core/java/src/net/i2p/util/TimeoutPipedOutputStream.java

@@ -0,0 +1,34 @@
+package net.i2p.util;
+
+import java.io.IOException;
+import java.io.PipedOutputStream;
+
+/**
+ *  Helper for TimeoutPipedInputStream.
+ *  There isn't any timeout implemented here.
+ *
+ *  To support InternalSocket.setSoTimeout().
+ *  Package private, not a part of the public API, not for general use.
+ *
+ *  @see TimeoutPipedInputStream
+ *  @since 0.9.34
+ */
+class TimeoutPipedOutputStream extends PipedOutputStream {
+
+    private final TimeoutPipedInputStream sink;
+
+    public TimeoutPipedOutputStream(TimeoutPipedInputStream snk) throws IOException {
+        super(snk);
+        sink = snk;
+    }
+
+    /**
+     *  Overridden only so we can tell snk.
+     *  We have to do this because TPIS can't get to pkg private receivedLast() in super.
+     */
+    @Override
+    public void close() throws IOException {
+        sink.x_receivedLast();
+        super.close();
+    }
+}

history.txt

@@ -1,3 +1,10 @@
+2018-02-22 zzz
+ * Util: Support setSoTimeout() for InternalSockets
+
+2018-02-21 zzz
+ * Console: Hide options on /configupdate if a package (ticket #2172)
+ * SusiMail: Add Date header to sent messages
+
 2018-02-20 zzz
  * Console:
    - Redirect to HTTPS if available (ticket #2160)

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 = 8;
+    public final static long BUILD = 9;
 
     /** for example "-test" */
     public final static String EXTRA = "";