diff --git a/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNIDelivery.java b/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNIDelivery.java
index 87e3082..cb4d899 100644
--- a/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNIDelivery.java
+++ b/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNIDelivery.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.proton.engine.jni;
 
 import org.apache.qpid.proton.ProtonCEquivalent;
+import org.apache.qpid.proton.ProtonUnsupportedOperationException;
 import org.apache.qpid.proton.engine.Delivery;
 import org.apache.qpid.proton.engine.Link;
 import org.apache.qpid.proton.jni.Proton;
@@ -222,6 +223,12 @@
     }
 
     @Override
+    public boolean isPartial()
+    {
+        throw new ProtonUnsupportedOperationException();
+    }
+
+    @Override
     @ProtonCEquivalent("pn_delivery_settled")
     public boolean isSettled()
     {
diff --git a/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/messenger/jni/JNIMessengerFactory.java b/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/messenger/jni/JNIMessengerFactory.java
index 2b43d6f..09dad43 100644
--- a/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/messenger/jni/JNIMessengerFactory.java
+++ b/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/messenger/jni/JNIMessengerFactory.java
@@ -19,6 +19,7 @@
  */
 package org.apache.qpid.proton.messenger.jni;
 
+import org.apache.qpid.proton.ProtonUnsupportedOperationException;
 import org.apache.qpid.proton.messenger.Messenger;
 import org.apache.qpid.proton.messenger.MessengerFactory;
 
@@ -28,13 +29,13 @@
     @Override
     public Messenger createMessenger()
     {
-        throw new UnsupportedOperationException();
+        throw new ProtonUnsupportedOperationException();
     }
 
     @Override
     public Messenger createMessenger(String name)
     {
-        throw new UnsupportedOperationException();
+        throw new ProtonUnsupportedOperationException();
     }
 
 }
diff --git a/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Delivery.java b/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Delivery.java
index deb5b14..8334198 100644
--- a/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Delivery.java
+++ b/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Delivery.java
@@ -68,4 +68,6 @@
     public Object getContext();
 
     public boolean isUpdated();
+
+    public boolean isPartial();
 }
diff --git a/proton-j/proton-api/src/main/resources/proton.py b/proton-j/proton-api/src/main/resources/proton.py
index 5e3c6bb..71987fa 100644
--- a/proton-j/proton-api/src/main/resources/proton.py
+++ b/proton-j/proton-api/src/main/resources/proton.py
@@ -25,11 +25,12 @@
     EndpointState, TransportException
 from org.apache.qpid.proton.message import \
     MessageFormat, MessageFactory, Message as JMessage
-from org.apache.qpid.proton.messenger import MessengerException, Status
+from org.apache.qpid.proton.messenger import MessengerFactory, MessengerException, Status
 from org.apache.qpid.proton.amqp.messaging import Source, Target, Accepted, AmqpValue
 from org.apache.qpid.proton.amqp import UnsignedInteger
 from jarray import zeros
 from java.util import EnumSet, UUID as JUUID
+from java.util.concurrent import TimeoutException as Timeout
 
 LANGUAGE = "Java"
 
@@ -55,6 +56,7 @@
 protonFactoryLoader = ProtonFactoryLoader()
 engineFactory = protonFactoryLoader.loadFactory(EngineFactory)
 messageFactory = protonFactoryLoader.loadFactory(MessageFactory)
+messengerFactory = protonFactoryLoader.loadFactory(MessengerFactory)
 
 
 class Endpoint(object):
@@ -511,15 +513,14 @@
   def __init__(self, *args, **kwargs):
     raise Skipped()
 
-class Timeout(Exception):
-  pass
-
 class Messenger(object):
 
   def __init__(self, *args, **kwargs):
-    #comment out or remove line below to enable messenger tests
-    raise Skipped()
-    self.impl = MessengerImpl()
+    try:
+      self.impl = messengerFactory.createMessenger()
+    except ProtonUnsupportedOperationException:
+      raise Skipped()
+
 
   def start(self):
     self.impl.start()
@@ -541,10 +542,9 @@
     self.impl.recv(n)
 
   def get(self, message=None):
-    if message is None:
-      self.impl.get()
-    else:
-      message.impl = self.impl.get()
+    result = self.impl.get()
+    if message and result:
+      message.impl = result
     return self.impl.incomingTracker()
 
   @property
diff --git a/proton-j/proton/src/main/java/org/apache/qpid/proton/driver/impl/ConnectorImpl.java b/proton-j/proton/src/main/java/org/apache/qpid/proton/driver/impl/ConnectorImpl.java
index 0bac31a..972c4dd 100644
--- a/proton-j/proton/src/main/java/org/apache/qpid/proton/driver/impl/ConnectorImpl.java
+++ b/proton-j/proton/src/main/java/org/apache/qpid/proton/driver/impl/ConnectorImpl.java
@@ -55,7 +55,7 @@
 
     private ByteBuffer _readBuffer = ByteBuffer.allocate(readBufferSize);
     private ByteBuffer _writeBuffer = ByteBuffer.allocate(writeBufferSize);
-    private boolean _readPending;
+    private boolean _readPending = true;
 
     ConnectorImpl(DriverImpl driver, Listener<C> listener, SocketChannel c, C context, SelectionKey key)
     {
@@ -81,33 +81,56 @@
                 _readPending = false;
                 if (isClosed()) return;
             }
+            else
+            {
+                processInput();
+            }
             write();
         }
     }
 
-    void read() throws IOException
+    private void read() throws IOException
     {
         int bytesRead = 0;
         while ((bytesRead = _channel.read(_readBuffer)) > 0)
         {
-            _readBuffer.flip();
-            int consumed = _transport.input(_readBuffer.array(), _readBuffer.position(), _readBuffer.limit());
-            _readBuffer.position(consumed == Transport.END_OF_STREAM ? _readBuffer.limit() : consumed);
-            if (_logger.isLoggable(Level.FINE))
-            {
-                _logger.log(Level.FINE, "consumed " + consumed + " bytes, " + _readBuffer.remaining() + " available");
-            }
-            _readBuffer.compact();
+            processInput();
         }
         if (bytesRead == -1) {
             close();
         }
     }
 
-    void write() throws IOException
+    private int processInput() throws IOException
+    {
+        _readBuffer.flip();
+        int total = 0;
+        while (_readBuffer.hasRemaining())
+        {
+            int consumed = _transport.input(_readBuffer.array(), _readBuffer.position(), _readBuffer.remaining());
+            if (consumed == Transport.END_OF_STREAM)
+            {
+                continue;
+            }
+            else if (consumed == 0)
+            {
+                break;
+            }
+            _readBuffer.position(_readBuffer.position() + consumed);
+            if (_logger.isLoggable(Level.FINE))
+            {
+                _logger.log(Level.FINE, "consumed " + consumed + " bytes, " + _readBuffer.remaining() + " available");
+            }
+            total += consumed;
+        }
+        _readBuffer.compact();
+        return total;
+    }
+
+    private void write() throws IOException
     {
         int interest = _key.interestOps();
-        int start = _writeBuffer.position();
+        boolean empty = _writeBuffer.position() == 0;
         boolean done = false;
         while (!done)
         {
@@ -119,19 +142,20 @@
             {
                 _logger.log(Level.FINE, "wrote " + wrote + " bytes, " + _writeBuffer.remaining() + " remaining");
             }
-            _writeBuffer.compact();
-            if (_writeBuffer.position() > 0)
+            if (_writeBuffer.hasRemaining())
             {
                 //weren't able to write all available data, ask to be notfied when we can write again
+                _writeBuffer.compact();
                 interest |= SelectionKey.OP_WRITE;
                 done = true;
             }
             else
             {
-                //we are done if buffer was empty to begin with and we did not produce enough to fill it
+                //we are done if buffer was empty to begin with and we did not produce anything
+                _writeBuffer.clear();
                 interest &= ~SelectionKey.OP_WRITE;
-                done = start == 0 && produced < _writeBuffer.capacity();
-                start = 0;
+                done = empty && produced == 0;
+                empty = true;
             }
         }
         _key.interestOps(interest);
diff --git a/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java b/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
index 60ec3ab..76c8736 100644
--- a/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
+++ b/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
@@ -408,6 +408,11 @@
         _complete = true;
     }
 
+    public boolean isPartial()
+    {
+        return !_complete;
+    }
+
     void setRemoteDeliveryState(DeliveryState remoteDeliveryState)
     {
         _remoteDeliveryState = remoteDeliveryState;
diff --git a/proton-j/proton/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java b/proton-j/proton/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java
index 8da4779..6815471 100644
--- a/proton-j/proton/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java
+++ b/proton-j/proton/src/main/java/org/apache/qpid/proton/messenger/impl/MessengerImpl.java
@@ -134,7 +134,6 @@
             try
             {
                 c.process();
-                c.close();
             }
             catch (IOException e)
             {
@@ -153,6 +152,14 @@
                 _logger.log(Level.WARNING, "Error while closing listener", e);
             }
         }
+        try
+        {
+            waitUntil(_allClosed);
+        }
+        catch(TimeoutException e)
+        {
+            _logger.log(Level.WARNING, "Timed out while waiting for close", e);
+        }
         _driver.destroy();
     }
 
@@ -215,7 +222,7 @@
             Delivery delivery = connection.getWorkHead();
             while (delivery != null)
             {
-                if (delivery.isReadable())
+                if (delivery.isReadable() && !delivery.isPartial())
                 {
                     _logger.log(Level.FINE, "Readable delivery found: " + delivery);
                     int size = read((Receiver) delivery.getLink());
@@ -463,9 +470,16 @@
             {
                 session.close();
             }
-            if (connection.getLocalState() == EndpointState.ACTIVE && connection.getRemoteState() == EndpointState.CLOSED)
+            if (connection.getRemoteState() == EndpointState.CLOSED)
             {
-                connection.close();
+                if (connection.getLocalState() == EndpointState.ACTIVE)
+                {
+                    connection.close();
+                }
+                else if (connection.getLocalState() == EndpointState.CLOSED)
+                {
+                    c.close();
+                }
             }
 
             if (c.isClosed())
@@ -501,7 +515,7 @@
 
         boolean wait = deadline > System.currentTimeMillis();
         boolean first = true;
-        boolean done = condition.test();
+        boolean done = false;
 
         while (first || (!done && wait))
         {
@@ -513,6 +527,10 @@
             done = done || condition.test();
             first = false;
         }
+        if (!done)
+        {
+            throw new TimeoutException();
+        }
     }
 
     private Connection lookup(String host, String service)
@@ -581,7 +599,7 @@
             for (Connector c : _driver.connectors())
             {
                 Connection connection = c.getConnection();
-                for (Link link : new Links(connection, ACTIVE, ACTIVE))
+                for (Link link : new Links(connection, ACTIVE, ANY))
                 {
                     if (link instanceof Sender)
                     {
@@ -642,9 +660,8 @@
                 Delivery delivery = connection.getWorkHead();
                 while (delivery != null)
                 {
-                    if (delivery.isReadable())
+                    if (delivery.isReadable() && !delivery.isPartial())
                     {
-                        //TODO: check for partial delivery?
                         return true;
                     }
                     else
@@ -657,8 +674,19 @@
         }
     }
 
+    private class AllClosed implements Predicate
+    {
+        public boolean test()
+        {
+            if (_driver.connectors().iterator().hasNext()) return false;
+            else return true;
+        }
+
+    }
+
     private final SentSettled _sentSettled = new SentSettled();
     private final MessageAvailable _messageAvailable = new MessageAvailable();
+    private final AllClosed _allClosed = new AllClosed();
 
     private interface LinkFinder<C extends Link>
     {