qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r..@apache.org
Subject [06/19] qpid-proton git commit: PROTON-881: Add Recv sample and required code changes / additions to the reactor
Date Thu, 07 May 2015 08:31:37 GMT
PROTON-881: Add Recv sample and required code changes / additions to the reactor


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

Branch: refs/heads/proton-j-reactor
Commit: 0ac98e76be51fbd890613518541364301b191cc2
Parents: cd09de6
Author: Adrian Preston <prestona@uk.ibm.com>
Authored: Wed Apr 22 21:35:36 2015 +0100
Committer: Adrian Preston <prestona@uk.ibm.com>
Committed: Wed May 6 23:23:55 2015 +0100

----------------------------------------------------------------------
 .../qpid/proton/example/reactor/Recv.java       |  79 +++++++++++++
 .../apache/qpid/proton/reactor/Acceptor.java    |  27 +++++
 .../qpid/proton/reactor/FlowController.java     |  72 ++++++++++++
 .../org/apache/qpid/proton/reactor/Reactor.java |   3 +
 .../qpid/proton/reactor/impl/AcceptorImpl.java  | 112 +++++++++++++++++++
 .../qpid/proton/reactor/impl/IOHandler.java     |  25 +++--
 .../qpid/proton/reactor/impl/ReactorImpl.java   |  12 +-
 .../qpid/proton/reactor/impl/SelectorImpl.java  |  10 +-
 8 files changed, 326 insertions(+), 14 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/examples/java/reactor/src/main/java/org/apache/qpid/proton/example/reactor/Recv.java
----------------------------------------------------------------------
diff --git a/examples/java/reactor/src/main/java/org/apache/qpid/proton/example/reactor/Recv.java
b/examples/java/reactor/src/main/java/org/apache/qpid/proton/example/reactor/Recv.java
new file mode 100644
index 0000000..a5c6291
--- /dev/null
+++ b/examples/java/reactor/src/main/java/org/apache/qpid/proton/example/reactor/Recv.java
@@ -0,0 +1,79 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.proton.example.reactor;
+
+import java.io.IOException;
+
+import org.apache.qpid.proton.Proton;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
+import org.apache.qpid.proton.engine.BaseHandler;
+import org.apache.qpid.proton.engine.Delivery;
+import org.apache.qpid.proton.engine.Event;
+import org.apache.qpid.proton.engine.Receiver;
+import org.apache.qpid.proton.message.Message;
+import org.apache.qpid.proton.reactor.FlowController;
+import org.apache.qpid.proton.reactor.Handshaker;
+import org.apache.qpid.proton.reactor.Reactor;
+
+public class Recv extends BaseHandler {
+
+    private Recv() {
+        add(new Handshaker());
+        add(new FlowController());
+    }
+
+    @Override
+    public void onReactorInit(Event event) {
+        try {
+            // Create an amqp acceptor.
+            event.getReactor().acceptor("0.0.0.0", 5672);
+
+            // There is an optional third argument to the Reactor.acceptor
+            // call. Using it, we could supply a handler here that would
+            // become the handler for all accepted connections. If we omit
+            // it, the reactor simply inherets all the connection events.
+        } catch(IOException ioException) {
+            ioException.printStackTrace();    // TODO: what is the right answer?
+        }
+    }
+
+    @Override
+    public void onDelivery(Event event) {
+        Receiver recv = (Receiver)event.getLink();
+        Delivery delivery = recv.current();
+        if (delivery.isReadable() && !delivery.isPartial()) {
+            int size = delivery.pending();
+            byte[] buffer = new byte[size];
+            int read = recv.recv(buffer, 0, buffer.length);
+            recv.advance();
+
+            Message msg = Proton.message();
+            msg.decode(buffer, 0, read);
+            System.out.println(((AmqpValue)msg.getBody()).getValue());
+        }
+    }
+
+    public static void main(String[] args) throws IOException {
+        Reactor r = Proton.reactor(new Recv());
+        r.run();
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/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
new file mode 100644
index 0000000..91b20a3
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/Acceptor.java
@@ -0,0 +1,27 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.proton.reactor;
+
+public interface Acceptor {
+
+    void close();
+}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/proton-j/src/main/java/org/apache/qpid/proton/reactor/FlowController.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/FlowController.java b/proton-j/src/main/java/org/apache/qpid/proton/reactor/FlowController.java
new file mode 100644
index 0000000..c8b999b
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/FlowController.java
@@ -0,0 +1,72 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.proton.reactor;
+
+import org.apache.qpid.proton.engine.BaseHandler;
+import org.apache.qpid.proton.engine.Event;
+import org.apache.qpid.proton.engine.Link;
+import org.apache.qpid.proton.engine.Receiver;
+
+public class FlowController extends BaseHandler {
+
+    private int drained;
+    private int window;
+
+    public FlowController(int window) {
+        // XXX: a window of 1 doesn't work because we won't necessarily get
+        // notified when the one allowed delivery is settled
+        if (window <= 1) throw new IllegalArgumentException();
+        this.window = window;
+        this.drained = 0;
+    }
+
+    public FlowController() {
+        this(1024);
+    }
+
+    private void topup(Receiver link, int window) {
+        int delta = window - link.getCredit();
+        link.flow(delta);
+    }
+
+    @Override
+    public void onUnhandled(Event event) {
+        int window = this.window;
+        Link link = event.getLink();
+
+        switch(event.getType()) {
+        case LINK_LOCAL_OPEN:
+        case LINK_REMOTE_OPEN:
+        case LINK_FLOW:
+        case DELIVERY:
+            if (link instanceof Receiver) {
+                this.drained += link.drained();
+                if (this.drained == 0) {
+                    topup((Receiver)link, window);
+                }
+            }
+            break;
+        default:
+            break;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/proton-j/src/main/java/org/apache/qpid/proton/reactor/Reactor.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/Reactor.java b/proton-j/src/main/java/org/apache/qpid/proton/reactor/Reactor.java
index 0c56a48..68375b1 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/reactor/Reactor.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/Reactor.java
@@ -97,4 +97,7 @@ public interface Reactor {
     // TODO: acceptorClose
 
     Connection connection(Handler handler);
+
+    Acceptor acceptor(String host, int port) throws IOException;
+    Acceptor acceptor(String host, int port, Handler handler) throws IOException;
 }

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/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
new file mode 100644
index 0000000..efe0eb4
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/AcceptorImpl.java
@@ -0,0 +1,112 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.proton.reactor.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+
+import org.apache.qpid.proton.Proton;
+import org.apache.qpid.proton.engine.Connection;
+import org.apache.qpid.proton.engine.Handler;
+import org.apache.qpid.proton.engine.Sasl;
+import org.apache.qpid.proton.engine.Sasl.SaslOutcome;
+import org.apache.qpid.proton.engine.Transport;
+import org.apache.qpid.proton.reactor.Acceptor;
+import org.apache.qpid.proton.reactor.Reactor;
+import org.apache.qpid.proton.reactor.Selectable;
+import org.apache.qpid.proton.reactor.Selectable.Callback;
+
+public class AcceptorImpl implements Acceptor {
+
+    private class AcceptorReadable implements Callback {
+        @Override
+        public void run(Selectable selectable) {
+            Reactor reactor = selectable.getReactor();
+            try {
+                SocketChannel socketChannel = ((ServerSocketChannel)selectable.getChannel()).accept();
+                Handler handler = (Handler)selectable.getAttachment();
+                if (handler == null) {
+                    // TODO: set selectable.getAttachment() to null?
+                    handler = reactor.getHandler();
+                }
+                Connection conn = reactor.connection(handler);
+                Transport trans = Proton.transport();
+                // TODO: the C code calls pn_transport_set_server(trans) - is there a Java
equivalent we need to worry about?
+                Sasl sasl = trans.sasl();
+                sasl.server();  // TODO: it would be nice if SASL was more pluggable than
this (but this is what the C API currently does...)
+                //sasl.allowSkip(true); // TODO: this in in the C code - but the proton-j
code throws a ProtonUnsupportedOperationException (as it is not implemented)
+                sasl.setMechanisms("ANONYMOUS");
+                sasl.done(SaslOutcome.PN_SASL_OK);
+                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?
+            }
+        }
+    }
+
+    private class AcceptorFinalize implements Callback {
+        @Override
+        public void run(Selectable selectable) {
+            try {
+                selectable.getChannel().close();
+            } catch(IOException ioException) {
+                ioException.printStackTrace();
+                // TODO: what now?
+            }
+        }
+    }
+
+    private final Selectable sel;
+
+    protected AcceptorImpl(Reactor reactor, String host, int port, Handler handler) throws
IOException {
+        ServerSocketChannel ssc = ServerSocketChannel.open();
+        ssc.bind(new InetSocketAddress(host, port));
+        sel = reactor.selectable();
+        sel.setChannel(ssc);
+        sel.onReadable(new AcceptorReadable());
+        sel.onFinalize(new AcceptorFinalize()); // TODO: currently, this is not called from
anywhere!!
+        sel.setReactor(reactor);
+        sel.setAttachment(handler);
+        sel.setReading(true);
+        reactor.update(sel);
+    }
+
+    @Override
+    public void close() {
+        if (!sel.isTerminal()) {
+            Reactor reactor = sel.getReactor();
+            try {
+                sel.getChannel().close();
+            } catch(IOException ioException) {
+                ioException.printStackTrace();
+                // TODO: what now?
+            }
+            sel.setChannel(null);
+            sel.terminate();
+            reactor.update(sel);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/IOHandler.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/IOHandler.java b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/IOHandler.java
index ee988ee..a182722 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/IOHandler.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/IOHandler.java
@@ -78,6 +78,7 @@ public class IOHandler extends BaseHandler {
         }
         Transport transport = Proton.transport();
         Sasl sasl = transport.sasl();
+        sasl.client();
         sasl.setMechanisms("ANONYMOUS");
         transport.bind(connection);
     }
@@ -86,7 +87,7 @@ public class IOHandler extends BaseHandler {
     private void handleBound(Reactor reactor, Event event) {
         Connection connection = event.getConnection();
         String hostname = connection.getHostname();
-        if (hostname == null) {
+        if (hostname == null || hostname.equals("")) {
             return;
         }
 
@@ -116,7 +117,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_capacity from connection.c
-    private int capacity(Selectable selectable) {
+    private static int capacity(Selectable selectable) {
         Transport transport = selectable.getTransport();
         int capacity = transport.capacity();
         if (capacity < 0) {
@@ -128,7 +129,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_pending from connection.c
-    private int pending(Selectable selectable) {
+    private static int pending(Selectable selectable) {
         Transport transport = selectable.getTransport();
         int pending = transport.pending();
         if (pending < 0) {
@@ -140,7 +141,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_deadline from connection.c
-    private long deadline(Selectable selectable) {
+    private static long deadline(Selectable selectable) {
         Reactor reactor = selectable.getReactor();
         Transport transport = selectable.getTransport();
         long deadline = transport.tick(reactor.now());
@@ -148,7 +149,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_update from connection.c
-    private void update(Selectable selectable) {
+    private static void update(Selectable selectable) {
         int c = capacity(selectable);
         int p = pending(selectable);
         selectable.setReading(c > 0);
@@ -157,7 +158,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_readable from connection.c
-    private class ConnectionReadable implements Callback {
+    private static class ConnectionReadable implements Callback {
         @Override
         public void run(Selectable selectable) {
             Reactor reactor = selectable.getReactor();
@@ -191,7 +192,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_writable from connection.c
-    private class ConnectionWritable implements Callback {
+    private static class ConnectionWritable implements Callback {
         @Override
         public void run(Selectable selectable) {
             Reactor reactor = selectable.getReactor();
@@ -224,7 +225,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_error from connection.c
-    private class ConnectionError implements Callback {
+    private static class ConnectionError implements Callback {
         @Override
         public void run(Selectable selectable) {
             Reactor reactor = selectable.getReactor();
@@ -235,7 +236,7 @@ public class IOHandler extends BaseHandler {
     }
 
     // pni_connection_expired from connection.c
-    private class ConnectionExpired implements Callback {
+    private static class ConnectionExpired implements Callback {
         @Override
         public void run(Selectable selectable) {
             Reactor reactor = selectable.getReactor();
@@ -250,7 +251,7 @@ public class IOHandler extends BaseHandler {
         }
     }
 
-    private class ConnectionFinalize implements Callback {
+    private static class ConnectionFinalize implements Callback {
         @Override
         public void run(Selectable selectable) {
             try {
@@ -263,11 +264,11 @@ public class IOHandler extends BaseHandler {
     }
 
     // pn_reactor_selectable_transport
-    private Selectable selectableTransport(Reactor reactor, Socket socket, Transport transport)
{
+    protected static Selectable selectableTransport(Reactor reactor, Socket socket, Transport
transport) {
         // TODO: this code needs to be able to deal with a null socket (this is our equivalent
of PN_INVALID_SOCKET)
         Selectable selectable = reactor.selectable();
         selectable.setChannel(socket != null ? socket.getChannel() : null);
-        selectable.onReadable(new ConnectionReadable());
+        selectable.onReadable(new ConnectionReadable());    // TODO: *IF* these callbacks
are stateless, do we more than one instance of them?
         selectable.onWritable(new ConnectionWritable());
         selectable.onError(new ConnectionError());
         selectable.onExpired(new ConnectionExpired());

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/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 0a7f84d..9c4f817 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
@@ -37,6 +37,7 @@ import org.apache.qpid.proton.engine.Handler;
 import org.apache.qpid.proton.engine.impl.CollectorImpl;
 import org.apache.qpid.proton.engine.impl.ConnectionImpl;
 import org.apache.qpid.proton.engine.impl.HandlerEndpointImpl;
+import org.apache.qpid.proton.reactor.Acceptor;
 import org.apache.qpid.proton.reactor.Reactor;
 import org.apache.qpid.proton.reactor.ReactorChild;
 import org.apache.qpid.proton.reactor.Selectable;
@@ -454,7 +455,16 @@ public class ReactorImpl implements Reactor {
         connection.collect(collector);
         children.add(connection);
         ((ConnectionImpl)connection).setReactor(this);
-        // TODO: C code adds a reference back to the reactor from connection
         return connection;
     }
+
+    @Override
+    public Acceptor acceptor(String host, int port) throws IOException {
+        return this.acceptor(host, port, null);
+    }
+
+    @Override
+    public Acceptor acceptor(String host, int port, Handler handler) throws IOException {
+        return new AcceptorImpl(this, host, port, handler);
+    }
 }

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/0ac98e76/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 35a6555..e4e284e 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
@@ -23,6 +23,7 @@ package org.apache.qpid.proton.reactor.impl;
 
 import java.io.IOException;
 import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
 import java.util.HashSet;
 import java.util.Iterator;
 
@@ -58,7 +59,13 @@ public class SelectorImpl implements Selector {
     public void update(Selectable selectable) {
         if (selectable.getChannel() != null) {
             int interestedOps = 0;
-            if (selectable.isReading()) interestedOps |= SelectionKey.OP_READ;
+            if (selectable.isReading()) {
+                if (selectable.getChannel() instanceof ServerSocketChannel) {
+                    interestedOps |= SelectionKey.OP_ACCEPT;
+                } else {
+                    interestedOps |= SelectionKey.OP_READ;
+                }
+            }
             if (selectable.isWriting()) interestedOps |= SelectionKey.OP_WRITE;
             SelectionKey key = selectable.getChannel().keyFor(selector);
             key.interestOps(interestedOps);
@@ -111,6 +118,7 @@ public class SelectorImpl implements Selector {
         for (SelectionKey key : selector.selectedKeys()) {
             Selectable selectable = (Selectable)key.attachment();
             if (key.isReadable()) readable.add(selectable);
+            if (key.isAcceptable()) readable.add(selectable);
             if (key.isWritable()) writeable.add(selectable);
         }
         selector.selectedKeys().clear();


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


Mime
View raw message