qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rgodf...@apache.org
Subject svn commit: r1344202 - in /qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl: DeliveryImpl.java TransportImpl.java TransportSession.java
Date Wed, 30 May 2012 11:30:59 GMT
Author: rgodfrey
Date: Wed May 30 11:30:59 2012
New Revision: 1344202

URL: http://svn.apache.org/viewvc?rev=1344202&view=rev
Log:
Proton-j : allow for multiple transfer frames

Modified:
    qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
    qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportImpl.java
    qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportSession.java

Modified: qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/DeliveryImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/DeliveryImpl.java?rev=1344202&r1=1344201&r2=1344202&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/DeliveryImpl.java (original)
+++ qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/DeliveryImpl.java Wed
May 30 11:30:59 2012
@@ -52,6 +52,7 @@ public class DeliveryImpl implements Del
     private TransportDelivery _transportDelivery;
     private byte[] _data;
     private int _dataSize;
+    private boolean _complete;
 
     public DeliveryImpl(final byte[] tag, final LinkImpl link, DeliveryImpl previous)
     {
@@ -169,7 +170,7 @@ public class DeliveryImpl implements Del
         {
             clearFlag(IO_WORK);
         }
-        return consumed;  //TODO - Implement
+        return (_complete && consumed == 0) ? TransportImpl.END_OF_STREAM : consumed;
 //TODO - Implement
     }
 
     private void clearFlag(int ioWork)
@@ -236,8 +237,9 @@ public class DeliveryImpl implements Del
         }
     }
 
-    void clearTransportWork()
+    DeliveryImpl clearTransportWork()
     {
+        DeliveryImpl next = _transportWorkNext;
         getLink().getConnectionImpl().removeTransportWork(this);
         if(_transportWorkPrev != null)
         {
@@ -250,11 +252,17 @@ public class DeliveryImpl implements Del
         }
         _transportWorkNext = null;
         _transportWorkPrev = null;
+        return next;
     }
 
     void addToTransportWorkList()
     {
-        getLink().getConnectionImpl().addTransportWork(this);
+        if(_transportWorkNext == null
+           && _transportWorkPrev == null
+           && getLink().getConnectionImpl().getTransportWorkHead() != this)
+        {
+            getLink().getConnectionImpl().addTransportWork(this);
+        }
     }
 
 
@@ -281,6 +289,10 @@ public class DeliveryImpl implements Del
 
     void setTransportWorkNext(DeliveryImpl transportWorkNext)
     {
+        if(transportWorkNext == this)
+        {
+            (new Exception("Aaaargh")).printStackTrace();
+        }
         _transportWorkNext = transportWorkNext;
     }
 
@@ -323,7 +335,8 @@ public class DeliveryImpl implements Del
         }
         System.arraycopy(bytes,offset,_data,_dataSize,length);
         _dataSize+=length;
-        addToWorkList();
+//        addToWorkList();
+        addToTransportWorkList();
         return length;  //TODO - Implement.
     }
 
@@ -342,7 +355,7 @@ public class DeliveryImpl implements Del
         return _dataSize;  //TODO - Implement.
     }
 
-    public void setData(byte[] data)
+    void setData(byte[] data)
     {
         _data = data;
     }
@@ -370,4 +383,9 @@ public class DeliveryImpl implements Del
                 && getLink().current() == this
                 && _dataSize > 0;
     }
+
+    void setComplete()
+    {
+        _complete = true;
+    }
 }

Modified: qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportImpl.java?rev=1344202&r1=1344201&r2=1344202&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportImpl.java (original)
+++ qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportImpl.java Wed
May 30 11:30:59 2012
@@ -205,8 +205,8 @@ public class TransportImpl extends Endpo
 
 
                 TransportLink transportLink = sender.getTransportLink();
+
                 UnsignedInteger deliveryId = transportLink.getDeliveryCount();
-                transportLink.setDeliveryCount(deliveryId.add(UnsignedInteger.ONE));
                 TransportDelivery transportDelivery = new TransportDelivery(deliveryId, delivery,
transportLink);
 
 
@@ -214,6 +214,10 @@ public class TransportImpl extends Endpo
                 transfer.setDeliveryId(deliveryId);
                 transfer.setDeliveryTag(new Binary(delivery.getTag()));
                 transfer.setHandle(transportLink.getLocalHandle());
+                if(delivery.getLink().current() == delivery)
+                {
+                    transfer.setMore(true);
+                }
                 transfer.setMessageFormat(UnsignedInteger.ZERO);
 
                 // TODO - large frames
@@ -228,8 +232,24 @@ public class TransportImpl extends Endpo
                 offset += frameBytes;
                 length -= frameBytes;
 
+                // TODO partial consumption
+                delivery.setData(null);
+                delivery.setDataLength(0);
+
+                if(delivery.getLink().current() != delivery)
+                {
+                    transportLink.setDeliveryCount(transportLink.getDeliveryCount().add(UnsignedInteger.ONE));
+                }
+
+                delivery = delivery.clearTransportWork();
+
+
+
+            }
+            else
+            {
+                delivery = delivery.getTransportWorkNext();
             }
-            delivery = delivery.getTransportWorkNext();
         }
         return written;
     }
@@ -685,7 +705,7 @@ public class TransportImpl extends Endpo
 
     public void handleTransfer(Transfer transfer, Binary payload, Integer channel)
     {
-        System.out.println("CH["+channel+"] : " + transfer);
+        System.out.println("CH["+channel+"] : " + transfer + " ["+payload+"]");
         // TODO - check channel < max_channel
         TransportSession transportSession = _remoteSessions[channel];
         if(transportSession != null)

Modified: qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportSession.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportSession.java?rev=1344202&r1=1344201&r2=1344202&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportSession.java
(original)
+++ qpid/proton/trunk/proton-j/src/org/apache/qpid/proton/engine/impl/TransportSession.java
Wed May 30 11:30:59 2012
@@ -49,6 +49,7 @@ class TransportSession
     private UnsignedInteger _remoteOutgoingWindow;
     private UnsignedInteger _remoteNextIncomingId;
     private UnsignedInteger _remoteNextOutgoingId;
+    private Map<UnsignedInteger, DeliveryImpl> _unsettledDeliveriesById = new HashMap<UnsignedInteger,
DeliveryImpl>();
 
     public TransportSession(SessionImpl session)
     {
@@ -181,11 +182,15 @@ class TransportSession
 
     public void handleTransfer(Transfer transfer, Binary payload)
     {
-
+        DeliveryImpl delivery;
         if(transfer.getDeliveryId() == null || transfer.getDeliveryId().equals(_currentDeliveryId))
         {
+            TransportReceiver transportReceiver = (TransportReceiver) getLinkFromRemoteHandle(transfer.getHandle());
+            ReceiverImpl receiver = transportReceiver.getReceiver();
+            Binary deliveryTag = transfer.getDeliveryTag();
+            delivery = _unsettledDeliveriesById.get(_currentDeliveryId);
+
 
-            // TODO - handle large messages
         }
         else
         {
@@ -195,24 +200,27 @@ class TransportSession
             TransportReceiver transportReceiver = (TransportReceiver) getLinkFromRemoteHandle(transfer.getHandle());
             ReceiverImpl receiver = transportReceiver.getReceiver();
             Binary deliveryTag = transfer.getDeliveryTag();
-            DeliveryImpl delivery = receiver.delivery(deliveryTag.getArray(), deliveryTag.getArrayOffset(),
+            delivery = receiver.delivery(deliveryTag.getArray(), deliveryTag.getArrayOffset(),
                                                       deliveryTag.getLength());
             TransportDelivery transportDelivery = new TransportDelivery(_currentDeliveryId,
delivery, transportReceiver);
             delivery.setTransportDelivery(transportDelivery);
-            // TODO - should this be a copy?
-            if(payload != null)
-            {
-                delivery.setData(payload.getArray());
-                delivery.setDataLength(payload.getLength());
-                delivery.setDataOffset(payload.getArrayOffset());
-            }
-            delivery.addIOWork();
+            _unsettledDeliveriesById.put(_currentDeliveryId, delivery);
 
 
         }
+        // TODO - should this be a copy?
+        if(payload != null)
+        {
+            delivery.setData(payload.getArray());
+            delivery.setDataLength(payload.getLength());
+            delivery.setDataOffset(payload.getArrayOffset());
+        }
+        delivery.addIOWork();
+
 
         if(!(transfer.getMore() || transfer.getAborted()))
         {
+            delivery.setComplete();
             _incomingWindowSize = _incomingWindowSize.subtract(UnsignedInteger.ONE);
         }
 



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


Mime
View raw message