qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From raj...@apache.org
Subject qpid-proton git commit: More workarounds to allow testing of both codecs side by side.
Date Mon, 05 Oct 2015 22:03:39 GMT
Repository: qpid-proton
Updated Branches:
  refs/heads/rajith-codec d2ca43c4e -> 171dba029


More workarounds to allow testing of both codecs side by side.


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

Branch: refs/heads/rajith-codec
Commit: 171dba0292db890c2c2dd153ddae09a3ab071698
Parents: d2ca43c
Author: Rajith Attapattu <rajith@apache.org>
Authored: Mon Oct 5 18:02:49 2015 -0400
Committer: Rajith Attapattu <rajith@apache.org>
Committed: Mon Oct 5 18:02:49 2015 -0400

----------------------------------------------------------------------
 .../apache/qpid/proton/BenchmarkReceiver.java   |   63 +-
 .../org/apache/qpid/proton/BenchmarkSender.java |  156 +-
 .../org/apache/qpid/proton/TestReceiver.java    |   20 -
 .../java/org/apache/qpid/proton/TestSender.java |   23 -
 .../apache/qpid/proton/messenger/Messenger.java |    2 +-
 .../qpid/proton/messenger/Messenger2.java       |  278 ++++
 .../proton/messenger/impl/MessengerImpl.java    |    4 +-
 .../proton/messenger/impl/MessengerImpl2.java   | 1547 ++++++++++++++++++
 8 files changed, 1971 insertions(+), 122 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkReceiver.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkReceiver.java b/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkReceiver.java
index e3aaf62..46a89a9 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkReceiver.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkReceiver.java
@@ -1,21 +1,40 @@
 package org.apache.qpid.proton;
 
-import org.apache.qpid.proton.message2.Message;
-import org.apache.qpid.proton.messenger.Messenger;
-
 public class BenchmarkReceiver
-{   
-    public static void main(String[] args) throws Exception
+{
+    static void messengerOld(int count) throws Exception
     {
-        int count = Integer.getInteger("count", 1);
-        
-        Messenger rec = Proton.messenger();
+        org.apache.qpid.proton.messenger.Messenger rec = new org.apache.qpid.proton.messenger.impl.MessengerImpl();
+        rec.start();
+        rec.subscribe("amqp://~localhost:5672");
+        int i = 0;
+        org.apache.qpid.proton.message.Message m = null;
+        long start = 0;
+        while (i < count)
+        {
+            rec.recv();
+            while ((m = rec.get()) != null)
+            {
+                i++;
+                start = m.getCreationTime();
+            }
+        }
+        System.out.println("Count " + count + " i " + i);
+
+        double elapsed = (System.nanoTime() - start) / 1000000;
+        System.out.println(String.format("Old codec Time elapsed (end-to-end) for %s msgs : %s (milli secs)", count, elapsed));
+    }
+
+    
+    static void messengerNew(int count)  throws Exception
+    {
+        org.apache.qpid.proton.messenger.Messenger2 rec = new org.apache.qpid.proton.messenger.impl.MessengerImpl2();
         rec.start();
         rec.subscribe("amqp://~localhost:5672");
         int i = 0;
-        Message m = null;
+        org.apache.qpid.proton.message2.Message m = null;
         long start = 0;
-        while (i <count)
+        while (i < count)
         {
             rec.recv();
             while ((m = rec.get()) != null)
@@ -23,10 +42,24 @@ public class BenchmarkReceiver
                 i++;
                 start = m.getCreationTime();
             }
-        }    
-        System.out.println("Count " + count + " i "  + i);
-        
-        double elapsed = (System.nanoTime() - start)/1000000; 
-        System.out.println(String.format("Time elapsed (end-to-end) for %s msgs : %s (milli secs)", count, elapsed));
+        }
+
+        double elapsed = (System.nanoTime() - start) / 1000000;
+        System.out.println(String.format("New codec Time elapsed (end-to-end) for %s msgs : %s (milli secs)", count, elapsed));
+    }
+ 
+    public static void main(String[] args) throws Exception
+    {
+        int count = Integer.getInteger("count", 1);
+        boolean isNewCodec = Boolean.getBoolean("new-codec");
+
+        if (isNewCodec)
+        {
+            messengerNew(count);
+        }
+        else
+        {
+            messengerOld(count);
+        }
     }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkSender.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkSender.java b/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkSender.java
index a118c2d..a01296d 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkSender.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/BenchmarkSender.java
@@ -6,103 +6,137 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.qpid.proton.message2.Message;
-import org.apache.qpid.proton.messenger.Messenger;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
 
 public class BenchmarkSender
 {
     static String largeString = "Knowing that millions of people around the world would be watching in person and on television and expecting great";
+
     static String smallString = "Hello world";
+
     static long n = Long.MAX_VALUE;
 
-    public static Message getMessage(int op)
+    enum OP {STR_SMALL, STR_BIG, LONG, LIST_SMALL, LIST_BIG, MAP_SMALL, MAP_BIG, NESTED_SMALL, NESTED_BIG};
+ 
+    public static Object getPayload(int op)
     {
-        Message m = Proton.message2();
-        m.setAddress("amqp://localhost:5672");
         switch (op)
         {
-        case 0 :
-            m.setBody(smallString);
-            break;
-        case 1 :
-            m.setBody(largeString);
-            break;
-        case 2 :
-            m.setBody(n);
-            break;
-        case 3 :
-           List<String> listSm = new ArrayList<String>();
-           for (int i=0; i<5; i++)
-           {
-               listSm.add(smallString);
-           } 
-           m.setBody(listSm);
-           break;
-        case 4 :
+        case 0:
+            return smallString;
+        case 1:
+            return largeString;
+        case 2:
+            return n;
+        case 3:
+            List<String> listSm = new ArrayList<String>();
+            for (int i = 0; i < 5; i++)
+            {
+                listSm.add(smallString);
+            }
+            return listSm;
+        case 4:
             List<String> listBig = new ArrayList<String>();
-            for (int i=0; i<100; i++)
+            for (int i = 0; i < 100; i++)
             {
                 listBig.add(smallString);
-            } 
-            m.setBody(listBig);
-            break;
+            }
+            return listBig;
         case 5:
-          Map<String, Integer> mapSm = new HashMap<String, Integer>(); 
-          for (int i=0; i<5; i++)
-          {
-              mapSm.put(String.valueOf(i), i);
-          }  
-          m.setBody(mapSm);
-          break;
+            Map<String, Integer> mapSm = new HashMap<String, Integer>();
+            for (int i = 0; i < 5; i++)
+            {
+                mapSm.put(String.valueOf(i), i);
+            }
+            return mapSm;
         case 6:
-            Map<String, Integer> mapBig = new HashMap<String, Integer>(); 
-            for (int i=0; i<100; i++)
+            Map<String, Integer> mapBig = new HashMap<String, Integer>();
+            for (int i = 0; i < 100; i++)
             {
                 mapBig.put(String.valueOf(i), i);
-            }  
-            m.setBody(mapBig);
+            }
+            return mapBig;
         case 7:
-            List<Integer> l1 = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
-            Map<String, List<Integer>> nestedMapSm = new HashMap<String, List<Integer>>(); 
-            for (int i=0; i<5; i++)
+            List<Integer> l1 = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+            Map<String, List<Integer>> nestedMapSm = new HashMap<String, List<Integer>>();
+            for (int i = 0; i < 5; i++)
             {
                 nestedMapSm.put(String.valueOf(i), l1);
             }
-            m.setBody(nestedMapSm);
-            break;
+            return nestedMapSm;
         case 8:
-            List<Integer> l2 = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
-            Map<String, List<Integer>> nestedMapBig = new HashMap<String, List<Integer>>(); 
-            for (int i=0; i<100; i++)
+            List<Integer> l2 = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+            Map<String, List<Integer>> nestedMapBig = new HashMap<String, List<Integer>>();
+            for (int i = 0; i < 100; i++)
             {
                 nestedMapBig.put(String.valueOf(i), l2);
             }
-            m.setBody(nestedMapBig);
-            break;
+            return nestedMapBig;
+        default:
+            return smallString;
         }
-        return m;
     }
-    
-    public static void main(String[] args) throws Exception
+
+    static void messengerOld(int count, int op) throws Exception
     {
-        int count = Integer.getInteger("count", 1);
-        int op = Integer.getInteger("op", 4);
-        
-        Message msg = getMessage(op);
-        Messenger sender = Proton.messenger();
+        Object obj = getPayload(op);
+        org.apache.qpid.proton.message.Message msg = Proton.message();
+        msg.setAddress("amqp://localhost:5672");
+        msg.setBody(new AmqpValue(obj));
+        org.apache.qpid.proton.messenger.Messenger sender = new org.apache.qpid.proton.messenger.impl.MessengerImpl();
+        sender.start();
+
+        long start = System.nanoTime();
+        for (int i = 0; i < count; i++)
+        {
+            msg.setCreationTime(start);
+            sender.put(msg);
+        }
+        sender.send();
+
+        double elapsed = (System.nanoTime() - start) / 1000000;
+        System.out.println(String.format("Old codec Time elapsed for sending %s msgs for op [%s] : %s (milli secs)", count, OP.values()[op],
+                elapsed));
+    }
+
+    static void messengerNew(int count, int op) throws Exception
+    {
+        Object obj = getPayload(op);
+        org.apache.qpid.proton.message2.Message msg = Proton.message2();
+        msg.setAddress("amqp://localhost:5672");
+        msg.setBody(obj);
+        org.apache.qpid.proton.messenger.Messenger2 sender = new org.apache.qpid.proton.messenger.impl.MessengerImpl2();
         sender.start();
-        
+
         long start = System.nanoTime();
-        for (int i=0; i<count; i++)
+        for (int i = 0; i < count; i++)
         {
             msg.setCreationTime(start);
             sender.put(msg);
         }
         sender.send();
 
-        double elapsed = (System.nanoTime() - start)/1000000; 
-        System.out.println(String.format("Time elapsed for sending %s msgs for op [%s] : %s (milli secs)", count, op, elapsed));
-        
+        double elapsed = (System.nanoTime() - start) / 1000000;
+        System.out.println(String.format("New Codec Time elapsed for sending %s msgs for op [%s] : %s (milli secs)", count, op,
+                elapsed));
+    }
+
+    public static void main(String[] args) throws Exception
+    {
+        int count = Integer.getInteger("count", 1);
+        int op = Integer.getInteger("op", 1);
+
+        boolean isNewCodec = Boolean.getBoolean("new-codec");
+
+        if (isNewCodec)
+        {
+            messengerNew(count, op);
+        }
+        else
+        {
+            messengerOld(count, op);
+        }
+
         final Object ob = new Object();
         synchronized (ob)
         {

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/TestReceiver.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/TestReceiver.java b/proton-j/src/main/java/org/apache/qpid/proton/TestReceiver.java
deleted file mode 100644
index 4143caf..0000000
--- a/proton-j/src/main/java/org/apache/qpid/proton/TestReceiver.java
+++ /dev/null
@@ -1,20 +0,0 @@
-package org.apache.qpid.proton;
-import java.io.IOException;
-
-import org.apache.qpid.proton.message2.Message;
-import org.apache.qpid.proton.messenger.Messenger;
-
-
-public class TestReceiver
-{
-    public static void main(String[] args) throws IOException
-    {
-        Messenger messenger = Proton.messenger();
-        messenger.start();
-        messenger.subscribe("amqp://~localhost:5672");
-        messenger.recv(); 
-        Message m = messenger.get();
-        System.out.println("Received " + m);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/TestSender.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/TestSender.java b/proton-j/src/main/java/org/apache/qpid/proton/TestSender.java
deleted file mode 100644
index f522aa8..0000000
--- a/proton-j/src/main/java/org/apache/qpid/proton/TestSender.java
+++ /dev/null
@@ -1,23 +0,0 @@
-package org.apache.qpid.proton;
-import java.io.IOException;
-
-import org.apache.qpid.proton.message2.AmqpValue;
-import org.apache.qpid.proton.message2.Message;
-import org.apache.qpid.proton.messenger.Messenger;
-
-public class TestSender
-{
-    public static void main(String[] args) throws IOException, InterruptedException
-    {
-        Messenger messenger = Proton.messenger();
-        messenger.start();
-        for (int i=0; i < 1; i++)
-        {
-            Message m = Proton.message2();
-            m.setAddress("amqp://localhost:5672");
-            m.setBody(new AmqpValue("Msg" + i));
-            messenger.put(m);
-        }
-        messenger.send();
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger.java b/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger.java
index cd6d93d..6d3f362 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger.java
@@ -23,7 +23,7 @@ package org.apache.qpid.proton.messenger;
 import java.io.IOException;
 
 import org.apache.qpid.proton.TimeoutException;
-import org.apache.qpid.proton.message2.Message;
+import org.apache.qpid.proton.message.Message;
 
 import org.apache.qpid.proton.messenger.impl.MessengerImpl;
 

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger2.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger2.java b/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger2.java
new file mode 100644
index 0000000..47ab412
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/messenger/Messenger2.java
@@ -0,0 +1,278 @@
+/*
+ *
+ * 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.messenger;
+
+import java.io.IOException;
+
+import org.apache.qpid.proton.TimeoutException;
+import org.apache.qpid.proton.message2.Message;
+import org.apache.qpid.proton.messenger.impl.MessengerImpl2;
+
+/**
+ *
+ *  Messenger defines a high level interface for sending and receiving
+ *  messages. Every Messenger contains a single logical queue of
+ *  incoming messages and a single logical queue of outgoing
+ *  messages. These messages in these queues may be destined for, or
+ *  originate from, a variety of addresses.
+ *
+ *  <h3>Address Syntax</h3>
+ *
+ *  An address has the following form:
+ *
+ *    [ amqp[s]:// ] [user[:password]@] domain [/[name]]
+ *
+ *  Where domain can be one of:
+ *
+ *    host | host:port | ip | ip:port | name
+ *
+ *  The following are valid examples of addresses:
+ *
+ *   - example.org
+ *   - example.org:1234
+ *   - amqp://example.org
+ *   - amqps://example.org
+ *   - example.org/incoming
+ *   - amqps://example.org/outgoing
+ *   - amqps://fred:trustno1@example.org
+ *   - 127.0.0.1:1234
+ *   - amqps://127.0.0.1:1234
+ *
+ *  <h3>Sending &amp; Receiving Messages</h3>
+ *
+ *  The Messenger interface works in conjuction with the Message
+ *  class. The Message class is a mutable holder of message content.
+ *  The put method will encode the content in a given Message object
+ *  into the outgoing message queue leaving that Message object free
+ *  to be modified or discarded without having any impact on the
+ *  content in the outgoing queue.
+ *
+ *  Similarly, the get method will decode the content in the incoming
+ *  message queue into the supplied Message object.
+*/
+public interface Messenger2
+{
+
+    public static final class Factory
+    {
+        public static Messenger2 create() {
+            return new MessengerImpl2();
+        }
+
+        public static Messenger2 create(String name) {
+            return new MessengerImpl2(name);
+        }
+    }
+
+    /**
+     * Flag for use with reject(), accept() and settle() methods.
+     */
+    static final int CUMULATIVE = 0x01;
+
+    /**
+     * Places the content contained in the message onto the outgoing
+     * queue of the Messenger. This method will never block. The
+     * send call may be used to block until the messages are
+     * sent. Either a send() or a recv() call is neceesary at present
+     * to cause the messages to actually be sent out.
+     */
+    void put(Message message) throws MessengerException;
+
+    /**
+     * Blocks until the outgoing queue is empty and, in the event that
+     * an outgoing window has been set, until the messages in that
+     * window have been received by the target to which they were
+     * sent, or the operation times out. The timeout property
+     * controls how long a Messenger will block before timing out.
+     */
+    void send() throws TimeoutException;
+
+    void send(int n) throws TimeoutException;
+
+    /**
+     * Subscribes the Messenger to messages originating from the
+     * specified source. The source is an address as specified in the
+     * Messenger introduction with the following addition. If the
+     * domain portion of the address begins with the '~' character,
+     * the Messenger will interpret the domain as host/port, bind
+     * to it, and listen for incoming messages. For example
+     * "~0.0.0.0", "amqp://~0.0.0.0" will bind to any local interface
+     * and listen for incoming messages.
+     */
+    void subscribe(String source) throws MessengerException;
+    /**
+     * Receives an arbitrary number of messages into the
+     * incoming queue of the Messenger. This method will block until
+     * at least one message is available or the operation times out.
+     */
+    void recv() throws TimeoutException;
+    /**
+     * Receives up to the specified number of messages into the
+     * incoming queue of the Messenger. This method will block until
+     * at least one message is available or the operation times out.
+     */
+    void recv(int count) throws TimeoutException;
+    /**
+     * Returns the capacity of the incoming message queue of
+     * messenger. Note this count does not include those messages
+     * already available on the incoming queue (see
+     * incoming()). Rather it returns the number of incoming queue
+     * entries available for receiving messages
+     */
+    int receiving();
+    /**
+     * Returns the message from the head of the incoming message
+     * queue.
+     */
+    Message get();
+
+    /**
+     * Transitions the Messenger to an active state. A Messenger is
+     * initially created in an inactive state. When inactive, a
+     * Messenger will not send or receive messages from its internal
+     * queues. A Messenger must be started before calling send() or
+     * recv().
+     */
+    void start() throws IOException;
+    /**
+     * Transitions the Messenger to an inactive state. An inactive
+     * Messenger will not send or receive messages from its internal
+     * queues. A Messenger should be stopped before being discarded to
+     * ensure a clean shutdown handshake occurs on any internally managed
+     * connections.
+     */
+    void stop();
+
+    boolean stopped();
+
+    /** Sends or receives any outstanding messages queued for a
+     * messenger.  If timeout is zero, no blocking is done.  A timeout
+     * of -1 blocks forever, otherwise timeout is the maximum time (in
+     * millisecs) to block.  Returns True if work was performed.
+     */
+    boolean work(long timeout) throws TimeoutException;
+
+    void interrupt();
+
+    void setTimeout(long timeInMillis);
+    long getTimeout();
+
+    boolean isBlocking();
+    void setBlocking(boolean b);
+
+    /**
+     * Returns a count of the messages currently on the outgoing queue
+     * (i.e. those that have been put() but not yet actually sent
+     * out).
+     */
+    int outgoing();
+    /**
+     * Returns a count of the messages available on the incoming
+     * queue.
+     */
+    int incoming();
+
+    int getIncomingWindow();
+    void setIncomingWindow(int window);
+
+    int getOutgoingWindow();
+    void setOutgoingWindow(int window);
+
+    /**
+     * Returns a token which can be used to accept or reject the
+     * message returned in the previous get() call.
+     */
+    Tracker incomingTracker();
+    /**
+     * Returns a token which can be used to track the status of the
+     * message of the previous put() call.
+     */
+    Tracker outgoingTracker();
+
+    /**
+     * Rejects messages retrieved from the incoming message queue. The
+     * tracker object for a message is obtained through a call to
+     * incomingTracker() following a get(). If the flags argument
+     * contains CUMULATIVE, then all message up to the one identified
+     * by the tracker will be rejected.
+     */
+    void reject(Tracker tracker, int flags);
+    /**
+     * Accepts messages retrieved from the incoming message queue. The
+     * tracker object for a message is obtained through a call to
+     * incomingTracker() following a get(). If the flags argument
+     * contains CUMULATIVE, then all message up to the one identified
+     * by the tracker will be accepted.
+     */
+    void accept(Tracker tracker, int flags);
+    void settle(Tracker tracker, int flags);
+
+    /**
+     * Gets the last known remote state of the delivery associated
+     * with the given tracker.
+     */
+    Status getStatus(Tracker tracker);
+
+    void route(String pattern, String address);
+
+    void rewrite(String pattern, String address);
+
+    /**
+     * Set the path to the certificate file.
+     */
+    void setCertificate(String certificate);
+
+    /**
+     * Get the path to the certificate file.
+     */
+    String getCertificate();
+
+    /**
+     * Set the path to private key file.
+     */
+    void setPrivateKey(String privateKey);
+
+    /**
+     * Get the path to the private key file.
+     */
+    String getPrivateKey();
+
+    /**
+     * Set the password for private key file.
+     */
+    void setPassword(String password);
+
+    /**
+     * Get the password for the priate key file.
+     */
+    String getPassword();
+
+    /**
+     * Set the path to the trusted certificate database.
+     */
+    void setTrustedCertificates(String trusted);
+
+    /**
+     * Get the path to the trusted certificate database.
+     */
+    String getTrustedCertificates();
+
+}

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java
index f0d7119..e6475b9 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java
@@ -45,7 +45,7 @@ import org.apache.qpid.proton.engine.Session;
 import org.apache.qpid.proton.engine.SslDomain;
 import org.apache.qpid.proton.engine.Ssl;
 import org.apache.qpid.proton.engine.Transport;
-import org.apache.qpid.proton.message2.Message;
+import org.apache.qpid.proton.message.Message;
 import org.apache.qpid.proton.messenger.Messenger;
 import org.apache.qpid.proton.messenger.MessengerException;
 import org.apache.qpid.proton.messenger.Status;
@@ -461,7 +461,7 @@ public class MessengerImpl implements Messenger
         StoreEntry entry = _incomingStore.get( null );
         if (entry != null)
         {
-            Message message = Proton.message2();
+            Message message = Proton.message();
             message.decode( entry.getEncodedMsg(), 0, entry.getEncodedLength() );
 
             _incomingTracker = new TrackerImpl(TrackerImpl.Type.INCOMING,

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/171dba02/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl2.java
----------------------------------------------------------------------
diff --git a/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl2.java b/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl2.java
new file mode 100644
index 0000000..ef861bf
--- /dev/null
+++ b/proton-j/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl2.java
@@ -0,0 +1,1547 @@
+/*
+ *
+ * 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.messenger.impl;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.qpid.proton.Proton;
+import org.apache.qpid.proton.InterruptException;
+import org.apache.qpid.proton.TimeoutException;
+import org.apache.qpid.proton.driver.Connector;
+import org.apache.qpid.proton.driver.Driver;
+import org.apache.qpid.proton.driver.Listener;
+import org.apache.qpid.proton.engine.Connection;
+import org.apache.qpid.proton.engine.Delivery;
+import org.apache.qpid.proton.engine.EndpointState;
+import org.apache.qpid.proton.engine.Link;
+import org.apache.qpid.proton.engine.Receiver;
+import org.apache.qpid.proton.engine.Sasl;
+import org.apache.qpid.proton.engine.Sender;
+import org.apache.qpid.proton.engine.Session;
+import org.apache.qpid.proton.engine.SslDomain;
+import org.apache.qpid.proton.engine.Ssl;
+import org.apache.qpid.proton.engine.Transport;
+import org.apache.qpid.proton.message2.Message;
+import org.apache.qpid.proton.messenger.Messenger2;
+import org.apache.qpid.proton.messenger.MessengerException;
+import org.apache.qpid.proton.messenger.Status;
+import org.apache.qpid.proton.messenger.Tracker;
+import org.apache.qpid.proton.amqp.messaging.Source;
+import org.apache.qpid.proton.amqp.messaging.Target;
+import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
+import org.apache.qpid.proton.amqp.transport.SenderSettleMode;
+
+public class MessengerImpl2 implements Messenger2
+{
+    private enum LinkCreditMode
+    {
+        // method for replenishing credit
+        LINK_CREDIT_EXPLICIT,   // recv(N)
+        LINK_CREDIT_AUTO;       // recv()
+    }
+
+    private static final EnumSet<EndpointState> UNINIT = EnumSet.of(EndpointState.UNINITIALIZED);
+    private static final EnumSet<EndpointState> ACTIVE = EnumSet.of(EndpointState.ACTIVE);
+    private static final EnumSet<EndpointState> CLOSED = EnumSet.of(EndpointState.CLOSED);
+    private static final EnumSet<EndpointState> ANY = EnumSet.allOf(EndpointState.class);
+
+    private final Logger _logger = Logger.getLogger("proton.messenger");
+    private final String _name;
+    private long _timeout = -1;
+    private boolean _blocking = true;
+    private long _nextTag = 1;
+    private Driver _driver;
+    private LinkCreditMode _credit_mode = LinkCreditMode.LINK_CREDIT_EXPLICIT;
+    private final int _credit_batch = 1024;   // credit_mode == LINK_CREDIT_AUTO
+    private int _credit;        // available
+    private int _distributed;    // outstanding credit
+    private int _receivers;      // total # receiver Links
+    private int _draining;       // # Links in drain state
+    private List<Receiver> _credited = new ArrayList<Receiver>();
+    private List<Receiver> _blocked = new ArrayList<Receiver>();
+    private long _next_drain;
+    private TrackerImpl _incomingTracker;
+    private TrackerImpl _outgoingTracker;
+    private Store _incomingStore = new Store();
+    private Store _outgoingStore = new Store();
+    private List<Connector> _awaitingDestruction = new ArrayList<Connector>();
+    private int _sendThreshold;
+
+    private Transform _routes = new Transform();
+    private Transform _rewrites = new Transform();
+
+    private String _certificate;
+    private String _privateKey;
+    private String _password;
+    private String _trustedDb;
+
+
+    /**
+     * @deprecated This constructor's visibility will be reduced to the default scope in a future release.
+     * Client code outside this module should use a {@link MessengerFactory} instead
+     */
+    @Deprecated public MessengerImpl2()
+    {
+        this(java.util.UUID.randomUUID().toString());
+    }
+
+    /**
+     * @deprecated This constructor's visibility will be reduced to the default scope in a future release.
+     * Client code outside this module should use a {@link MessengerFactory} instead
+     */
+    @Deprecated public MessengerImpl2(String name)
+    {
+        _name = name;
+    }
+
+    public void setTimeout(long timeInMillis)
+    {
+        _timeout = timeInMillis;
+    }
+
+    public long getTimeout()
+    {
+        return _timeout;
+    }
+
+    public boolean isBlocking()
+    {
+        return _blocking;
+    }
+
+    public void setBlocking(boolean b)
+    {
+        _blocking = b;
+    }
+
+    public void setCertificate(String certificate)
+    {
+        _certificate = certificate;
+    }
+
+    public String getCertificate()
+    {
+        return _certificate;
+    }
+
+    public void setPrivateKey(String privateKey)
+    {
+        _privateKey = privateKey;
+    }
+
+    public String getPrivateKey()
+    {
+        return _privateKey;
+    }
+
+    public void setPassword(String password)
+    {
+        _password = password;
+    }
+
+    public String getPassword()
+    {
+        return _password;
+    }
+
+    public void setTrustedCertificates(String trusted)
+    {
+        _trustedDb = trusted;
+    }
+
+    public String getTrustedCertificates()
+    {
+        return _trustedDb;
+    }
+
+    public void start() throws IOException
+    {
+        _driver = Proton.driver();
+    }
+
+    public void stop()
+    {
+        if (_driver != null) {
+            if(_logger.isLoggable(Level.FINE))
+            {
+                _logger.fine(this + " about to stop");
+            }
+            //close all connections
+            for (Connector<?> c : _driver.connectors())
+            {
+                Connection connection = c.getConnection();
+                connection.close();
+            }
+            //stop listeners
+            for (Listener<?> l : _driver.listeners())
+            {
+                try
+                {
+                    l.close();
+                }
+                catch (IOException e)
+                {
+                    _logger.log(Level.WARNING, "Error while closing listener", e);
+                }
+            }
+            waitUntil(_allClosed);
+        }
+    }
+
+    public boolean stopped()
+    {
+        return _allClosed.test();
+    }
+
+    public boolean work(long timeout) throws TimeoutException
+    {
+        if (_driver == null) { return false; }
+        _worked = false;
+        return waitUntil(_workPred, timeout);
+    }
+
+    public void interrupt()
+    {
+        if (_driver != null) {
+            _driver.wakeup();
+        }
+    }
+
+    private String defaultRewrite(String address) {
+        if (address != null && address.contains("@")) {
+            Address addr = new Address(address);
+            String scheme = addr.getScheme();
+            String host = addr.getHost();
+            String port = addr.getPort();
+            String name = addr.getName();
+
+            StringBuilder sb = new StringBuilder();
+            if (scheme != null) {
+                sb.append(scheme).append("://");
+            }
+            if (host != null) {
+                sb.append(host);
+            }
+            if (port != null) {
+                sb.append(":").append(port);
+            }
+            if (name != null) {
+                sb.append("/").append(name);
+            }
+            return sb.toString();
+        } else {
+            return address;
+        }
+    }
+
+
+    private String _original;
+
+    private void rewriteMessage(Message m)
+    {
+        _original = m.getAddress();
+        if (_rewrites.apply(_original)) {
+            m.setAddress(_rewrites.result());
+        } else {
+            m.setAddress(defaultRewrite(_original));
+        }
+    }
+
+    private void restoreMessage(Message m)
+    {
+        m.setAddress(_original);
+    }
+
+    private String routeAddress(String addr)
+    {
+        if (_routes.apply(addr)) {
+            return _routes.result();
+        } else {
+            return addr;
+        }
+    }
+
+    public void put(Message m) throws MessengerException
+    {
+        if (_driver == null) {
+            throw new IllegalStateException("cannot put while messenger is stopped");
+        }
+
+        if(_logger.isLoggable(Level.FINE))
+        {
+            _logger.fine(this + " about to put message: " + m);
+        }
+
+        StoreEntry entry = _outgoingStore.put( m.getAddress() );
+        _outgoingTracker = new TrackerImpl(TrackerImpl.Type.OUTGOING,
+                                           _outgoingStore.trackEntry(entry));
+
+        String routedAddress = routeAddress(m.getAddress());
+        Address address = new Address(routedAddress);
+        if (address.getHost() == null)
+        {
+            throw new MessengerException("unable to send to address: " + routedAddress);
+        }
+
+        rewriteMessage(m);
+
+        try {
+            adjustReplyTo(m);
+
+            int encoded;
+            byte[] buffer = new byte[5*1024];
+            while (true)
+            {
+                try
+                {
+                    encoded = m.encode(buffer, 0, buffer.length);
+                    break;
+                } catch (java.nio.BufferOverflowException e) {
+                    buffer = new byte[buffer.length*2];
+                }
+            }
+            entry.setEncodedMsg( buffer, encoded );
+        }
+        finally
+        {
+            restoreMessage(m);
+        }
+
+        Sender sender = getLink(address, new SenderFinder(address.getName()));
+        pumpOut(m.getAddress(), sender);
+    }
+
+    private void reclaimLink(Link link)
+    {
+        if (link instanceof Receiver)
+        {
+            int credit = link.getCredit();
+            if (credit > 0)
+            {
+                _credit += credit;
+                _distributed -= credit;
+            }
+        }
+
+        Delivery delivery = link.head();
+        while (delivery != null)
+        {
+            StoreEntry entry = (StoreEntry) delivery.getContext();
+            if (entry != null)
+            {
+                entry.setDelivery(null);
+                if (delivery.isBuffered()) {
+                    entry.setStatus(Status.ABORTED);
+                }
+            }
+            delivery = delivery.next();
+        }
+        linkRemoved(link);
+    }
+
+    private int pumpOut( String address, Sender sender )
+    {
+        StoreEntry entry = _outgoingStore.get( address );
+        if (entry == null) {
+            sender.drained();
+            return 0;
+        }
+
+        byte[] tag = String.valueOf(_nextTag++).getBytes();
+        Delivery delivery = sender.delivery(tag);
+        entry.setDelivery( delivery );
+        _logger.log(Level.FINE, "Sending on delivery: " + delivery);
+        int n = sender.send( entry.getEncodedMsg(), 0, entry.getEncodedLength());
+        if (n < 0) {
+            _outgoingStore.freeEntry( entry );
+            _logger.log(Level.WARNING, "Send error: " + n);
+            return n;
+        } else {
+            sender.advance();
+            _outgoingStore.freeEntry( entry );
+            return 0;
+        }
+    }
+
+    public void send() throws TimeoutException
+    {
+        send(-1);
+    }
+
+    public void send(int n) throws TimeoutException
+    {
+        if (_driver == null) {
+            throw new IllegalStateException("cannot send while messenger is stopped");
+        }
+
+        if(_logger.isLoggable(Level.FINE))
+        {
+            _logger.fine(this + " about to send");
+        }
+
+        if (n == -1)
+            _sendThreshold = 0;
+        else
+        {
+            _sendThreshold = outgoing() - n;
+            if (_sendThreshold < 0)
+                _sendThreshold = 0;
+        }
+
+        waitUntil(_sentSettled);
+    }
+
+    public void recv(int n) throws TimeoutException
+    {
+        if (_driver == null) {
+            throw new IllegalStateException("cannot recv while messenger is stopped");
+        }
+
+        if (_logger.isLoggable(Level.FINE) && n != -1)
+        {
+            _logger.fine(this + " about to wait for up to " + n + " messages to be received");
+        }
+
+        if (n == -1)
+        {
+            _credit_mode = LinkCreditMode.LINK_CREDIT_AUTO;
+        }
+        else
+        {
+            _credit_mode = LinkCreditMode.LINK_CREDIT_EXPLICIT;
+            if (n > _distributed)
+                _credit = n - _distributed;
+            else        // cancel unallocated
+                _credit = 0;
+        }
+
+        distributeCredit();
+
+        waitUntil(_messageAvailable);
+    }
+
+    public void recv() throws TimeoutException
+    {
+        recv(-1);
+    }
+
+    public int receiving()
+    {
+        return _credit + _distributed;
+    }
+
+    public Message get()
+    {
+        StoreEntry entry = _incomingStore.get( null );
+        if (entry != null)
+        {
+            Message message = Proton.message2();
+            message.decode( entry.getEncodedMsg(), 0, entry.getEncodedLength() );
+
+            _incomingTracker = new TrackerImpl(TrackerImpl.Type.INCOMING,
+                                               _incomingStore.trackEntry(entry));
+
+            _incomingStore.freeEntry( entry );
+            return message;
+        }
+        return null;
+    }
+
+    private int pumpIn(String address, Receiver receiver)
+    {
+        Delivery delivery = receiver.current();
+        if (delivery.isReadable() && !delivery.isPartial())
+        {
+            StoreEntry entry = _incomingStore.put( address );
+            entry.setDelivery( delivery );
+
+            _logger.log(Level.FINE, "Readable delivery found: " + delivery);
+
+            int size = delivery.pending();
+            byte[] buffer = new byte[size];
+            int read = receiver.recv( buffer, 0, buffer.length );
+            if (read != size) {
+                throw new IllegalStateException();
+            }
+            entry.setEncodedMsg( buffer, size );
+            receiver.advance();
+
+            // account for the used credit, replenish if
+            // low (< 20% maximum per-link batch) and
+            // extra credit available
+            assert(_distributed > 0);
+            _distributed--;
+            if (!receiver.getDrain() && _blocked.isEmpty() && _credit > 0)
+            {
+                final int max = perLinkCredit();
+                final int lo_thresh = (int)(max * 0.2 + 0.5);
+                if (receiver.getRemoteCredit() < lo_thresh)
+                {
+                    final int more = Math.min(_credit, max - receiver.getRemoteCredit());
+                    _credit -= more;
+                    _distributed += more;
+                    receiver.flow(more);
+                }
+            }
+            // check if blocked
+            if (receiver.getRemoteCredit() == 0 && _credited.contains(receiver))
+            {
+                _credited.remove(receiver);
+                if (receiver.getDrain())
+                {
+                    receiver.setDrain(false);
+                    assert( _draining > 0 );
+                    _draining--;
+                }
+                _blocked.add(receiver);
+            }
+        }
+        return 0;
+    }
+
+    public void subscribe(String source) throws MessengerException
+    {
+        if (_driver == null) {
+            throw new IllegalStateException("messenger is stopped");
+        }
+
+        String routed = routeAddress(source);
+        Address address = new Address(routed);
+
+        String hostName = address.getHost();
+        if (hostName == null) throw new MessengerException("Invalid address (hostname cannot be null): " + routed);
+        int port = Integer.valueOf(address.getImpliedPort());
+        if (address.isPassive())
+        {
+            if(_logger.isLoggable(Level.FINE))
+            {
+                _logger.fine(this + " about to subscribe to source " + source + " using address " + hostName + ":" + port);
+            }
+            ListenerContext ctx = new ListenerContext(address);
+            _driver.createListener(hostName, port, ctx);
+        }
+        else
+        {
+            if(_logger.isLoggable(Level.FINE))
+            {
+                _logger.fine(this + " about to subscribe to source " + source);
+            }
+            getLink(address, new ReceiverFinder(address.getName()));
+        }
+    }
+
+    public int outgoing()
+    {
+        return _outgoingStore.size() + queued(true);
+    }
+
+    public int incoming()
+    {
+        return _incomingStore.size() + queued(false);
+    }
+
+    public int getIncomingWindow()
+    {
+        return _incomingStore.getWindow();
+    }
+
+    public void setIncomingWindow(int window)
+    {
+        _incomingStore.setWindow(window);
+    }
+
+    public int getOutgoingWindow()
+    {
+        return _outgoingStore.getWindow();
+    }
+
+    public void setOutgoingWindow(int window)
+    {
+        _outgoingStore.setWindow(window);
+    }
+
+    public Tracker incomingTracker()
+    {
+        return _incomingTracker;
+    }
+    public Tracker outgoingTracker()
+    {
+        return _outgoingTracker;
+    }
+
+    private Store getTrackerStore(Tracker tracker)
+    {
+        return ((TrackerImpl)tracker).isOutgoing() ? _outgoingStore : _incomingStore;
+    }
+
+    @Override
+    public void reject(Tracker tracker, int flags)
+    {
+        int id = ((TrackerImpl)tracker).getSequence();
+        getTrackerStore(tracker).update(id, Status.REJECTED, flags, false, false);
+    }
+
+    @Override
+    public void accept(Tracker tracker, int flags)
+    {
+        int id = ((TrackerImpl)tracker).getSequence();
+        getTrackerStore(tracker).update(id, Status.ACCEPTED, flags, false, false);
+    }
+
+    @Override
+    public void settle(Tracker tracker, int flags)
+    {
+        int id = ((TrackerImpl)tracker).getSequence();
+        getTrackerStore(tracker).update(id, Status.UNKNOWN, flags, true, true);
+    }
+
+    public Status getStatus(Tracker tracker)
+    {
+        int id = ((TrackerImpl)tracker).getSequence();
+        StoreEntry e = getTrackerStore(tracker).getEntry(id);
+        if (e != null)
+        {
+            return e.getStatus();
+        }
+        return Status.UNKNOWN;
+    }
+
+    @Override
+    public void route(String pattern, String address)
+    {
+        _routes.rule(pattern, address);
+    }
+
+    @Override
+    public void rewrite(String pattern, String address)
+    {
+        _rewrites.rule(pattern, address);
+    }
+
+    private int queued(boolean outgoing)
+    {
+        int count = 0;
+        if (_driver != null) {
+            for (Connector<?> c : _driver.connectors())
+            {
+                Connection connection = c.getConnection();
+                for (Link link : new Links(connection, ACTIVE, ANY))
+                {
+                    if (outgoing)
+                    {
+                        if (link instanceof Sender) count += link.getQueued();
+                    }
+                    else
+                    {
+                        if (link instanceof Receiver) count += link.getQueued();
+                    }
+                }
+            }
+        }
+        return count;
+    }
+
+    private void bringDestruction()
+    {
+        for (Connector<?> c : _awaitingDestruction)
+        {
+            c.destroy();
+        }
+        _awaitingDestruction.clear();
+    }
+
+    private void processAllConnectors()
+    {
+        distributeCredit();
+        for (Connector<?> c : _driver.connectors())
+        {
+            processEndpoints(c);
+            try
+            {
+                if (c.process()) {
+                    _worked = true;
+                }
+            }
+            catch (IOException e)
+            {
+                _logger.log(Level.SEVERE, "Error processing connection", e);
+            }
+        }
+        bringDestruction();
+        distributeCredit();
+    }
+
+    private void processActive()
+    {
+        //process active listeners
+        for (Listener<?> l = _driver.listener(); l != null; l = _driver.listener())
+        {
+            _worked = true;
+            Connector<?> c = l.accept();
+            Connection connection = Proton.connection();
+            connection.setContainer(_name);
+            ListenerContext ctx = (ListenerContext) l.getContext();
+            connection.setContext(new ConnectionContext(ctx.getAddress(), c));
+            c.setConnection(connection);
+            Transport transport = c.getTransport();
+            //TODO: full SASL
+            Sasl sasl = c.sasl();
+            if (sasl != null)
+            {
+                sasl.server();
+                sasl.setMechanisms(new String[]{"ANONYMOUS"});
+                sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
+            }
+            transport.ssl(ctx.getDomain());
+            connection.open();
+        }
+        // process connectors, reclaiming credit on closed connectors
+        for (Connector<?> c = _driver.connector(); c != null; c = _driver.connector())
+        {
+            _worked = true;
+            if (c.isClosed())
+            {
+                _awaitingDestruction.add(c);
+                reclaimCredit(c.getConnection());
+            }
+            else
+            {
+                _logger.log(Level.FINE, "Processing active connector " + c);
+                try
+                {
+                    c.process();
+                    processEndpoints(c);
+                    c.process();
+                }
+                catch (IOException e)
+                {
+                    _logger.log(Level.SEVERE, "Error processing connection", e);
+                }
+            }
+        }
+        bringDestruction();
+        distributeCredit();
+    }
+
+    private void processEndpoints(Connector c)
+    {
+        Connection connection = c.getConnection();
+
+        if (connection.getLocalState() == EndpointState.UNINITIALIZED)
+        {
+            connection.open();
+        }
+
+        Delivery delivery = connection.getWorkHead();
+        while (delivery != null)
+        {
+            Link link = delivery.getLink();
+            if (delivery.isUpdated())
+            {
+                if (link instanceof Sender)
+                {
+                    delivery.disposition(delivery.getRemoteState());
+                }
+                StoreEntry e = (StoreEntry) delivery.getContext();
+                if (e != null) e.updated();
+            }
+
+            if (delivery.isReadable())
+            {
+                pumpIn( link.getSource().getAddress(), (Receiver)link );
+            }
+
+            Delivery next = delivery.getWorkNext();
+            delivery.clear();
+            delivery = next;
+        }
+
+        for (Session session : new Sessions(connection, UNINIT, ANY))
+        {
+            session.open();
+            _logger.log(Level.FINE, "Opened session " + session);
+        }
+        for (Link link : new Links(connection, UNINIT, ANY))
+        {
+            //TODO: the following is not correct; should only copy those properties that we understand
+            link.setSource(link.getRemoteSource());
+            link.setTarget(link.getRemoteTarget());
+            linkAdded(link);
+            link.open();
+            _logger.log(Level.FINE, "Opened link " + link);
+        }
+
+        distributeCredit();
+
+        for (Link link : new Links(connection, ACTIVE, ACTIVE))
+        {
+            if (link instanceof Sender)
+            {
+                pumpOut(link.getTarget().getAddress(), (Sender)link);
+            }
+        }
+
+        for (Session session : new Sessions(connection, ACTIVE, CLOSED))
+        {
+            session.close();
+        }
+
+        for (Link link : new Links(connection, ANY, CLOSED))
+        {
+            if (link.getLocalState() == EndpointState.ACTIVE)
+            {
+                link.close();
+            }
+            else
+            {
+                reclaimLink(link);
+            }
+        }
+
+        if (connection.getRemoteState() == EndpointState.CLOSED)
+        {
+            if (connection.getLocalState() == EndpointState.ACTIVE)
+            {
+                connection.close();
+            }
+        }
+    }
+
+    private boolean waitUntil(Predicate condition) throws TimeoutException
+    {
+        if (_blocking) {
+            boolean done = waitUntil(condition, _timeout);
+            if (!done) {
+                _logger.log(Level.SEVERE, String.format
+                            ("Timeout when waiting for condition %s after %s ms",
+                             condition, _timeout));
+                throw new TimeoutException();
+            }
+            return done;
+        } else {
+            return waitUntil(condition, 0);
+        }
+    }
+
+    private boolean waitUntil(Predicate condition, long timeout)
+    {
+        if (_driver == null) {
+            throw new IllegalStateException("cannot wait while messenger is stopped");
+        }
+
+        processAllConnectors();
+
+        // wait until timeout expires or until test is true
+        long now = System.currentTimeMillis();
+        final long deadline = timeout < 0 ? Long.MAX_VALUE : now + timeout;
+        boolean done = false;
+
+        while (true)
+        {
+            done = condition.test();
+            if (done) break;
+
+            long remaining;
+            if (timeout < 0)
+                remaining = -1;
+            else {
+                remaining = deadline - now;
+                if (remaining < 0) break;
+            }
+
+            // Update the credit scheduler. If the scheduler detects
+            // credit imbalance on the links, wake up in time to
+            // service credit drain
+            distributeCredit();
+            if (_next_drain != 0)
+            {
+                long wakeup = (_next_drain > now) ? _next_drain - now : 0;
+                remaining = (remaining == -1) ? wakeup : Math.min(remaining, wakeup);
+            }
+
+            boolean woken;
+            woken = _driver.doWait(remaining);
+            processActive();
+            if (woken) {
+                throw new InterruptException();
+            }
+            now = System.currentTimeMillis();
+        }
+
+        return done;
+    }
+
+    private Connection lookup(Address address)
+    {
+        for (Connector<?> c : _driver.connectors())
+        {
+            Connection connection = c.getConnection();
+            ConnectionContext ctx = (ConnectionContext) connection.getContext();
+            if (ctx.matches(address))
+            {
+                return connection;
+            }
+        }
+        return null;
+    }
+
+    private void reclaimCredit(Connection connection)
+    {
+        for (Link link : new Links(connection, ANY, ANY))
+        {
+            reclaimLink(link);
+        }
+    }
+
+    private void distributeCredit()
+    {
+        if (_receivers == 0) return;
+
+        if (_credit_mode == LinkCreditMode.LINK_CREDIT_AUTO)
+        {
+            // replenish, but limit the max total messages buffered
+            final int max = _receivers * _credit_batch;
+            final int used = _distributed + incoming();
+            if (max > used)
+                _credit = max - used;
+        }
+
+        // reclaim any credit left over after draining links has completed
+        if (_draining > 0)
+        {
+            Iterator<Receiver> itr = _credited.iterator();
+            while (itr.hasNext())
+            {
+                Receiver link = (Receiver) itr.next();
+                if (link.getDrain())
+                {
+                    if (!link.draining())
+                    {
+                        // drain completed for this link
+                        int drained = link.drained();
+                        assert(_distributed >= drained);
+                        _distributed -= drained;
+                        _credit += drained;
+                        link.setDrain(false);
+                        _draining--;
+                        itr.remove();
+                        _blocked.add(link);
+                    }
+                }
+            }
+        }
+
+        // distribute available credit to blocked links
+        final int batch = perLinkCredit();
+        while (_credit > 0 && !_blocked.isEmpty())
+        {
+            Receiver link = _blocked.get(0);
+            _blocked.remove(0);
+
+            final int more = Math.min(_credit, batch);
+            _distributed += more;
+            _credit -= more;
+
+            link.flow(more);
+            _credited.add(link);
+
+            // flow changed, must process it
+            ConnectionContext ctx = (ConnectionContext) link.getSession().getConnection().getContext();
+            try
+            {
+                ctx.getConnector().process();
+            } catch (IOException e) {
+                _logger.log(Level.SEVERE, "Error processing connection", e);
+            }
+        }
+
+        if (_blocked.isEmpty())
+        {
+            _next_drain = 0;
+        }
+        else
+        {
+            // not enough credit for all links - start draining granted credit
+            if (_draining == 0)
+            {
+                // don't do it too often - pace ourselves (it's expensive)
+                if (_next_drain == 0)
+                {
+                    _next_drain = System.currentTimeMillis() + 250;
+                }
+                else if (_next_drain <= System.currentTimeMillis())
+                {
+                    // initiate drain, free up at most enough to satisfy blocked
+                    _next_drain = 0;
+                    int needed = _blocked.size() * batch;
+
+                    for (Receiver link : _credited)
+                    {
+                        if (!link.getDrain()) {
+                            link.setDrain(true);
+                            needed -= link.getRemoteCredit();
+                            _draining++;
+                            // drain requested on link, must process it
+                            ConnectionContext ctx = (ConnectionContext) link.getSession().getConnection().getContext();
+                            try
+                            {
+                                ctx.getConnector().process();
+                            } catch (IOException e) {
+                                _logger.log(Level.SEVERE, "Error processing connection", e);
+                            }
+                            if (needed <= 0) break;
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private interface Predicate
+    {
+        boolean test();
+    }
+
+    private class SentSettled implements Predicate
+    {
+        public boolean test()
+        {
+            //are all sent messages settled?
+            int total = _outgoingStore.size();
+
+            for (Connector<?> c : _driver.connectors())
+            {
+                // TBD
+                // check if transport is done generating output
+                // pn_transport_t *transport = pn_connector_transport(ctor);
+                // if (transport) {
+                //    if (!pn_transport_quiesced(transport)) {
+                //        pn_connector_process(ctor);
+                //        return false;
+                //    }
+                // }
+
+                Connection connection = c.getConnection();
+                for (Link link : new Links(connection, ACTIVE, ANY))
+                {
+                    if (link instanceof Sender)
+                    {
+                        total += link.getQueued();
+                    }
+                }
+
+                // TBD: there is no per-link unsettled
+                // deliveries iterator, so for now get the
+                // deliveries by walking the outgoing trackers
+                Iterator<StoreEntry> entries = _outgoingStore.trackedEntries();
+                while (entries.hasNext() && total <= _sendThreshold)
+                {
+                    StoreEntry e = (StoreEntry) entries.next();
+                    if (e != null )
+                    {
+                        Delivery d = e.getDelivery();
+                        if (d != null)
+                        {
+                            if (d.getRemoteState() == null && !d.remotelySettled())
+                            {
+                                total++;
+                            }
+                        }
+                    }
+                }
+            }
+            return total <= _sendThreshold;
+        }
+    }
+
+    private class MessageAvailable implements Predicate
+    {
+        public boolean test()
+        {
+            //do we have at least one pending message?
+            if (_incomingStore.size() > 0) return true;
+            for (Connector<?> c : _driver.connectors())
+            {
+                Connection connection = c.getConnection();
+                Delivery delivery = connection.getWorkHead();
+                while (delivery != null)
+                {
+                    if (delivery.isReadable() && !delivery.isPartial())
+                    {
+                        return true;
+                    }
+                    else
+                    {
+                        delivery = delivery.getWorkNext();
+                    }
+                }
+            }
+            // if no connections, or not listening, exit as there won't ever be a message
+            if (!_driver.listeners().iterator().hasNext() && !_driver.connectors().iterator().hasNext())
+                return true;
+
+            return false;
+        }
+    }
+
+    private class AllClosed implements Predicate
+    {
+        public boolean test()
+        {
+            if (_driver == null) {
+                return true;
+            }
+
+            for (Connector<?> c : _driver.connectors()) {
+                if (!c.isClosed()) {
+                    return false;
+                }
+            }
+
+            _driver.destroy();
+            _driver = null;
+
+            return true;
+        }
+    }
+
+    private boolean _worked = false;
+
+    private class WorkPred implements Predicate
+    {
+        public boolean test()
+        {
+            return _worked;
+        }
+    }
+
+    private final SentSettled _sentSettled = new SentSettled();
+    private final MessageAvailable _messageAvailable = new MessageAvailable();
+    private final AllClosed _allClosed = new AllClosed();
+    private final WorkPred _workPred = new WorkPred();
+
+    private interface LinkFinder<C extends Link>
+    {
+        C test(Link link);
+        C create(Session session);
+    }
+
+    private class SenderFinder implements LinkFinder<Sender>
+    {
+        private final String _path;
+
+        SenderFinder(String path)
+        {
+            _path = path == null ? "" : path;
+        }
+
+        public Sender test(Link link)
+        {
+            if (link instanceof Sender && matchTarget((Target) link.getTarget(), _path))
+            {
+                return (Sender) link;
+            }
+            else
+            {
+                return null;
+            }
+        }
+
+        public Sender create(Session session)
+        {
+            Sender sender = session.sender(_path);
+            Target target = new Target();
+            target.setAddress(_path);
+            sender.setTarget(target);
+            // the C implemenation does this:
+            Source source = new Source();
+            source.setAddress(_path);
+            sender.setSource(source);
+            if (getOutgoingWindow() > 0)
+            {
+                // use explicit settlement via dispositions (not pre-settled)
+                sender.setSenderSettleMode(SenderSettleMode.UNSETTLED);
+                sender.setReceiverSettleMode(ReceiverSettleMode.SECOND);  // desired
+            }
+            return sender;
+        }
+    }
+
+    private class ReceiverFinder implements LinkFinder<Receiver>
+    {
+        private final String _path;
+
+        ReceiverFinder(String path)
+        {
+            _path = path == null ? "" : path;
+        }
+
+        public Receiver test(Link link)
+        {
+            if (link instanceof Receiver && matchSource((Source) link.getSource(), _path))
+            {
+                return (Receiver) link;
+            }
+            else
+            {
+                return null;
+            }
+        }
+
+        public Receiver create(Session session)
+        {
+            Receiver receiver = session.receiver(_path);
+            Source source = new Source();
+            source.setAddress(_path);
+            receiver.setSource(source);
+            // the C implemenation does this:
+            Target target = new Target();
+            target.setAddress(_path);
+            receiver.setTarget(target);
+            if (getIncomingWindow() > 0)
+            {
+                // use explicit settlement via dispositions (not pre-settled)
+                receiver.setSenderSettleMode(SenderSettleMode.UNSETTLED);  // desired
+                receiver.setReceiverSettleMode(ReceiverSettleMode.SECOND);
+            }
+            return receiver;
+        }
+    }
+
+    private <C extends Link> C getLink(Address address, LinkFinder<C> finder)
+    {
+        Connection connection = lookup(address);
+        if (connection == null)
+        {
+            String host = address.getHost();
+            int port = Integer.valueOf(address.getImpliedPort());
+            Connector<?> connector = _driver.createConnector(host, port, null);
+            _logger.log(Level.FINE, "Connecting to " + host + ":" + port);
+            connection = Proton.connection();
+            connection.setContainer(_name);
+            connection.setHostname(host);
+            connection.setContext(new ConnectionContext(address, connector));
+            connector.setConnection(connection);
+            Sasl sasl = connector.sasl();
+            if (sasl != null)
+            {
+                sasl.client();
+                sasl.setMechanisms(new String[]{"ANONYMOUS"});
+            }
+            if ("amqps".equalsIgnoreCase(address.getScheme())) {
+                Transport transport = connector.getTransport();
+                SslDomain domain = makeDomain(address, SslDomain.Mode.CLIENT);
+                if (_trustedDb != null) {
+                    domain.setPeerAuthentication(SslDomain.VerifyMode.VERIFY_PEER);
+                    //domain.setPeerAuthentication(SslDomain.VerifyMode.VERIFY_PEER_NAME);
+                } else {
+                    domain.setPeerAuthentication(SslDomain.VerifyMode.ANONYMOUS_PEER);
+                }
+                Ssl ssl = transport.ssl(domain);
+                //ssl.setPeerHostname(host);
+            }
+            connection.open();
+        }
+
+        for (Link link : new Links(connection, ACTIVE, ANY))
+        {
+            C result = finder.test(link);
+            if (result != null) return result;
+        }
+        Session session = connection.session();
+        session.open();
+        C link = finder.create(session);
+        linkAdded(link);
+        link.open();
+        return link;
+    }
+
+    private static class Links implements Iterable<Link>
+    {
+        private final Connection _connection;
+        private final EnumSet<EndpointState> _local;
+        private final EnumSet<EndpointState> _remote;
+
+        Links(Connection connection, EnumSet<EndpointState> local, EnumSet<EndpointState> remote)
+        {
+            _connection = connection;
+            _local = local;
+            _remote = remote;
+        }
+
+        public java.util.Iterator<Link> iterator()
+        {
+            return new LinkIterator(_connection, _local, _remote);
+        }
+    }
+
+    private static class LinkIterator implements java.util.Iterator<Link>
+    {
+        private final EnumSet<EndpointState> _local;
+        private final EnumSet<EndpointState> _remote;
+        private Link _next;
+
+        LinkIterator(Connection connection, EnumSet<EndpointState> local, EnumSet<EndpointState> remote)
+        {
+            _local = local;
+            _remote = remote;
+            _next = connection.linkHead(_local, _remote);
+        }
+
+        public boolean hasNext()
+        {
+            return _next != null;
+        }
+
+        public Link next()
+        {
+            try
+            {
+                return _next;
+            }
+            finally
+            {
+                _next = _next.next(_local, _remote);
+            }
+        }
+
+        public void remove()
+        {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+    private static class Sessions implements Iterable<Session>
+    {
+        private final Connection _connection;
+        private final EnumSet<EndpointState> _local;
+        private final EnumSet<EndpointState> _remote;
+
+        Sessions(Connection connection, EnumSet<EndpointState> local, EnumSet<EndpointState> remote)
+        {
+            _connection = connection;
+            _local = local;
+            _remote = remote;
+        }
+
+        public java.util.Iterator<Session> iterator()
+        {
+            return new SessionIterator(_connection, _local, _remote);
+        }
+    }
+
+    private static class SessionIterator implements java.util.Iterator<Session>
+    {
+        private final EnumSet<EndpointState> _local;
+        private final EnumSet<EndpointState> _remote;
+        private Session _next;
+
+        SessionIterator(Connection connection, EnumSet<EndpointState> local, EnumSet<EndpointState> remote)
+        {
+            _local = local;
+            _remote = remote;
+            _next = connection.sessionHead(_local, _remote);
+        }
+
+        public boolean hasNext()
+        {
+            return _next != null;
+        }
+
+        public Session next()
+        {
+            try
+            {
+                return _next;
+            }
+            finally
+            {
+                _next = _next.next(_local, _remote);
+            }
+        }
+
+        public void remove()
+        {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+    private void adjustReplyTo(Message m)
+    {
+        String original = m.getReplyTo();
+        if (original != null) {
+            if (original.startsWith("~/"))
+            {
+                m.setReplyTo("amqp://" + _name + "/" + original.substring(2));
+            }
+            else if (original.equals("~"))
+            {
+                m.setReplyTo("amqp://" + _name);
+            }
+        }
+    }
+
+    private static boolean matchTarget(Target target, String path)
+    {
+        if (target == null) return path.isEmpty();
+        else return path.equals(target.getAddress());
+    }
+
+    private static boolean matchSource(Source source, String path)
+    {
+        if (source == null) return path.isEmpty();
+        else return path.equals(source.getAddress());
+    }
+
+    @Override
+    public String toString()
+    {
+        StringBuilder builder = new StringBuilder();
+        builder.append("MessengerImpl [_name=").append(_name).append("]");
+        return builder.toString();
+    }
+
+    // compute the maximum amount of credit each receiving link is
+    // entitled to.  The actual credit given to the link depends on
+    // what amount of credit is actually available.
+    private int perLinkCredit()
+    {
+        if (_receivers == 0) return 0;
+        int total = _credit + _distributed;
+        return Math.max(total/_receivers, 1);
+    }
+
+    // a new link has been created, account for it.
+    private void linkAdded(Link link)
+    {
+        if (link instanceof Receiver)
+        {
+            _receivers++;
+            _blocked.add((Receiver)link);
+            link.setContext(Boolean.TRUE);
+        }
+    }
+
+    // a link is being removed, account for it.
+    private void linkRemoved(Link _link)
+    {
+        if (_link instanceof Receiver && (Boolean) _link.getContext())
+        {
+            _link.setContext(Boolean.FALSE);
+            Receiver link = (Receiver)_link;
+            assert _receivers > 0;
+            _receivers--;
+            if (link.getDrain())
+            {
+                link.setDrain(false);
+                assert _draining > 0;
+                _draining--;
+            }
+            if (_blocked.contains(link))
+                _blocked.remove(link);
+            else if (_credited.contains(link))
+                _credited.remove(link);
+            else
+                assert(false);
+        }
+    }
+
+    private class ConnectionContext
+    {
+        private Address _address;
+        private Connector _connector;
+
+        public ConnectionContext(Address address, Connector connector)
+        {
+            _address = address;
+            _connector = connector;
+        }
+
+        public Address getAddress()
+        {
+            return _address;
+        }
+
+        public boolean matches(Address address)
+        {
+            String host = address.getHost();
+            String port = address.getImpliedPort();
+            Connection conn = _connector.getConnection();
+            return host.equals(conn.getRemoteContainer()) ||
+                (_address.getHost().equals(host) && _address.getImpliedPort().equals(port));
+        }
+
+        public Connector getConnector()
+        {
+            return _connector;
+        }
+    }
+
+    private SslDomain makeDomain(Address address, SslDomain.Mode mode)
+    {
+        SslDomain domain = Proton.sslDomain();
+        domain.init(mode);
+        if (_certificate != null) {
+            domain.setCredentials(_certificate, _privateKey, _password);
+        }
+        if (_trustedDb != null) {
+            domain.setTrustedCaDb(_trustedDb);
+        }
+
+        if ("amqps".equalsIgnoreCase(address.getScheme())) {
+            domain.allowUnsecuredClient(false);
+        } else {
+            domain.allowUnsecuredClient(true);
+        }
+
+        return domain;
+    }
+
+
+    private class ListenerContext
+    {
+        private Address _address;
+        private SslDomain _domain;
+
+        public ListenerContext(Address address)
+        {
+            _address = address;
+            _domain = makeDomain(address, SslDomain.Mode.SERVER);
+        }
+
+        public SslDomain getDomain()
+        {
+            return _domain;
+        }
+
+        public Address getAddress()
+        {
+            return _address;
+        }
+
+    }
+}


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


Mime
View raw message