You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2015/07/06 01:45:13 UTC
[12/38] qpid-proton git commit: PROTON-881: Add error handling for
"accept socket connection" path through Acceptor
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 <pr...@uk.ibm.com>
Authored: Sat May 2 22:30:40 2015 +0100
Committer: Adrian Preston <pr...@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