You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2013/08/01 15:58:55 UTC
svn commit: r1509238 - in /qpid/proton/trunk:
proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/
proton-c/src/engine/
proton-j/proton-api/src/main/java/org/apache/qpid/proton/amqp/
proton-j/proton-api/src/main/java/org/apache/qpid/...
Author: rhs
Date: Thu Aug 1 13:58:54 2013
New Revision: 1509238
URL: http://svn.apache.org/r1509238
Log:
Fix for PROTON-331, this includes the following changes:
- factored duplicate frame writing code into common FrameWriter class
- fixed max frame size negotiation
- added accessors for remote max frame size
- simplified buffer management inside TransportImpl
- fixed processing of the transport work queue
- fixed session flow control
- added per session tracking and accessors for incoming/outgoing bytes
- added configuration for session capacity
- fixed Binary.toString()
- added get/setContext to java Connection
- added useful messages to the java shim's unsupported operation exceptions
- fixed an error in the C engines computation of outgoing bytes
- added some test assertions
Added:
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java
Modified:
qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNISession.java
qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNITransport.java
qpid/proton/trunk/proton-c/src/engine/engine.c
qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/amqp/Binary.java
qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Connection.java
qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Session.java
qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Transport.java
qpid/proton/trunk/proton-j/proton-api/src/main/resources/proton.py
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameParser.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ReceiverImpl.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SenderImpl.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java
qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java
qpid/proton/trunk/tests/python/proton_tests/engine.py
Modified: qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNISession.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNISession.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNISession.java (original)
+++ qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNISession.java Thu Aug 1 13:58:54 2013
@@ -52,6 +52,30 @@ public class JNISession implements Sessi
}
@Override
+ public int getIncomingCapacity()
+ {
+ return (int) Proton.pn_session_get_incoming_capacity(_impl);
+ }
+
+ @Override
+ public void setIncomingCapacity(int capacity)
+ {
+ Proton.pn_session_set_incoming_capacity(_impl, capacity);
+ }
+
+ @Override
+ public int getIncomingBytes()
+ {
+ return (int) Proton.pn_session_incoming_bytes(_impl);
+ }
+
+ @Override
+ public int getOutgoingBytes()
+ {
+ return (int) Proton.pn_session_outgoing_bytes(_impl);
+ }
+
+ @Override
@ProtonCEquivalent("pn_sender")
public Sender sender(String name)
{
@@ -206,4 +230,5 @@ public class JNISession implements Sessi
}
return null;
}
+
}
Modified: qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNITransport.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNITransport.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNITransport.java (original)
+++ qpid/proton/trunk/proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/JNITransport.java Thu Aug 1 13:58:54 2013
@@ -364,6 +364,12 @@ public class JNITransport implements Tra
return (int) Proton.pn_transport_get_max_frame(_impl);
}
+ @Override
+ public int getRemoteMaxFrameSize()
+ {
+ return (int) Proton.pn_transport_get_remote_max_frame(_impl);
+ }
+
@Override
public Object getContext()
Modified: qpid/proton/trunk/proton-c/src/engine/engine.c
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/src/engine/engine.c?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/src/engine/engine.c (original)
+++ qpid/proton/trunk/proton-c/src/engine/engine.c Thu Aug 1 13:58:54 2013
@@ -2541,7 +2541,6 @@ int pn_process_tpwork_sender(pn_transpor
ssn_state->remote_incoming_window > 0 && link_state->link_credit > 0) {
pn_bytes_t bytes = pn_buffer_bytes(delivery->bytes);
pn_set_payload(transport->disp, bytes.start, bytes.size);
- link->session->outgoing_bytes -= bytes.size;
pn_bytes_t tag = pn_buffer_bytes(delivery->tag);
int count = pn_post_transfer_frame(transport->disp,
ssn_state->local_channel,
@@ -2556,7 +2555,9 @@ int pn_process_tpwork_sender(pn_transpor
ssn_state->outgoing_transfer_count += count;
ssn_state->remote_incoming_window -= count;
- pn_buffer_trim(delivery->bytes, bytes.size - transport->disp->output_size, 0);
+ int sent = bytes.size - transport->disp->output_size;
+ pn_buffer_trim(delivery->bytes, sent, 0);
+ link->session->outgoing_bytes -= sent;
if (!pn_buffer_size(delivery->bytes) && delivery->done) {
state->sent = true;
link_state->delivery_count++;
Modified: qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/amqp/Binary.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/amqp/Binary.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/amqp/Binary.java (original)
+++ qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/amqp/Binary.java Thu Aug 1 13:58:54 2013
@@ -115,9 +115,9 @@ public final class Binary
StringBuilder str = new StringBuilder();
- for (int i = _offset; i < _length; i++)
+ for (int i = 0; i < _length; i++)
{
- byte c = _data[i];
+ byte c = _data[_offset + i];
if (c > 31 && c < 127 && c != '\\')
{
Modified: qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Connection.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Connection.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Connection.java (original)
+++ qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Connection.java Thu Aug 1 13:58:54 2013
@@ -98,4 +98,8 @@ public interface Connection extends Endp
EngineLogger getEngineLogger();
void setEngineLogger(EngineLogger engineLogger);
+
+ Object getContext();
+
+ void setContext(Object context);
}
Modified: qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Session.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Session.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Session.java (original)
+++ qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Session.java Thu Aug 1 13:58:54 2013
@@ -43,4 +43,13 @@ public interface Session extends Endpoin
public Session next(EnumSet<EndpointState> local, EnumSet<EndpointState> remote);
public Connection getConnection();
+
+ public int getIncomingCapacity();
+
+ public void setIncomingCapacity(int bytes);
+
+ public int getIncomingBytes();
+
+ public int getOutgoingBytes();
+
}
Modified: qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Transport.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Transport.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Transport.java (original)
+++ qpid/proton/trunk/proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/Transport.java Thu Aug 1 13:58:54 2013
@@ -63,7 +63,7 @@ import java.nio.ByteBuffer;
*/
public interface Transport extends Endpoint
{
- public static final int DEFAULT_MAX_FRAME_SIZE = 16 * 1024;
+ public static final int DEFAULT_MAX_FRAME_SIZE = -1;
/** the lower bound for the agreed maximum frame size (in bytes). */
public int MIN_MAX_FRAME_SIZE = 512;
@@ -174,6 +174,8 @@ public interface Transport extends Endpo
void setMaxFrameSize(int size);
+ int getRemoteMaxFrameSize();
+
EngineLogger getEngineLogger();
void setEngineLogger(EngineLogger engineLogger);
}
Modified: qpid/proton/trunk/proton-j/proton-api/src/main/resources/proton.py
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton-api/src/main/resources/proton.py?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton-api/src/main/resources/proton.py (original)
+++ qpid/proton/trunk/proton-j/proton-api/src/main/resources/proton.py Thu Aug 1 13:58:54 2013
@@ -190,20 +190,24 @@ class Condition(object):
return self.impl
def wrap_connection(impl):
- if impl: return Connection(_impl = impl)
+ if impl:
+ return impl.getContext()
+ else:
+ return None
class Connection(Endpoint):
- def __init__(self, _impl=None):
+ def __init__(self):
Endpoint.__init__(self)
- self.impl = _impl or Proton.connection()
+ self.impl = Proton.connection()
+ self.impl.setContext(self)
self.desired_capabilities = None
self.offered_capabilities = None
self.properties = None
@property
def writable(self):
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Connection.writable")
def session(self):
return wrap_session(self.impl.session())
@@ -282,6 +286,22 @@ class Session(Endpoint):
def receiver(self, name):
return wrap_link(self.impl.receiver(name))
+ def _get_incoming_capacity(self):
+ return self.impl.getIncomingCapacity()
+
+ def _set_incoming_capacity(self, capacity):
+ self.impl.setIncomingCapacity(capacity)
+
+ incoming_capacity = property(_get_incoming_capacity, _set_incoming_capacity)
+
+ @property
+ def outgoing_bytes(self):
+ return self.impl.getOutgoingBytes()
+
+ @property
+ def incoming_bytes(self):
+ return self.impl.getIncomingBytes()
+
def wrap_link(impl):
if impl is None: return None
elif isinstance(impl, JSender):
@@ -347,7 +367,7 @@ class Link(Endpoint):
@property
def available(self):
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Link.available")
@property
def queued(self):
@@ -383,7 +403,7 @@ class DataDummy:
pass
def put_array(self, *args, **kwargs):
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Data.put_array")
class Terminus(object):
@@ -450,7 +470,7 @@ class Terminus(object):
class Sender(Link):
def offered(self, n):
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Sender.offered")
def send(self, bytes):
return self.impl.send(bytes, 0, len(bytes))
@@ -746,7 +766,6 @@ class Transport(object):
def _set_max_frame_size(self, value):
self.impl.setMaxFrameSize(value)
- raise ProtonUnsupportedOperationException()
max_frame_size = property(_get_max_frame_size, _set_max_frame_size,
doc="""
@@ -755,17 +774,16 @@ Sets the maximum size for received frame
@property
def remote_max_frame_size(self):
- #return pn_transport_get_remote_max_frame(self._trans)
- raise ProtonUnsupportedOperationException()
+ return self.impl.getRemoteMaxFrameSize()
# AMQP 1.0 idle-time-out
def _get_idle_timeout(self):
#return pn_transport_get_idle_timeout(self._trans)
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Transport.idle_timeout")
def _set_idle_timeout(self, value):
#pn_transport_set_idle_timeout(self._trans, value)
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Transport.idle_timeout")
idle_timeout = property(_get_idle_timeout, _set_idle_timeout,
doc="""
@@ -775,17 +793,17 @@ The idle timeout of the connection (in m
@property
def remote_idle_timeout(self):
#return pn_transport_get_remote_idle_timeout(self._trans)
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Transport.remote_idle_timeout")
@property
def frames_output(self):
#return pn_transport_get_frames_output(self._trans)
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Transport.frames_output")
@property
def frames_input(self):
#return pn_transport_get_frames_input(self._trans)
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Transport.frames_input")
class UnmappedType:
@@ -1650,15 +1668,15 @@ class SSL(object):
class Driver(object):
""" Proton-c platform abstraction - not needed."""
def __init__(self, *args, **kwargs):
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Driver")
class Connector(object):
""" Proton-c platform abstraction - not needed."""
def __init__(self, *args, **kwargs):
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Connector")
class Listener(object):
""" Proton-c platform abstraction - not needed."""
def __init__(self, *args, **kwargs):
- raise ProtonUnsupportedOperationException()
+ raise ProtonUnsupportedOperationException("Listener")
def convertToPyArray(t,a,f):
if a == None or len(a) == 0:
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java Thu Aug 1 13:58:54 2013
@@ -66,6 +66,8 @@ public class ConnectionImpl extends Endp
private EngineLogger _engineLogger;
+ private Object _context;
+
private static final Symbol[] EMPTY_SYMBOL_ARRAY = new Symbol[0];
/**
@@ -534,4 +536,15 @@ public class ConnectionImpl extends Endp
{
_engineLogger = engineLogger;
}
+
+ public Object getContext()
+ {
+ return _context;
+ }
+
+ public void setContext(Object context)
+ {
+ _context = context;
+ }
+
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameParser.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameParser.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameParser.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameParser.java Thu Aug 1 13:58:54 2013
@@ -89,7 +89,11 @@ class FrameParser implements TransportIn
{
_frameHandler = frameHandler;
_decoder = decoder;
- _inputBuffer = newWriteableBuffer(maxFrameSize);
+ if (maxFrameSize > 0) {
+ _inputBuffer = newWriteableBuffer(maxFrameSize);
+ } else {
+ _inputBuffer = newWriteableBuffer(4*1024);
+ }
}
private TransportResult input(ByteBuffer in)
Added: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java?rev=1509238&view=auto
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java (added)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameWriter.java Thu Aug 1 13:58:54 2013
@@ -0,0 +1,210 @@
+/*
+ *
+ * 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.engine.impl;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.transport.FrameBody;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.WritableBuffer;
+import org.apache.qpid.proton.framing.TransportFrame;
+
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * FrameWriter
+ *
+ */
+
+class FrameWriter
+{
+
+ static final byte AMQP_FRAME_TYPE = 0;
+ static final byte SASL_FRAME_TYPE = (byte) 1;
+
+ private EncoderImpl _encoder;
+ private ByteBuffer _bbuf;
+ private WritableBuffer _buffer;
+ private int _maxFrameSize;
+ private byte _frameType;
+ private ProtocolTracer _protocolTracer;
+
+ private int _frameStart = 0;
+ private int _payloadStart;
+ private int _performativeSize;
+
+ FrameWriter(EncoderImpl encoder, int maxFrameSize, byte frameType,
+ ProtocolTracer protocolTracer)
+ {
+ _encoder = encoder;
+ _bbuf = ByteBuffer.allocate(1024);
+ _buffer = new WritableBuffer.ByteBufferWrapper(_bbuf);
+ _encoder.setByteBuffer(_buffer);
+ _maxFrameSize = maxFrameSize;
+ _frameType = frameType;
+ _protocolTracer = protocolTracer;
+ }
+
+ void setMaxFrameSize(int maxFrameSize)
+ {
+ _maxFrameSize = maxFrameSize;
+ }
+
+ private void grow()
+ {
+ ByteBuffer old = _bbuf;
+ _bbuf = ByteBuffer.allocate(_bbuf.capacity() * 2);
+ _buffer = new WritableBuffer.ByteBufferWrapper(_bbuf);
+ old.flip();
+ _bbuf.put(old);
+ _encoder.setByteBuffer(_buffer);
+ }
+
+ void writeHeader(byte[] header)
+ {
+ _buffer.put(header, 0, header.length);
+ }
+
+ private void startFrame()
+ {
+ _frameStart = _buffer.position();
+ }
+
+ private void writePerformative(Object frameBody)
+ {
+ while (_buffer.remaining() < 8) {
+ grow();
+ }
+
+ while (true)
+ {
+ try
+ {
+ _buffer.position(_frameStart + 8);
+ _encoder.writeObject(frameBody);
+ break;
+ }
+ catch (BufferOverflowException e)
+ {
+ grow();
+ }
+ }
+
+ _payloadStart = _buffer.position();
+ _performativeSize = _payloadStart - _frameStart;
+ }
+
+ private void endFrame(int channel)
+ {
+ int frameSize = _buffer.position() - _frameStart;
+ int limit = _buffer.position();
+ _buffer.position(_frameStart);
+ _buffer.putInt(frameSize);
+ _buffer.put((byte) 2);
+ _buffer.put(_frameType);
+ _buffer.putShort((short) channel);
+ _buffer.position(limit);
+
+ int offset = _bbuf.arrayOffset() + _frameStart;
+ //System.out.println("RAW: \"" + new Binary(_bbuf.array(), offset, frameSize) + "\"");
+ }
+
+ void writeFrame(int channel, Object frameBody, ByteBuffer payload,
+ Runnable onPayloadTooLarge)
+ {
+ startFrame();
+
+ writePerformative(frameBody);
+
+ if(_maxFrameSize > 0 && payload != null && (payload.remaining() + _performativeSize) > _maxFrameSize)
+ {
+ if(onPayloadTooLarge != null)
+ {
+ onPayloadTooLarge.run();
+ }
+ writePerformative(frameBody);
+ }
+
+ ByteBuffer originalPayload = null;
+ if( payload!=null )
+ {
+ originalPayload = payload.duplicate();
+ }
+
+ // XXX: this is a bit of a hack but it eliminates duplicate
+ // code, further refactor will fix this
+ if (_frameType == AMQP_FRAME_TYPE) {
+ TransportFrame frame = new TransportFrame(channel, (FrameBody) frameBody, Binary.create(originalPayload));
+ TransportImpl.log(this, TransportImpl.OUTGOING, frame);
+
+ if( _protocolTracer!=null )
+ {
+ _protocolTracer.sentFrame(frame);
+ }
+ }
+
+ int capacity;
+ if (_maxFrameSize > 0) {
+ capacity = _maxFrameSize - _performativeSize;
+ } else {
+ capacity = Integer.MAX_VALUE;
+ }
+ int payloadSize = Math.min(payload == null ? 0 : payload.remaining(), capacity);
+
+ if(payloadSize > 0)
+ {
+ while (_buffer.remaining() < payloadSize) {
+ grow();
+ }
+
+ int oldLimit = payload.limit();
+ payload.limit(payload.position() + payloadSize);
+ _buffer.put(payload);
+ payload.limit(oldLimit);
+ }
+
+ endFrame(channel);
+ }
+
+ void writeFrame(Object frameBody)
+ {
+ writeFrame(0, frameBody, null, null);
+ }
+
+ int readBytes(ByteBuffer dst)
+ {
+ ByteBuffer src = _bbuf.duplicate();
+ src.flip();
+
+ int size = Math.min(src.remaining(), dst.remaining());
+ int limit = src.limit();
+ src.limit(size);
+ dst.put(src);
+ src.limit(limit);
+ _bbuf.rewind();
+ _bbuf.put(src);
+
+ //System.out.println("RAW: \"" + new Binary(dst.array(), dst.arrayOffset(), dst.position()) + "\"");
+
+ return size;
+ }
+
+}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ReceiverImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ReceiverImpl.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ReceiverImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ReceiverImpl.java Thu Aug 1 13:58:54 2013
@@ -21,6 +21,7 @@
package org.apache.qpid.proton.engine.impl;
import java.util.Iterator;
+import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Receiver;
@@ -30,15 +31,21 @@ public class ReceiverImpl extends LinkIm
@Override
public boolean advance()
{
- if(current() != null)
+ DeliveryImpl current = current();
+ if(current != null)
{
- current().setDone();
+ current.setDone();
}
final boolean advance = super.advance();
if(advance)
{
decrementQueued();
decrementCredit();
+ getSession().incrementIncomingBytes(-current.pending());
+ getSession().incrementIncomingDeliveries(-1);
+ if (getSession().getTransportSession().getIncomingWindowSize().equals(UnsignedInteger.ZERO)) {
+ modified();
+ }
}
return advance;
}
@@ -70,7 +77,14 @@ public class ReceiverImpl extends LinkIm
public int recv(final byte[] bytes, int offset, int size)
{
- return _current.recv(bytes, offset, size);
+ int consumed = _current.recv(bytes, offset, size);
+ if (consumed > 0) {
+ getSession().incrementIncomingBytes(-consumed);
+ if (getSession().getTransportSession().getIncomingWindowSize().equals(UnsignedInteger.ZERO)) {
+ modified();
+ }
+ }
+ return consumed;
}
public Iterator<Delivery> unsettled()
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java Thu Aug 1 13:58:54 2013
@@ -40,7 +40,6 @@ import org.apache.qpid.proton.amqp.secur
import org.apache.qpid.proton.amqp.security.SaslMechanisms;
import org.apache.qpid.proton.amqp.security.SaslResponse;
import org.apache.qpid.proton.codec.AMQPDefinedTypes;
-import org.apache.qpid.proton.codec.CompositeWritableBuffer;
import org.apache.qpid.proton.codec.DecoderImpl;
import org.apache.qpid.proton.codec.EncoderImpl;
import org.apache.qpid.proton.codec.WritableBuffer;
@@ -59,7 +58,7 @@ public class SaslImpl implements Sasl, S
private final ByteBuffer _inputBuffer;
private final ByteBuffer _outputBuffer;
- private final ByteBuffer _outputOverflowBuffer;
+ private final FrameWriter _frameWriter;
private ByteBuffer _pending;
@@ -92,10 +91,10 @@ public class SaslImpl implements Sasl, S
{
_inputBuffer = newWriteableBuffer(maxFrameSize);
_outputBuffer = newWriteableBuffer(maxFrameSize);
- _outputOverflowBuffer = newReadableBuffer(maxFrameSize);
AMQPDefinedTypes.registerAllTypes(_decoder,_encoder);
_frameParser = new SaslFrameParser(this, _decoder);
+ _frameWriter = new FrameWriter(_encoder, maxFrameSize, FrameWriter.SASL_FRAME_TYPE, null);
}
@Override
@@ -106,23 +105,8 @@ public class SaslImpl implements Sasl, S
private void writeSaslOutput()
{
- if(_outputOverflowBuffer.hasRemaining())
- {
- _outputBuffer.put(_outputOverflowBuffer);
- }
- if(!_outputOverflowBuffer.hasRemaining())
- {
- _outputOverflowBuffer.clear();
-
- CompositeWritableBuffer compositeOutputBuffer =
- new CompositeWritableBuffer(
- new WritableBuffer.ByteBufferWrapper(_outputBuffer),
- new WritableBuffer.ByteBufferWrapper(_outputOverflowBuffer));
-
- process(compositeOutputBuffer);
-
- _outputOverflowBuffer.flip();
- }
+ process();
+ _frameWriter.readBytes(_outputBuffer);
if(_logger.isLoggable(Level.FINER))
{
@@ -130,9 +114,9 @@ public class SaslImpl implements Sasl, S
}
}
- private void process(WritableBuffer buffer)
+ private void process()
{
- processHeader(buffer);
+ processHeader();
if(_role == Role.SERVER)
{
@@ -141,7 +125,7 @@ public class SaslImpl implements Sasl, S
SaslMechanisms mechanisms = new SaslMechanisms();
mechanisms.setSaslServerMechanisms(_mechanisms);
- writeFrame(buffer, mechanisms);
+ writeFrame(mechanisms);
_mechanismsSent = true;
_state = SaslState.PN_SASL_STEP;
}
@@ -150,7 +134,7 @@ public class SaslImpl implements Sasl, S
{
SaslChallenge challenge = new SaslChallenge();
challenge.setChallenge(getChallengeResponse());
- writeFrame(buffer, challenge);
+ writeFrame(challenge);
setChallengeResponse(null);
}
@@ -159,14 +143,14 @@ public class SaslImpl implements Sasl, S
org.apache.qpid.proton.amqp.security.SaslOutcome outcome =
new org.apache.qpid.proton.amqp.security.SaslOutcome();
outcome.setCode(SaslCode.values()[_outcome.getCode()]);
- writeFrame(buffer, outcome);
+ writeFrame(outcome);
}
}
else if(_role == Role.CLIENT)
{
if(getState() == SaslState.PN_SASL_IDLE && _chosenMechanism != null)
{
- processInit(buffer);
+ processInit();
_state = SaslState.PN_SASL_STEP;
//HACK: if we received an outcome before
@@ -179,32 +163,15 @@ public class SaslImpl implements Sasl, S
if(getState() == SaslState.PN_SASL_STEP && getChallengeResponse() != null)
{
- processResponse(buffer);
+ processResponse();
}
}
}
- private void writeFrame(WritableBuffer buffer, SaslFrameBody frameBody)
+ private void writeFrame(SaslFrameBody frameBody)
{
- int oldPosition = buffer.position();
- buffer.position(buffer.position()+8);
- _encoder.setByteBuffer(buffer);
- _encoder.writeObject(frameBody);
-
- int frameSize = buffer.position() - oldPosition;
- int limit = buffer.position();
- buffer.position(oldPosition);
- buffer.putInt(frameSize);
- buffer.put((byte) 2);
- buffer.put(SASL_FRAME_TYPE);
- buffer.putShort((short) 0);
- buffer.position(limit);
-
- if(_logger.isLoggable(Level.FINE))
- {
- _logger.fine(this + " wrote frame " + frameBody + " (" + frameSize + ") bytes");
- }
+ _frameWriter.writeFrame(frameBody);
}
@Override
@@ -231,12 +198,12 @@ public class SaslImpl implements Sasl, S
return size;
}
- final int processHeader(WritableBuffer outputBuffer)
+ final int processHeader()
{
if(!_headerWritten)
{
- outputBuffer.put(AmqpHeader.SASL_HEADER,0, AmqpHeader.SASL_HEADER.length);
+ _frameWriter.writeHeader(AmqpHeader.SASL_HEADER);
_headerWritten = true;
return AmqpHeader.SASL_HEADER.length;
@@ -437,15 +404,15 @@ public class SaslImpl implements Sasl, S
return outcome == SaslOutcome.PN_SASL_OK ? SaslState.PN_SASL_PASS : SaslState.PN_SASL_FAIL;
}
- private void processResponse(WritableBuffer buffer)
+ private void processResponse()
{
SaslResponse response = new SaslResponse();
response.setResponse(getChallengeResponse());
setChallengeResponse(null);
- writeFrame(buffer, response);
+ writeFrame(response);
}
- private void processInit(WritableBuffer buffer)
+ private void processInit()
{
SaslInit init = new SaslInit();
init.setHostname(_hostname);
@@ -456,7 +423,7 @@ public class SaslImpl implements Sasl, S
setChallengeResponse(null);
}
_initSent = true;
- writeFrame(buffer, init);
+ writeFrame(init);
}
@Override
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SenderImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SenderImpl.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SenderImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SenderImpl.java Thu Aug 1 13:58:54 2013
@@ -52,7 +52,11 @@ public class SenderImpl extends LinkImp
{
throw new IllegalArgumentException();//TODO.
}
- return current.send(bytes, offset, length);
+ int sent = current.send(bytes, offset, length);
+ if (sent > 0) {
+ getSession().incrementOutgoingBytes(sent);
+ }
+ return sent;
}
public void abort()
@@ -87,6 +91,7 @@ public class SenderImpl extends LinkImp
{
decrementCredit();
delivery.addToTransportWorkList();
+ getSession().incrementOutgoingDeliveries(1);
}
return advance;
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java Thu Aug 1 13:58:54 2013
@@ -32,6 +32,11 @@ public class SessionImpl extends Endpoin
private Map<String, SenderImpl> _senders = new LinkedHashMap<String, SenderImpl>();
private Map<String, ReceiverImpl> _receivers = new LinkedHashMap<String, ReceiverImpl>();
private TransportSession _transportSession;
+ private int _incomingCapacity = 1024*1024;
+ private int _incomingBytes = 0;
+ private int _outgoingBytes = 0;
+ private int _incomingDeliveries = 0;
+ private int _outgoingDeliveries = 0;
private LinkNode<SessionImpl> _node;
@@ -141,8 +146,53 @@ public class SessionImpl extends Endpoin
_receivers.remove(receiver.getName());
}
- public boolean clearIncomingWindowResize()
+ @Override
+ public int getIncomingCapacity()
+ {
+ return _incomingCapacity;
+ }
+
+ @Override
+ public void setIncomingCapacity(int capacity)
{
- return getTransportSession().clearIncomingWindowResize();
+ _incomingCapacity = capacity;
}
+
+ @Override
+ public int getIncomingBytes()
+ {
+ return _incomingBytes;
+ }
+
+ void incrementIncomingBytes(int delta)
+ {
+ _incomingBytes += delta;
+ }
+
+ @Override
+ public int getOutgoingBytes()
+ {
+ return _outgoingBytes;
+ }
+
+ void incrementOutgoingBytes(int delta)
+ {
+ _outgoingBytes += delta;
+ }
+
+ void incrementIncomingDeliveries(int delta)
+ {
+ _incomingDeliveries += delta;
+ }
+
+ int getOutgoingDeliveries()
+ {
+ return _outgoingDeliveries;
+ }
+
+ void incrementOutgoingDeliveries(int delta)
+ {
+ _outgoingDeliveries += delta;
+ }
+
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java Thu Aug 1 13:58:54 2013
@@ -42,7 +42,6 @@ import org.apache.qpid.proton.amqp.trans
import org.apache.qpid.proton.amqp.transport.Role;
import org.apache.qpid.proton.amqp.transport.Transfer;
import org.apache.qpid.proton.codec.AMQPDefinedTypes;
-import org.apache.qpid.proton.codec.CompositeWritableBuffer;
import org.apache.qpid.proton.codec.DecoderImpl;
import org.apache.qpid.proton.codec.EncoderImpl;
import org.apache.qpid.proton.codec.WritableBuffer;
@@ -92,8 +91,9 @@ public class TransportImpl extends Endpo
private EncoderImpl _encoder = new EncoderImpl(_decoder);
private int _maxFrameSize = DEFAULT_MAX_FRAME_SIZE;
+ private int _remoteMaxFrameSize = 512;
- private ByteBuffer _outputOverflowBuffer;
+ private final FrameWriter _frameWriter;
private boolean _closeReceived;
private Open _open;
@@ -131,6 +131,9 @@ public class TransportImpl extends Endpo
_engineLogger = engineLogger;
_maxFrameSize = maxFrameSize;
+ _frameWriter = new FrameWriter(_encoder, _remoteMaxFrameSize,
+ FrameWriter.AMQP_FRAME_TYPE,
+ _protocolTracer);
}
@@ -158,7 +161,6 @@ public class TransportImpl extends Endpo
_frameParser = new FrameParser(_frameHandler , _decoder, _maxFrameSize);
_inputProcessor = _frameParser;
_outputProcessor = new TransportOutputAdaptor(this, _maxFrameSize);
- _outputOverflowBuffer = newReadableBuffer(_maxFrameSize);
}
}
@@ -168,6 +170,12 @@ public class TransportImpl extends Endpo
return _maxFrameSize;
}
+ @Override
+ public int getRemoteMaxFrameSize()
+ {
+ return _remoteMaxFrameSize;
+ }
+
/**
* TODO propagate the new value to {@link #_outputProcessor} etc
*/
@@ -247,35 +255,22 @@ public class TransportImpl extends Endpo
@Override
public void writeInto(ByteBuffer outputBuffer)
{
- if(_outputOverflowBuffer.hasRemaining())
- {
- pour(_outputOverflowBuffer, outputBuffer);
- }
- if(!_outputOverflowBuffer.hasRemaining())
- {
- _outputOverflowBuffer.clear();
+ processHeader();
+ processOpen();
+ processBegin();
+ processAttach();
+ processReceiverFlow();
+ // we process transport work twice intentionally, the first
+ // pass may end up settling deliveries that the second pass
+ // can clean up
+ processTransportWork();
+ processTransportWork();
+ processSenderFlow();
+ processDetach();
+ processEnd();
+ processClose();
- CompositeWritableBuffer compositeOutputBuffer =
- new CompositeWritableBuffer(
- new WritableBuffer.ByteBufferWrapper(outputBuffer),
- new WritableBuffer.ByteBufferWrapper(_outputOverflowBuffer));
-
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processHeader(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processOpen(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processBegin(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processAttach(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processReceiverDisposition(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processReceiverFlow(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processMessageData(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processSenderDisposition(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processSenderFlow(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processDetach(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processEnd(compositeOutputBuffer);
- if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processClose(compositeOutputBuffer);
- }}}}}}}}}}}}
-
- _outputOverflowBuffer.flip();
- }
+ _frameWriter.readBytes(outputBuffer);
}
@Override
@@ -284,7 +279,7 @@ public class TransportImpl extends Endpo
if(_sasl == null)
{
init();
- _sasl = new SaslImpl(_maxFrameSize);
+ _sasl = new SaslImpl(_remoteMaxFrameSize);
TransportWrapper transportWrapper = _sasl.wrap(_inputProcessor, _outputProcessor);
_inputProcessor = transportWrapper;
_outputProcessor = transportWrapper;
@@ -320,12 +315,12 @@ public class TransportImpl extends Endpo
return ssl(sslDomain, null);
}
- private void processDetach(WritableBuffer buffer)
+ private void processDetach()
{
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null && buffer.remaining() >= _maxFrameSize)
+ while(endpoint != null)
{
if(endpoint instanceof LinkImpl)
@@ -361,7 +356,7 @@ public class TransportImpl extends Endpo
}
- writeFrame(buffer, transportSession.getLocalChannel(), detach, null, null);
+ writeFrame(transportSession.getLocalChannel(), detach, null, null);
endpoint.clearModified();
// TODO - temporary hack for PROTON-154, this line should be removed and replaced
@@ -376,12 +371,29 @@ public class TransportImpl extends Endpo
}
}
- private void processSenderFlow(WritableBuffer buffer)
+ private void writeFlow(TransportSession ssn, TransportLink link)
+ {
+ Flow flow = new Flow();
+ flow.setNextIncomingId(ssn.getNextIncomingId());
+ flow.setNextOutgoingId(ssn.getNextOutgoingId());
+ ssn.updateWindows();
+ flow.setIncomingWindow(ssn.getIncomingWindowSize());
+ flow.setOutgoingWindow(ssn.getOutgoingWindowSize());
+ if (link != null) {
+ flow.setHandle(link.getLocalHandle());
+ flow.setDeliveryCount(link.getDeliveryCount());
+ flow.setLinkCredit(link.getLinkCredit());
+ flow.setDrain(link.getLink().getDrain());
+ }
+ writeFrame(ssn.getLocalChannel(), flow, null, null);
+ }
+
+ private void processSenderFlow()
{
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null && buffer.remaining() >= _maxFrameSize)
+ while(endpoint != null)
{
if(endpoint instanceof SenderImpl)
@@ -396,16 +408,8 @@ public class TransportImpl extends Endpo
transportLink.setDeliveryCount(transportLink.getDeliveryCount().add(credits));
transportLink.setLinkCredit(UnsignedInteger.ZERO);
- Flow flow = new Flow();
- flow.setHandle(transportLink.getLocalHandle());
- flow.setNextIncomingId(transportSession.getNextIncomingId());
- flow.setIncomingWindow(transportSession.getIncomingWindowSize());
- flow.setOutgoingWindow(transportSession.getOutgoingWindowSize());
- flow.setDeliveryCount(transportLink.getDeliveryCount());
- flow.setLinkCredit(transportLink.getLinkCredit());
- flow.setDrain(sender.getDrain());
- flow.setNextOutgoingId(transportSession.getNextOutgoingId());
- writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
+ writeFlow(transportSession, transportLink);
+
endpoint.clearModified();
}
@@ -416,162 +420,165 @@ public class TransportImpl extends Endpo
}
}
- private void processSenderDisposition(WritableBuffer buffer)
+ private void dumpQueue(String msg)
+ {
+ System.out.print(" " + msg + "{");
+ DeliveryImpl dlv = _connectionEndpoint.getTransportWorkHead();
+ while (dlv != null) {
+ System.out.print(new Binary(dlv.getTag()) + ", ");
+ dlv = dlv.getTransportWorkNext();
+ }
+ System.out.println("}");
+ }
+
+ private void processTransportWork()
{
if(_connectionEndpoint != null)
{
DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
- while(delivery != null && buffer.remaining() >= _maxFrameSize )
+
+ while(delivery != null)
{
- if((delivery.getLink() instanceof SenderImpl) && delivery.getTransportDelivery() != null)
- {
- TransportDelivery transportDelivery = delivery.getTransportDelivery();
- Disposition disposition = new Disposition();
- disposition.setFirst(transportDelivery.getDeliveryId());
- disposition.setLast(transportDelivery.getDeliveryId());
- disposition.setRole(Role.SENDER);
- disposition.setSettled(delivery.isSettled());
- if(delivery.isSettled())
- {
- transportDelivery.settled();
+ LinkImpl link = delivery.getLink();
+ if (link instanceof SenderImpl) {
+ if (processTransportWorkSender(delivery, (SenderImpl) link)) {
+ delivery = delivery.clearTransportWork();
+ } else {
+ delivery = delivery.getTransportWorkNext();
+ }
+ } else {
+ if (processTransportWorkReceiver(delivery, (ReceiverImpl) link)) {
+ delivery = delivery.clearTransportWork();
+ } else {
+ delivery = delivery.getTransportWorkNext();
}
- disposition.setState(delivery.getLocalState());
-
- writeFrame(buffer,
- delivery.getLink().getSession().getTransportSession().getLocalChannel(),
- disposition, null, null);
-
- delivery = delivery.clearTransportWork();
- }
- else
- {
- delivery = delivery.getTransportWorkNext();
}
}
}
}
- private void processMessageData(WritableBuffer buffer)
+ private boolean processTransportWorkSender(DeliveryImpl delivery,
+ SenderImpl snd)
{
- if(_connectionEndpoint != null)
- {
- DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
-
- while(delivery != null && buffer.remaining() >= _maxFrameSize)
- {
- if((delivery.getLink() instanceof SenderImpl) && !(delivery.isDone() && delivery.getDataLength() == 0)
- && delivery.getLink().getSession().getTransportSession().hasOutgoingCredit() &&
- delivery.getLink().getTransportLink().hasCredit())
- {
- SenderImpl sender = (SenderImpl) delivery.getLink();
-
- sender.decrementQueued();
-
+ TransportLink<SenderImpl> tpLink = snd.getTransportLink();
+ SessionImpl session = snd.getSession();
+ TransportSession tpSession = session.getTransportSession();
- TransportLink<?> transportLink = sender.getTransportLink();
+ boolean wasDone = delivery.isDone();
- UnsignedInteger deliveryId = transportLink.getDeliveryCount();
- TransportDelivery transportDelivery = new TransportDelivery(deliveryId, delivery, transportLink);
- delivery.setTransportDelivery(transportDelivery);
-
- final Transfer transfer = new Transfer();
- transfer.setDeliveryId(deliveryId);
- transfer.setDeliveryTag(new Binary(delivery.getTag()));
- transfer.setHandle(transportLink.getLocalHandle());
-
- if(delivery.isSettled())
- {
- transfer.setSettled(Boolean.TRUE);
- }
- else
- {
- sender.getSession().getTransportSession().addUnsettledOutgoing(deliveryId, delivery);
- }
+ if(!delivery.isDone() &&
+ (delivery.getDataLength() > 0 || delivery != snd.current()) &&
+ tpSession.hasOutgoingCredit() && tpLink.hasCredit())
+ {
+ // XXX: this looks wrong, need to test with multiple links on a session to verify
+ UnsignedInteger deliveryId = tpLink.getDeliveryCount();
+ TransportDelivery tpDelivery = new TransportDelivery(deliveryId, delivery, tpLink);
+ delivery.setTransportDelivery(tpDelivery);
- if(delivery.getLink().current() == delivery)
- {
- transfer.setMore(true);
- }
+ final Transfer transfer = new Transfer();
+ transfer.setDeliveryId(deliveryId);
+ transfer.setDeliveryTag(new Binary(delivery.getTag()));
+ transfer.setHandle(tpLink.getLocalHandle());
- transfer.setMessageFormat(UnsignedInteger.ZERO);
+ if(delivery.isSettled())
+ {
+ transfer.setSettled(Boolean.TRUE);
+ }
+ else
+ {
+ tpSession.addUnsettledOutgoing(deliveryId, delivery);
+ }
- // TODO - large frames
- ByteBuffer payload = delivery.getData() == null ? null : ByteBuffer.wrap(delivery.getData(), delivery.getDataOffset(), delivery.getDataLength());
+ if(snd.current() == delivery)
+ {
+ transfer.setMore(true);
+ }
- writeFrame(buffer,
- sender.getSession().getTransportSession().getLocalChannel(),
- transfer, payload, new PartialTransfer(transfer));
- sender.getSession().getTransportSession().incrementOutgoingId();
+ transfer.setMessageFormat(UnsignedInteger.ZERO);
- if(payload == null || !payload.hasRemaining())
- {
- delivery.setData(null);
- delivery.setDataLength(0);
- delivery.setDone();
+ // TODO - large frames
+ ByteBuffer payload = delivery.getData() == null ? null :
+ ByteBuffer.wrap(delivery.getData(), delivery.getDataOffset(),
+ delivery.getDataLength());
- if(delivery.getLink().current() != delivery)
- {
- transportLink.setDeliveryCount(transportLink.getDeliveryCount().add(UnsignedInteger.ONE));
- transportLink.setLinkCredit(transportLink.getLinkCredit().subtract(UnsignedInteger.ONE));
- }
+ writeFrame(tpSession.getLocalChannel(), transfer, payload,
+ new PartialTransfer(transfer));
+ tpSession.incrementOutgoingId();
+ tpSession.decrementRemoteIncomingWindow();
- delivery = delivery.clearTransportWork();
+ if(payload == null || !payload.hasRemaining())
+ {
+ session.incrementOutgoingBytes(-delivery.pending());
+ delivery.setData(null);
+ delivery.setDataLength(0);
- }
- else
- {
- delivery.setDataOffset(delivery.getDataOffset()+delivery.getDataLength()-payload.remaining());
- delivery.setDataLength(payload.remaining());
- }
- }
- else
- {
- delivery = delivery.getTransportWorkNext();
+ if (!transfer.getMore()) {
+ delivery.setDone();
+ tpLink.setDeliveryCount(tpLink.getDeliveryCount().add(UnsignedInteger.ONE));
+ tpLink.setLinkCredit(tpLink.getLinkCredit().subtract(UnsignedInteger.ONE));
+ session.incrementOutgoingDeliveries(-1);
+ snd.decrementQueued();
}
}
+ else
+ {
+ int delta = delivery.getDataLength() - payload.remaining();
+ delivery.setDataOffset(delivery.getDataOffset() + delta);
+ delivery.setDataLength(payload.remaining());
+ session.incrementOutgoingBytes(-delta);
+ }
}
- }
- private void processReceiverDisposition(WritableBuffer buffer)
- {
- if(_connectionEndpoint != null)
+ if(wasDone)
{
- DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
- while(delivery != null && buffer.remaining() >= _maxFrameSize)
+ TransportDelivery tpDelivery = delivery.getTransportDelivery();
+ Disposition disposition = new Disposition();
+ disposition.setFirst(tpDelivery.getDeliveryId());
+ disposition.setLast(tpDelivery.getDeliveryId());
+ disposition.setRole(Role.SENDER);
+ disposition.setSettled(delivery.isSettled());
+ if(delivery.isSettled())
{
- if((delivery.getLink() instanceof ReceiverImpl))
- {
- TransportDelivery transportDelivery = delivery.getTransportDelivery();
- Disposition disposition = new Disposition();
- disposition.setFirst(transportDelivery.getDeliveryId());
- disposition.setLast(transportDelivery.getDeliveryId());
- disposition.setRole(Role.RECEIVER);
- disposition.setSettled(delivery.isSettled());
-
- disposition.setState(delivery.getLocalState());
- writeFrame(buffer,
- delivery.getLink().getSession().getTransportSession().getLocalChannel(),
- disposition, null, null);
- if(delivery.isSettled())
- {
- transportDelivery.settled();
- }
- delivery = delivery.clearTransportWork();
- }
- else
- {
- delivery = delivery.getTransportWorkNext();
- }
+ tpDelivery.settled();
}
+ disposition.setState(delivery.getLocalState());
+
+ writeFrame(tpSession.getLocalChannel(), disposition, null,
+ null);
+ }
+
+ return delivery.isDone();
+ }
+
+ private boolean processTransportWorkReceiver(DeliveryImpl delivery,
+ ReceiverImpl rcv)
+ {
+ TransportDelivery tpDelivery = delivery.getTransportDelivery();
+ SessionImpl session = rcv.getSession();
+ TransportSession tpSession = session.getTransportSession();
+
+ Disposition disposition = new Disposition();
+ disposition.setFirst(tpDelivery.getDeliveryId());
+ disposition.setLast(tpDelivery.getDeliveryId());
+ disposition.setRole(Role.RECEIVER);
+ disposition.setSettled(delivery.isSettled());
+
+ disposition.setState(delivery.getLocalState());
+ writeFrame(tpSession.getLocalChannel(), disposition, null,
+ null);
+ if(delivery.isSettled())
+ {
+ tpDelivery.settled();
}
+ return true;
}
- private void processReceiverFlow(WritableBuffer buffer)
+ private void processReceiverFlow()
{
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null && buffer.remaining() >= _maxFrameSize)
+ while(endpoint != null)
{
if(endpoint instanceof ReceiverImpl)
{
@@ -582,20 +589,11 @@ public class TransportImpl extends Endpo
if(receiver.getLocalState() == EndpointState.ACTIVE)
{
int credits = receiver.clearUnsentCredits();
- transportSession.getSession().clearIncomingWindowResize();
- if(credits != 0 || receiver.getDrain())
+ if(credits != 0 || receiver.getDrain() ||
+ transportSession.getIncomingWindowSize().equals(UnsignedInteger.ZERO))
{
transportLink.addCredit(credits);
- Flow flow = new Flow();
- flow.setHandle(transportLink.getLocalHandle());
- flow.setNextIncomingId(transportSession.getNextIncomingId());
- flow.setIncomingWindow(transportSession.getIncomingWindowSize());
- flow.setOutgoingWindow(transportSession.getOutgoingWindowSize());
- flow.setDeliveryCount(transportLink.getDeliveryCount());
- flow.setLinkCredit(transportLink.getLinkCredit());
- flow.setDrain(receiver.getDrain());
- flow.setNextOutgoingId(transportSession.getNextOutgoingId());
- writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
+ writeFlow(transportSession, transportLink);
if(receiver.getLocalState() == EndpointState.ACTIVE)
{
endpoint.clearModified();
@@ -606,7 +604,7 @@ public class TransportImpl extends Endpo
endpoint = endpoint.transportNext();
}
endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null && buffer.remaining() >= _maxFrameSize)
+ while(endpoint != null)
{
if(endpoint instanceof SessionImpl)
{
@@ -616,15 +614,9 @@ public class TransportImpl extends Endpo
if(session.getLocalState() == EndpointState.ACTIVE)
{
- boolean windowResized = session.clearIncomingWindowResize();
- if(windowResized)
+ if(transportSession.getIncomingWindowSize().equals(UnsignedInteger.ZERO))
{
- Flow flow = new Flow();
- flow.setIncomingWindow(transportSession.getIncomingWindowSize());
- flow.setOutgoingWindow(transportSession.getOutgoingWindowSize());
- flow.setNextOutgoingId(transportSession.getNextOutgoingId());
- flow.setNextIncomingId(transportSession.getNextIncomingId());
- writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
+ writeFlow(transportSession, null);
}
}
}
@@ -633,13 +625,13 @@ public class TransportImpl extends Endpo
}
}
- private void processAttach(WritableBuffer buffer)
+ private void processAttach()
{
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null && buffer.remaining() >= _maxFrameSize)
+ while(endpoint != null)
{
if(endpoint instanceof LinkImpl)
{
@@ -693,7 +685,7 @@ public class TransportImpl extends Endpo
attach.setInitialDeliveryCount(UnsignedInteger.ZERO);
}
- writeFrame(buffer, transportSession.getLocalChannel(), attach, null, null);
+ writeFrame(transportSession.getLocalChannel(), attach, null, null);
transportLink.sentAttach();
if(link.getLocalState() == EndpointState.ACTIVE && (link instanceof SenderImpl || !link.hasCredit()))
{
@@ -707,16 +699,16 @@ public class TransportImpl extends Endpo
}
}
- private void processHeader(WritableBuffer buffer)
+ private void processHeader()
{
if(!_headerWritten)
{
- buffer.put(AmqpHeader.HEADER, 0, AmqpHeader.HEADER.length);
+ _frameWriter.writeHeader(AmqpHeader.HEADER);
_headerWritten = true;
}
}
- private void processOpen(WritableBuffer buffer)
+ private void processOpen()
{
if(_connectionEndpoint != null && _connectionEndpoint.getLocalState() != EndpointState.UNINITIALIZED && !_isOpenSent)
{
@@ -726,21 +718,24 @@ public class TransportImpl extends Endpo
open.setDesiredCapabilities(_connectionEndpoint.getDesiredCapabilities());
open.setOfferedCapabilities(_connectionEndpoint.getOfferedCapabilities());
open.setProperties(_connectionEndpoint.getProperties());
+ if (_maxFrameSize > 0) {
+ open.setMaxFrameSize(UnsignedInteger.valueOf(_maxFrameSize));
+ }
// TODO - populate;
_isOpenSent = true;
- writeFrame(buffer, 0, open, null, null);
+ writeFrame(0, open, null, null);
}
}
- private void processBegin(WritableBuffer buffer)
+ private void processBegin()
{
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null && buffer.remaining() >= _maxFrameSize)
+ while(endpoint != null)
{
if(endpoint instanceof SessionImpl)
{
@@ -760,7 +755,7 @@ public class TransportImpl extends Endpo
begin.setOutgoingWindow(transportSession.getOutgoingWindowSize());
begin.setNextOutgoingId(transportSession.getNextOutgoingId());
- writeFrame(buffer, channelId, begin, null, null);
+ writeFrame(channelId, begin, null, null);
transportSession.sentBegin();
if(session.getLocalState() == EndpointState.ACTIVE)
{
@@ -778,7 +773,7 @@ public class TransportImpl extends Endpo
TransportSession transportSession = _transportSessionState.get(session);
if(transportSession == null)
{
- transportSession = new TransportSession(session);
+ transportSession = new TransportSession(this, session);
session.setTransportSession(transportSession);
_transportSessionState.put(session, transportSession);
}
@@ -818,12 +813,12 @@ public class TransportImpl extends Endpo
return channel;
}
- private void processEnd(WritableBuffer buffer)
+ private void processEnd()
{
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null && buffer.remaining() >= _maxFrameSize)
+ while(endpoint != null)
{
SessionImpl session;
TransportSession transportSession;
@@ -841,7 +836,7 @@ public class TransportImpl extends Endpo
end.setError(localError);
}
- writeFrame(buffer, channel, end, null, null);
+ writeFrame(channel, end, null, null);
endpoint.clearModified();
}
@@ -874,7 +869,7 @@ public class TransportImpl extends Endpo
return false;
}
- private void processClose(WritableBuffer buffer)
+ private void processClose()
{
if(_connectionEndpoint != null && _connectionEndpoint.getLocalState() == EndpointState.CLOSED && !_isCloseSent)
{
@@ -890,62 +885,15 @@ public class TransportImpl extends Endpo
_isCloseSent = true;
- writeFrame(buffer, 0, close, null, null);
+ writeFrame(0, close, null, null);
}
}
}
- private void writeFrame(WritableBuffer buffer,
- int channel,
- FrameBody frameBody,
- ByteBuffer payload,
- Runnable onPayloadTooLarge)
+ private void writeFrame(int channel, FrameBody frameBody,
+ ByteBuffer payload, Runnable onPayloadTooLarge)
{
- int oldPosition = buffer.position();
- buffer.position(oldPosition+8);
- _encoder.setByteBuffer(buffer);
-
- _encoder.writeObject(frameBody);
-
- if(payload != null && (payload.remaining() + buffer.position() - oldPosition) > _maxFrameSize)
- {
- if(onPayloadTooLarge != null)
- {
- onPayloadTooLarge.run();
- }
- buffer.position(oldPosition+8);
- _encoder.writeObject(frameBody);
- }
-
- ByteBuffer originalPayload = null;
- if( payload!=null )
- {
- originalPayload = payload.duplicate();
- }
- TransportFrame frame = new TransportFrame(channel, frameBody, Binary.create(originalPayload));
- log(OUTGOING, frame);
-
- if( _protocolTracer!=null )
- {
- _protocolTracer.sentFrame(frame);
- }
-
- int payloadSize = Math.min(payload == null ? 0 : payload.remaining(), _maxFrameSize - (buffer.position() - oldPosition));
- if(payloadSize > 0)
- {
- int oldLimit = payload.limit();
- payload.limit(payload.position() + payloadSize);
- buffer.put(payload);
- payload.limit(oldLimit);
- }
- int frameSize = buffer.position() - oldPosition;
- int limit = buffer.position();
- buffer.position(oldPosition);
- buffer.putInt(frameSize);
- buffer.put((byte) 2);
- buffer.put(AMQP_FRAME_TYPE);
- buffer.putShort((short) channel);
- buffer.position(limit);
+ _frameWriter.writeFrame(channel, frameBody, payload, onPayloadTooLarge);
}
//==================================================================================================================
@@ -979,9 +927,10 @@ public class TransportImpl extends Endpo
_open = open;
}
- if(open.getMaxFrameSize().longValue() > 0 && open.getMaxFrameSize().longValue() < _maxFrameSize)
+ if(open.getMaxFrameSize().longValue() > 0)
{
- _maxFrameSize = (int) open.getMaxFrameSize().longValue();
+ _remoteMaxFrameSize = (int) open.getMaxFrameSize().longValue();
+ _frameWriter.setMaxFrameSize(_remoteMaxFrameSize);
}
}
@@ -1192,7 +1141,7 @@ public class TransportImpl extends Endpo
throw new IllegalStateException("Transport cannot accept frame: " + frame);
}
- log(INCOMING, frame);
+ log(this, INCOMING, frame);
if( _protocolTracer != null )
{
@@ -1297,19 +1246,35 @@ public class TransportImpl extends Endpo
_frameHandler = frameHandler;
}
- private static String INCOMING = "<-";
- private static String OUTGOING = "->";
+ static String INCOMING = "<-";
+ static String OUTGOING = "->";
- private void log(String event, TransportFrame frame)
+ private static final boolean getBooleanEnv(String name)
{
- /*StringBuilder msg = new StringBuilder();
- msg.append("[").append(System.identityHashCode(this)).append(":")
- .append(frame.getChannel()).append("]");
- msg.append(" ").append(event).append(" ").append(frame.getBody());
- if (frame.getPayload() != null) {
- msg.append(" \"").append(frame.getPayload()).append("\"");
+ String value = System.getenv(name);
+ return "true".equalsIgnoreCase(value) ||
+ "1".equals(value) ||
+ "yes".equalsIgnoreCase(value);
+ }
+
+ private static final boolean ENABLED = getBooleanEnv("PN_TRACE_FRM");
+
+ static void log(Object ctx, String event, TransportFrame frame)
+ {
+ if (ENABLED) {
+ StringBuilder msg = new StringBuilder();
+ msg.append("[").append(System.identityHashCode(ctx)).append(":")
+ .append(frame.getChannel()).append("]");
+ msg.append(" ").append(event).append(" ").append(frame.getBody());
+ if (frame.getPayload() != null) {
+ String payload = frame.getPayload().toString();
+ if (payload.length() > 80) {
+ payload = payload.substring(0, 80) + "(" + payload.length() + ")";
+ }
+ msg.append(" \"").append(payload).append("\"");
+ }
+ System.out.println(msg.toString());
}
- System.out.println(msg.toString());*/
}
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java Thu Aug 1 13:58:54 2013
@@ -33,7 +33,11 @@ class TransportOutputAdaptor implements
TransportOutputAdaptor(TransportOutputWriter transportOutputWriter, int maxFrameSize)
{
_transportOutputWriter = transportOutputWriter;
- _outputBuffer = newWriteableBuffer(maxFrameSize);
+ if (maxFrameSize > 0) {
+ _outputBuffer = newWriteableBuffer(maxFrameSize);
+ } else {
+ _outputBuffer = newWriteableBuffer(4*1024);
+ }
}
@Override
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java Thu Aug 1 13:58:54 2013
@@ -32,13 +32,14 @@ import org.apache.qpid.proton.amqp.trans
class TransportSession
{
+ private final TransportImpl _transport;
private final SessionImpl _session;
private int _localChannel = -1;
private int _remoteChannel = -1;
private boolean _openSent;
private UnsignedInteger _handleMax = UnsignedInteger.valueOf(1024);
- private UnsignedInteger _incomingWindowSize = UnsignedInteger.valueOf(TransportImpl.SESSION_WINDOW);
- private UnsignedInteger _outgoingWindowSize = UnsignedInteger.valueOf(TransportImpl.SESSION_WINDOW);
+ private UnsignedInteger _incomingWindowSize = UnsignedInteger.ZERO;
+ private UnsignedInteger _outgoingWindowSize = UnsignedInteger.ZERO;
private UnsignedInteger _nextOutgoingId = UnsignedInteger.ONE;
private UnsignedInteger _nextIncomingId = null;
@@ -57,13 +58,12 @@ class TransportSession
private Map<UnsignedInteger, DeliveryImpl>
_unsettledOutgoingDeliveriesById = new HashMap<UnsignedInteger, DeliveryImpl>();
private int _unsettledIncomingSize;
- private boolean _incomingWindowSizeChange;
- private boolean _outgoingWindowSizeChange;
private boolean _endReceived;
private boolean _beginSent;
- TransportSession(SessionImpl session)
+ TransportSession(TransportImpl transport, SessionImpl session)
{
+ _transport = transport;
_session = session;
}
@@ -133,23 +133,39 @@ class TransportSession
return _incomingWindowSize;
}
- public UnsignedInteger getOutgoingWindowSize()
- {
- return _outgoingWindowSize;
- }
-
- public void incrementOutgoingWindow()
+ public void updateWindows()
{
- _outgoingWindowSize = _outgoingWindowSize.add(UnsignedInteger.ONE);
+ // incoming window
+ int size = _transport.getMaxFrameSize();
+ if (size <= 0) {
+ _incomingWindowSize = UnsignedInteger.valueOf(2147483647); // biggest legal value
+ } else {
+ _incomingWindowSize = UnsignedInteger.valueOf((_session.getIncomingCapacity() - _session.getIncomingBytes())/size);
+ }
+
+ // outgoing window
+ int outgoingDeliveries = _session.getOutgoingDeliveries();
+ if (size <= 0) {
+ _outgoingWindowSize = UnsignedInteger.valueOf(outgoingDeliveries);
+ } else {
+ int outgoingBytes = _session.getOutgoingBytes();
+ int frames = outgoingBytes/size;
+ if (outgoingBytes % size > 0) {
+ frames++;
+ }
+ if (frames > outgoingDeliveries) {
+ _outgoingWindowSize = UnsignedInteger.valueOf(frames);
+ } else {
+ _outgoingWindowSize = UnsignedInteger.valueOf(outgoingDeliveries);
+ }
+ }
}
- public void decrementOutgoingWindow()
+ public UnsignedInteger getOutgoingWindowSize()
{
- _outgoingWindowSize = _outgoingWindowSize.subtract(UnsignedInteger.ONE);
+ return _outgoingWindowSize;
}
-
-
public UnsignedInteger getNextOutgoingId()
{
return _nextOutgoingId;
@@ -232,6 +248,7 @@ class TransportSession
TransportDelivery transportDelivery = new TransportDelivery(_currentDeliveryId, delivery, transportReceiver);
delivery.setTransportDelivery(transportDelivery);
_unsettledIncomingDeliveriesById.put(_currentDeliveryId, delivery);
+ getSession().incrementIncomingDeliveries(1);
}
if( transfer.getState()!=null )
{
@@ -256,6 +273,7 @@ class TransportSession
delivery.setDataOffset(0);
delivery.setDataLength(data.length);
}
+ getSession().incrementIncomingBytes(payload.getLength());
}
delivery.addIOWork();
@@ -263,7 +281,6 @@ class TransportSession
if(!(transfer.getMore() || transfer.getAborted()))
{
delivery.setComplete();
- _incomingWindowSize = _incomingWindowSize.subtract(UnsignedInteger.ONE);
delivery.getLink().getTransportLink().decrementLinkCredit();
delivery.getLink().getTransportLink().incrementDeliveryCount();
}
@@ -272,6 +289,12 @@ class TransportSession
delivery.setRemoteSettled(true);
}
+ _incomingWindowSize = _incomingWindowSize.subtract(UnsignedInteger.ONE);
+
+ // this will cause a flow to happen
+ if (_incomingWindowSize.equals(UnsignedInteger.ZERO)) {
+ delivery.getLink().modified();
+ }
}
public void freeLocalChannel()
@@ -284,6 +307,11 @@ class TransportSession
_remoteIncomingWindow = incomingWindow;
}
+ void decrementRemoteIncomingWindow()
+ {
+ _remoteIncomingWindow = _remoteIncomingWindow.subtract(UnsignedInteger.ONE);
+ }
+
private void setRemoteOutgoingWindow(UnsignedInteger outgoingWindow)
{
_remoteOutgoingWindow = outgoingWindow;
@@ -291,13 +319,20 @@ class TransportSession
void handleFlow(Flow flow)
{
- setRemoteIncomingWindow(flow.getIncomingWindow());
- setRemoteOutgoingWindow(flow.getOutgoingWindow());
- if(flow.getNextIncomingId() != null)
+ UnsignedInteger inext = flow.getNextIncomingId();
+ UnsignedInteger iwin = flow.getIncomingWindow();
+
+ if(inext != null)
{
- setRemoteNextIncomingId(flow.getNextIncomingId());
+ setRemoteNextIncomingId(inext);
+ setRemoteIncomingWindow(inext.add(iwin).subtract(_nextOutgoingId));
+ }
+ else
+ {
+ setRemoteIncomingWindow(iwin);
}
setRemoteNextOutgoingId(flow.getNextOutgoingId());
+ setRemoteOutgoingWindow(flow.getOutgoingWindow());
if(flow.getHandle() != null)
{
@@ -306,7 +341,6 @@ class TransportSession
}
-
}
private void setRemoteNextOutgoingId(UnsignedInteger nextOutgoingId)
@@ -351,15 +385,12 @@ class TransportSession
void addUnsettledOutgoing(UnsignedInteger deliveryId, DeliveryImpl delivery)
{
_unsettledOutgoingDeliveriesById.put(deliveryId, delivery);
- _outgoingWindowSize = _outgoingWindowSize.subtract(UnsignedInteger.valueOf(delivery.getTransportDelivery().getSessionSize()));
}
public boolean hasOutgoingCredit()
{
- return _remoteIncomingWindow == null
- ? false
- : _remoteIncomingWindow.add(_remoteNextIncomingId).compareTo(_nextOutgoingId)>0
- && _outgoingWindowSize.compareTo(UnsignedInteger.ZERO)>0;
+ return _remoteIncomingWindow == null ? false
+ : _remoteIncomingWindow.compareTo(UnsignedInteger.ZERO)>0;
}
@@ -373,40 +404,15 @@ class TransportSession
if(transportDelivery.getTransportLink().getLink() instanceof ReceiverImpl)
{
_unsettledIncomingDeliveriesById.remove(transportDelivery.getDeliveryId());
- _incomingWindowSize = _incomingWindowSize.add( UnsignedInteger.valueOf(transportDelivery.getSessionSize()));
- _incomingWindowSizeChange = true;
getSession().modified();
}
else
{
_unsettledOutgoingDeliveriesById.remove(transportDelivery.getDeliveryId());
- _outgoingWindowSize = _outgoingWindowSize.add(UnsignedInteger.valueOf(transportDelivery.getSessionSize()));
- _outgoingWindowSizeChange = true;
getSession().modified();
}
}
- public boolean clearIncomingWindowResize()
- {
- if(_incomingWindowSizeChange)
- {
- _incomingWindowSizeChange = false;
- return true;
- }
- return false;
- }
-
- public boolean clearOutgoingWindowResize()
- {
- if(_outgoingWindowSizeChange)
- {
- _outgoingWindowSizeChange = false;
- return true;
- }
- return false;
- }
-
-
public UnsignedInteger getNextIncomingId()
{
return _nextIncomingId;
Modified: qpid/proton/trunk/tests/python/proton_tests/engine.py
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/tests/python/proton_tests/engine.py?rev=1509238&r1=1509237&r2=1509238&view=diff
==============================================================================
--- qpid/proton/trunk/tests/python/proton_tests/engine.py (original)
+++ qpid/proton/trunk/tests/python/proton_tests/engine.py Thu Aug 1 13:58:54 2013
@@ -1110,7 +1110,7 @@ class CreditTest(Test):
assert self.rcv.queued == 0
self.pump()
assert self.rcv.credit == 1
- assert self.rcv.queued == 1
+ assert self.rcv.queued == 1, self.rcv.queued
c = self.rcv.current
assert c
@@ -1210,7 +1210,7 @@ class SessionCreditTest(Test):
while snd.credit:
d = snd.delivery("tag%s" % idx)
assert d
- n = snd.send("x"*size)
+ n = snd.send(chr(ord("a") + idx)*size)
assert n == size, (n, size)
assert snd.advance()
self.pump()
@@ -1220,17 +1220,20 @@ class SessionCreditTest(Test):
assert snd.session.outgoing_bytes < total_bytes, (snd.session.outgoing_bytes, total_bytes)
assert rcv.session.incoming_bytes < capacity, (rcv.session.incoming_bytes, capacity)
+ assert snd.session.outgoing_bytes + rcv.session.incoming_bytes == total_bytes, \
+ (snd.session.outgoing_bytes, rcv.session.incoming_bytes, total_bytes)
if snd.session.outgoing_bytes > 0:
available = rcv.session.incoming_capacity - rcv.session.incoming_bytes
- assert available < max_frame, available
+ assert available < max_frame, (available, max_frame)
for i in range(count):
d = rcv.current
+ assert d, i
pending = d.pending
before = rcv.session.incoming_bytes
assert rcv.advance()
after = rcv.session.incoming_bytes
- assert before - after == pending
+ assert before - after == pending, (before, after, pending)
snd_before = snd.session.incoming_bytes
self.pump()
snd_after = snd.session.incoming_bytes
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org