Merge pull request #46 from square/jwilson/close

Tests and code for various SPDY close scenarios.

src/main/java/libcore/net/spdy/SpdyConnection.java

@@ -72,18 +72,23 @@ public final class SpdyConnection implements Closeable {
     static final int TYPE_HEADERS = 0x8;
     static final int VERSION = 2;
 
-    private final SpdyReader spdyReader;
-    private final SpdyWriter spdyWriter;
-    private final ExecutorService readExecutor;
-    private final ExecutorService writeExecutor;
-    private final ExecutorService callbackExecutor;
+    /**
+     * True if this peer initiated the connection.
+     */
+    final boolean client;
 
     /**
      * User code to run in response to an incoming stream. Callbacks must not be
      * run on the callback executor.
      */
     private final IncomingStreamHandler handler;
 
+    private final SpdyReader spdyReader;
+    private final SpdyWriter spdyWriter;
+    private final ExecutorService readExecutor;
+    private final ExecutorService writeExecutor;
+    private final ExecutorService callbackExecutor;
+
     private final Map<Integer, SpdyStream> streams = new HashMap<Integer, SpdyStream>();
     private int nextStreamId;
 
@@ -95,9 +100,10 @@ public final class SpdyConnection implements Closeable {
     Settings settings;
 
     private SpdyConnection(Builder builder) {
+        client = builder.client;
+        handler = builder.handler;
         spdyReader = new SpdyReader(builder.in);
         spdyWriter = new SpdyWriter(builder.out);
-        handler = builder.handler;
         nextStreamId = builder.client ? 1 : 2;
         nextPingId = builder.client ? 1 : 2;
 
@@ -106,19 +112,12 @@ private SpdyConnection(Builder builder) {
                 new SynchronousQueue<Runnable>(), newThreadFactory(prefix + "Reader", false));
         writeExecutor = new ThreadPoolExecutor(0, 1, 60, TimeUnit.SECONDS,
                 new LinkedBlockingQueue<Runnable>(), newThreadFactory(prefix + "Writer", false));
-        callbackExecutor = new ThreadPoolExecutor(1, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
+        callbackExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
                 new SynchronousQueue<Runnable>(), newThreadFactory(prefix + "Callbacks", false));
 
         readExecutor.execute(new Reader());
     }
 
-    /**
-     * Returns true if this peer initiated the connection.
-     */
-    public synchronized boolean isClient() {
-        return nextStreamId % 2 == 1;
-    }
-
     private synchronized SpdyStream getStream(int id) {
         return streams.get(id);
     }
@@ -320,12 +319,19 @@ private class Reader implements Runnable, SpdyReader.Handler {
                 int priority, List<String> nameValueBlock) {
             final SpdyStream synStream = new SpdyStream(streamId, SpdyConnection.this,
                     nameValueBlock, flags);
-            SpdyStream previous;
+            final SpdyStream previous;
             synchronized (SpdyConnection.this) {
                 previous = streams.put(streamId, synStream);
             }
             if (previous != null) {
-                previous.close(SpdyStream.RST_PROTOCOL_ERROR);
+                writeExecutor.execute(new Runnable() {
+                    @Override public void run() {
+                        try {
+                            previous.close(SpdyStream.RST_PROTOCOL_ERROR);
+                        } catch (IOException ignored) {
+                        }
+                    }
+                });
                 return;
             }
             callbackExecutor.execute(new Runnable() {
@@ -372,7 +378,7 @@ private class Reader implements Runnable, SpdyReader.Handler {
         }
 
         @Override public void ping(int flags, int streamId) {
-            if (isClient() != (streamId % 2 == 1)) {
+            if (client != (streamId % 2 == 1)) {
                 // Respond to a client ping if this is a server and vice versa.
                 writePingLater(streamId, null);
             } else {

src/main/java/libcore/net/spdy/SpdyServer.java

@@ -82,7 +82,8 @@ private void send404(SpdyStream stream, String path) throws IOException {
                 "version", "HTTP/1.1",
                 "content-type", "text/plain"
         );
-        OutputStream out = stream.reply(responseHeaders);
+        stream.reply(responseHeaders, true);
+        OutputStream out = stream.getOutputStream();
         String text = "Not found: " + path;
         out.write(text.getBytes("UTF-8"));
         out.close();
@@ -91,11 +92,12 @@ private void send404(SpdyStream stream, String path) throws IOException {
     private void serveFile(SpdyStream stream, File file) throws IOException {
         InputStream in = new FileInputStream(file);
         byte[] buffer = new byte[8192];
-        OutputStream out = stream.reply(Arrays.asList(
+        stream.reply(Arrays.asList(
                 "status", "200",
                 "version", "HTTP/1.1",
                 "content-type", contentType(file)
-        ));
+        ), true);
+        OutputStream out = stream.getOutputStream();
         int count;
         while ((count = in.read(buffer)) != -1) {
             out.write(buffer, 0, count);