qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r..@apache.org
Subject [12/38] qpid-proton git commit: PROTON-881: Add error handling for "accept socket connection" path through Acceptor
Date Sun, 05 Jul 2015 23:45:13 GMT
PROTON-881: Add error handling for "accept socket connection" path through Acceptor

Adds error handling for various problems that can occur when accepting
a connection (to a listening socket).  Also adds unittests that cover
these cases.


Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/5748bb98
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/5748bb98
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/5748bb98

Branch: refs/heads/master
Commit: 5748bb9880432bab64c42d34f4d7277163077427
Parents: 51529f6
Author: Adrian Preston <prestona@uk.ibm.com>
Authored: Sat May 2 22:30:40 2015 +0100
Committer: Adrian Preston <prestona@uk.ibm.com>
Committed: Wed May 6 23:24:41 2015 +0100

----------------------------------------------------------------------
 .../apache/qpid/proton/reactor/Acceptor.java    |  5 +-
 .../qpid/proton/reactor/impl/AcceptorImpl.java  | 21 ++++--
 .../qpid/proton/reactor/impl/ReactorImpl.java   | 16 +++--
 .../reactor/impl/ReactorInternalException.java  | 23 ++++++
 .../qpid/proton/reactor/impl/SelectorImpl.java  |  2 +-
 .../apache/qpid/proton/reactor/ReactorTest.java | 76 ++++++++++++++++++++
 .../proton/reactor/impl/AcceptorImplTest.java   | 74 +++++++++++++++++++
 7 files changed, 205 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/5748bb98/proton-j/src/main/java/org/apache/qpid/proton/reactor/Acceptor.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/Acceptor.java b/proton-j/src/main/java/org/apache/qpid/proton/reactor/Acceptor.java
index 91b20a3..f2a320b 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/reactor/Acceptor.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/Acceptor.java
@@ -21,7 +21,10 @@
 
 package org.apache.qpid.proton.reactor;
 
-public interface Acceptor {
+import org.apache.qpid.proton.engine.Handler;
 
+public interface Acceptor extends ReactorChild {
+
+    void add(Handler handler);
     void close();
 }

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/5748bb98/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/AcceptorImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/AcceptorImpl.java
b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/AcceptorImpl.java
index efe0eb4..7084dfb 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/AcceptorImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/AcceptorImpl.java
@@ -45,6 +45,9 @@ public class AcceptorImpl implements Acceptor {
             Reactor reactor = selectable.getReactor();
             try {
                 SocketChannel socketChannel = ((ServerSocketChannel)selectable.getChannel()).accept();
+                if (socketChannel == null) {
+                    throw new ReactorInternalException("Selectable readable, but no socket
to accept");
+                }
                 Handler handler = (Handler)selectable.getAttachment();
                 if (handler == null) {
                     // TODO: set selectable.getAttachment() to null?
@@ -61,8 +64,7 @@ public class AcceptorImpl implements Acceptor {
                 trans.bind(conn);
                 IOHandler.selectableTransport(reactor, socketChannel.socket(), trans);  //
TODO: could we pass in a channel object instead of doing socketChannel.socket()?
             } catch(IOException ioException) {
-                ioException.printStackTrace();
-                // TODO: what do we do with this exception?
+                sel.error();
             }
         }
     }
@@ -81,10 +83,16 @@ public class AcceptorImpl implements Acceptor {
 
     private final Selectable sel;
 
+    // Split out from AcceptorImpl to make it easier for unittests to mock this class
+    // without having to open an actual port.
+    protected ServerSocketChannel openServerSocket() throws IOException {
+        return ServerSocketChannel.open();
+    }
+
     protected AcceptorImpl(Reactor reactor, String host, int port, Handler handler) throws
IOException {
-        ServerSocketChannel ssc = ServerSocketChannel.open();
+        ServerSocketChannel ssc = openServerSocket();
         ssc.bind(new InetSocketAddress(host, port));
-        sel = reactor.selectable();
+        sel = ((ReactorImpl)reactor).selectable(this);
         sel.setChannel(ssc);
         sel.onReadable(new AcceptorReadable());
         sel.onFinalize(new AcceptorFinalize()); // TODO: currently, this is not called from
anywhere!!
@@ -109,4 +117,9 @@ public class AcceptorImpl implements Acceptor {
             reactor.update(sel);
         }
     }
+
+    @Override
+    public void add(Handler handler) {
+        sel.add(handler);
+    }
 }

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/5748bb98/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorImpl.java
index 684d8f8..a3180ab 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorImpl.java
@@ -195,12 +195,14 @@ public class ReactorImpl implements Reactor {
 
     private class ReleaseCallback implements Callback {
         private final ReactorImpl reactor;
-        public ReleaseCallback(ReactorImpl reactor) {
+        private final ReactorChild child;
+        public ReleaseCallback(ReactorImpl reactor, ReactorChild child) {
             this.reactor = reactor;
+            this.child = child;
         }
         @Override
         public void run(Selectable selectable) {
-            if (reactor.children.remove(selectable)) {
+            if (reactor.children.remove(child)) {
                 --reactor.selectables;
             }
         }
@@ -208,18 +210,20 @@ public class ReactorImpl implements Reactor {
 
     @Override
     public Selectable selectable() {
+        return selectable(null);
+    }
+
+    public Selectable selectable(ReactorChild child) {
         Selectable result = new SelectableImpl();
         result.setCollector(collector);
         collector.put(Type.SELECTABLE_INIT, result);
         result.setReactor(this);
-        children.add(result);
-        result.onRelease(new ReleaseCallback(this));
+        children.add(child == null ? result : child);
+        result.onRelease(new ReleaseCallback(this, child == null ? result : child));
         ++selectables;
         return result;
     }
 
-
-
     @Override
     public void update(Selectable selectable) {
         SelectableImpl selectableImpl = (SelectableImpl)selectable;

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/5748bb98/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorInternalException.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorInternalException.java
b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorInternalException.java
new file mode 100644
index 0000000..d8d67ad
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/ReactorInternalException.java
@@ -0,0 +1,23 @@
+package org.apache.qpid.proton.reactor.impl;
+
+/**
+ * Thrown by the reactor when it encounters an internal error condition.
+ * This is analogous to an assertion failure in the proton-c reactor
+ * implementation.
+ */
+class ReactorInternalException extends RuntimeException {
+
+    private static final long serialVersionUID = 8979674526584642454L;
+
+    protected ReactorInternalException(String msg) {
+        super(msg);
+    }
+
+    protected ReactorInternalException(Throwable cause) {
+        super(cause);
+    }
+
+    protected ReactorInternalException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/5748bb98/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/SelectorImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/SelectorImpl.java
b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/SelectorImpl.java
index e4e284e..a53be7d 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/SelectorImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/SelectorImpl.java
@@ -40,7 +40,7 @@ public class SelectorImpl implements Selector {
     private final HashSet<Selectable> error = new HashSet<Selectable>();
 
     public SelectorImpl() throws IOException {
-        selector = java.nio.channels.Selector.open();
+        selector = java.nio.channels.Selector.open();   // TODO: need to ensure we close
this somewhere...
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/5748bb98/proton-j/src/test/java/org/apache/qpid/proton/reactor/ReactorTest.java
----------------------------------------------------------------------
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/reactor/ReactorTest.java b/proton-j/src/test/java/org/apache/qpid/proton/reactor/ReactorTest.java
new file mode 100644
index 0000000..2378d5f
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/reactor/ReactorTest.java
@@ -0,0 +1,76 @@
+package org.apache.qpid.proton.reactor;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.apache.qpid.proton.Proton;
+import org.apache.qpid.proton.engine.BaseHandler;
+import org.apache.qpid.proton.engine.Event;
+import org.apache.qpid.proton.engine.Event.Type;
+import org.junit.Test;
+
+public class ReactorTest {
+
+    /**
+     * Tests that creating a reactor and running it:
+     * <ul>
+     *   <li>Doesn't throw any exceptions.</li>
+     *   <li>Returns immediately from the run method (as there is no more work to do).</li>
+     * </ul>
+     * @throws IOException
+     */
+    @Test
+    public void runEmpty() throws IOException {
+        Reactor reactor = Proton.reactor();
+        assertNotNull(reactor);
+        reactor.run();
+    }
+
+    private static class TestHandler extends BaseHandler {
+        private final ArrayList<Type> actual = new ArrayList<Type>();
+
+        @Override
+        public void onUnhandled(Event event) {
+            actual.add(event.getType());
+        }
+
+        public void assertEvents(Type...expected) {
+            assertArrayEquals(expected, actual.toArray());
+        }
+    }
+
+    /**
+     * Tests basic operation of the Reactor.acceptor method by creating an acceptor
+     * which is immediately closed by the reactor.  The expected behaviour is for:
+     * <ul>
+     *   <li>The reactor to end immediately (as it has no more work to process).</li>
+     *   <li>The handler, associated with the acceptor, to receive init, update and
+     *       final events.</li>
+     *   <li>For it's lifetime, the acceptor is one of the reactor's children.</li>
+     * </ul>
+     * @throws IOException
+     */
+    @Test
+    public void basicAcceptor() throws IOException {
+        Reactor reactor = Proton.reactor();
+        final Acceptor acceptor = reactor.acceptor("localhost", 0);
+        assertNotNull(acceptor);
+        assertTrue("acceptor should be one of the reactor's children", reactor.children().contains(acceptor));
+        TestHandler acceptorHandler = new TestHandler();
+        acceptor.add(acceptorHandler);
+        reactor.getHandler().add(new BaseHandler() {
+            @Override
+            public void onReactorInit(Event event) {
+                acceptor.close();
+            }
+        });
+        reactor.run();
+        acceptorHandler.assertEvents(Type.SELECTABLE_INIT, Type.SELECTABLE_UPDATED, Type.SELECTABLE_FINAL);
+        assertFalse("acceptor should have been removed from the reactor's children", reactor.children().contains(acceptor));
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/5748bb98/proton-j/src/test/java/org/apache/qpid/proton/reactor/impl/AcceptorImplTest.java
----------------------------------------------------------------------
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/reactor/impl/AcceptorImplTest.java
b/proton-j/src/test/java/org/apache/qpid/proton/reactor/impl/AcceptorImplTest.java
new file mode 100644
index 0000000..44605ae
--- /dev/null
+++ b/proton-j/src/test/java/org/apache/qpid/proton/reactor/impl/AcceptorImplTest.java
@@ -0,0 +1,74 @@
+package org.apache.qpid.proton.reactor.impl;
+
+import java.io.IOException;
+import java.nio.channels.ServerSocketChannel;
+
+import org.apache.qpid.proton.engine.Handler;
+import org.apache.qpid.proton.reactor.Reactor;
+import org.apache.qpid.proton.reactor.ReactorChild;
+import org.apache.qpid.proton.reactor.Selectable.Callback;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class AcceptorImplTest {
+
+    /**
+     * Tests that if ServerSocketChannel.accept() throws an IOException the Acceptor will
+     * call Selectable.error() on it's underlying selector.
+     * @throws IOException
+     */
+    @Test
+    public void acceptThrowsException() throws IOException {
+        final Callback mockCallback = Mockito.mock(Callback.class);
+        final SelectableImpl selectable = new SelectableImpl();
+        selectable.onError(mockCallback);
+        ReactorImpl mockReactor = Mockito.mock(ReactorImpl.class);
+        Mockito.when(mockReactor.selectable(Mockito.any(ReactorChild.class))).thenReturn(selectable);
+        class MockAcceptorImpl extends AcceptorImpl {
+
+            protected MockAcceptorImpl(Reactor reactor, String host, int port, Handler handler)
throws IOException {
+                super(reactor, host, port, handler);
+            }
+
+            @Override
+            protected ServerSocketChannel openServerSocket() throws IOException {
+                ServerSocketChannel result = Mockito.mock(ServerSocketChannel.class);
+                Mockito.when(result.accept()).thenThrow(new IOException());
+                return result;
+            }
+        }
+        new MockAcceptorImpl(mockReactor, "host", 1234, null);
+        selectable.readable();
+        Mockito.verify(mockCallback).run(selectable);
+    }
+
+    /**
+     * Tests that if ServerSocketChannel.accept() returns <code>null</code> the
Acceptor will
+     * throw a ReactorInternalException (because the acceptor's underlying selectable should
+     * not have been marked as readable, if there is no connection to accept).
+     * @throws IOException
+     */
+    @Test(expected=ReactorInternalException.class)
+    public void acceptReturnsNull() throws IOException {
+        final Callback mockCallback = Mockito.mock(Callback.class);
+        final SelectableImpl selectable = new SelectableImpl();
+        selectable.onError(mockCallback);
+        ReactorImpl mockReactor = Mockito.mock(ReactorImpl.class);
+        Mockito.when(mockReactor.selectable(Mockito.any(ReactorChild.class))).thenReturn(selectable);
+        class MockAcceptorImpl extends AcceptorImpl {
+
+            protected MockAcceptorImpl(Reactor reactor, String host, int port, Handler handler)
throws IOException {
+                super(reactor, host, port, handler);
+            }
+
+            @Override
+            protected ServerSocketChannel openServerSocket() throws IOException {
+                ServerSocketChannel result = Mockito.mock(ServerSocketChannel.class);
+                Mockito.when(result.accept()).thenReturn(null);
+                return result;
+            }
+        }
+        new MockAcceptorImpl(mockReactor, "host", 1234, null);
+        selectable.readable();
+    }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message