Repository: qpid-proton-j Updated Branches: refs/heads/master ba9617831 -> be6d9209e
PROTON-1967 Reduce garbage created in the transport Used cached isntances of objects that are created and immediately discared after being sent into the frame writer. In order to fix tests the FrameBody types all need a copy method so that the events coming through the transport can be recorded for later analysis. Project: http://git-wip-us.apache.org/repos/asf/qpid-proton-j/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton-j/commit/be6d9209 Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton-j/tree/be6d9209 Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton-j/diff/be6d9209 Branch: refs/heads/master Commit: be6d9209e7e6e7311c1bdfae592469be29b918e3 Parents: ba96178 Author: Timothy Bish <tabish...@gmail.com> Authored: Mon Nov 19 15:51:42 2018 -0500 Committer: Timothy Bish <tabish...@gmail.com> Committed: Mon Nov 19 15:54:43 2018 -0500 ---------------------------------------------------------------------- .../org/apache/qpid/proton/amqp/Binary.java | 25 ++-- .../qpid/proton/amqp/transport/Attach.java | 50 +++++++- .../qpid/proton/amqp/transport/Begin.java | 40 ++++-- .../qpid/proton/amqp/transport/Close.java | 19 ++- .../qpid/proton/amqp/transport/Detach.java | 24 +++- .../qpid/proton/amqp/transport/Disposition.java | 19 +++ .../qpid/proton/amqp/transport/EmptyFrame.java | 6 + .../apache/qpid/proton/amqp/transport/End.java | 19 ++- .../apache/qpid/proton/amqp/transport/Flow.java | 31 ++++- .../qpid/proton/amqp/transport/FrameBody.java | 6 + .../apache/qpid/proton/amqp/transport/Open.java | 44 ++++++- .../qpid/proton/amqp/transport/Transfer.java | 25 +++- .../qpid/proton/engine/impl/TransportImpl.java | 126 +++++++++---------- .../qpid/proton/amqp/transport/AttachTest.java | 65 ++++++++++ .../qpid/proton/amqp/transport/BeginTest.java | 58 +++++++++ .../qpid/proton/amqp/transport/CloseTest.java | 36 ++++++ .../qpid/proton/amqp/transport/DetachTest.java | 45 +++++++ .../qpid/proton/amqp/transport/EndTest.java | 36 ++++++ .../qpid/proton/amqp/transport/FlowTest.java | 58 +++++++++ .../qpid/proton/amqp/transport/OpenTest.java | 59 +++++++++ .../proton/amqp/transport/TransferTest.java | 48 +++++++ .../qpid/proton/codec/TransferTypeTest.java | 2 +- .../proton/engine/impl/TransportImplTest.java | 2 +- 23 files changed, 746 insertions(+), 97 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/Binary.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/Binary.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/Binary.java index ab1bfe5..f65a117 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/Binary.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/Binary.java @@ -27,7 +27,6 @@ import org.apache.qpid.proton.codec.ReadableBuffer; public final class Binary { - private final byte[] _data; private final int _offset; private final int _length; @@ -102,7 +101,6 @@ public final class Binary return true; } - public int getArrayOffset() { return _offset; @@ -118,11 +116,11 @@ public final class Binary return _length; } + @Override public String toString() { StringBuilder str = new StringBuilder(); - for (int i = 0; i < _length; i++) { byte c = _data[_offset + i]; @@ -138,12 +136,10 @@ public final class Binary } return str.toString(); - } public static Binary combine(final Collection<Binary> binaries) { - if(binaries.size() == 1) { return binaries.iterator().next(); @@ -190,9 +186,11 @@ public final class Binary public static Binary create(ByteBuffer buffer) { - if( buffer == null ) + if (buffer == null) + { return null; - if( buffer.isDirect() || buffer.isReadOnly() ) + } + if (buffer.isDirect() || buffer.isReadOnly()) { byte data[] = new byte [buffer.remaining()]; ByteBuffer dup = buffer.duplicate(); @@ -205,4 +203,17 @@ public final class Binary } } + public static Binary copy(Binary source) + { + if (source == null) + { + return null; + } + else + { + byte[] data = new byte[source.getLength()]; + System.arraycopy(source.getArray(), source.getArrayOffset(), data, 0, source.getLength()); + return new Binary(data); + } + } } http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Attach.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Attach.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Attach.java index 1d1746f..fa6dbad 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Attach.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Attach.java @@ -23,17 +23,18 @@ package org.apache.qpid.proton.amqp.transport; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; + import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.Symbol; import org.apache.qpid.proton.amqp.UnsignedInteger; import org.apache.qpid.proton.amqp.UnsignedLong; -import java.util.Arrays; -import java.util.Map; - +@SuppressWarnings("rawtypes") public final class Attach implements FrameBody { - private String _name; private UnsignedInteger _handle; private Role _role = Role.SENDER; @@ -49,6 +50,39 @@ public final class Attach implements FrameBody private Symbol[] _desiredCapabilities; private Map _properties; + public Attach() {} + + @SuppressWarnings("unchecked") + public Attach(Attach other) + { + this._name = other._name; + this._handle = other._handle; + this._role = other._role; + this._sndSettleMode = other._sndSettleMode; + this._rcvSettleMode = other._rcvSettleMode; + if (other._source != null) { + this._source = other._source.copy(); + } + if (other._target != null) { + this._target = other._target.copy(); + } + if (other._unsettled != null) { + this._unsettled = new LinkedHashMap<>(other._unsettled); + } + this._incompleteUnsettled = other._incompleteUnsettled; + this._initialDeliveryCount = other._initialDeliveryCount; + this._maxMessageSize = other._maxMessageSize; + if (other._offeredCapabilities != null) { + this._offeredCapabilities = Arrays.copyOf(other._offeredCapabilities, other._offeredCapabilities.length); + } + if (other._desiredCapabilities != null) { + this._desiredCapabilities = Arrays.copyOf(other._desiredCapabilities, other._desiredCapabilities.length); + } + if (other._properties != null) { + this._properties = new LinkedHashMap<>(other._properties); + } + } + public String getName() { return _name; @@ -203,12 +237,12 @@ public final class Attach implements FrameBody _properties = properties; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleAttach(this, payload, context); } - @Override public String toString() { @@ -229,4 +263,10 @@ public final class Attach implements FrameBody ", properties=" + _properties + '}'; } + + @Override + public Attach copy() + { + return new Attach(this); + } } http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Begin.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Begin.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Begin.java index ead4050..c4859fb 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Begin.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Begin.java @@ -23,18 +23,18 @@ package org.apache.qpid.proton.amqp.transport; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; + import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.Symbol; import org.apache.qpid.proton.amqp.UnsignedInteger; import org.apache.qpid.proton.amqp.UnsignedShort; -import java.util.Arrays; -import java.util.Map; - - +@SuppressWarnings("rawtypes") public final class Begin implements FrameBody { - private UnsignedShort _remoteChannel; private UnsignedInteger _nextOutgoingId; private UnsignedInteger _incomingWindow; @@ -44,6 +44,27 @@ public final class Begin implements FrameBody private Symbol[] _desiredCapabilities; private Map _properties; + public Begin() {} + + @SuppressWarnings("unchecked") + public Begin(Begin other) + { + this._remoteChannel = other._remoteChannel; + this._nextOutgoingId = other._nextOutgoingId; + this._incomingWindow = other._incomingWindow; + this._outgoingWindow = other._outgoingWindow; + this._handleMax = other._handleMax; + if (other._offeredCapabilities != null) { + this._offeredCapabilities = Arrays.copyOf(other._offeredCapabilities, other._offeredCapabilities.length); + } + if (other._desiredCapabilities != null) { + this._desiredCapabilities = Arrays.copyOf(other._desiredCapabilities, other._desiredCapabilities.length); + } + if (other._properties != null) { + this._properties = new LinkedHashMap<>(other._properties); + } + } + public UnsignedShort getRemoteChannel() { return _remoteChannel; @@ -139,12 +160,12 @@ public final class Begin implements FrameBody _properties = properties; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleBegin(this, payload, context); } - @Override public String toString() { @@ -159,5 +180,10 @@ public final class Begin implements FrameBody ", properties=" + _properties + '}'; } + + @Override + public Begin copy() + { + return new Begin(this); + } } - \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Close.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Close.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Close.java index ca141f8..05145bb 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Close.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Close.java @@ -29,6 +29,17 @@ public final class Close implements FrameBody { private ErrorCondition _error; + public Close() {} + + public Close(Close other) + { + if (other._error != null) + { + this._error = new ErrorCondition(); + this._error.copyFrom(other._error); + } + } + public ErrorCondition getError() { return _error; @@ -39,6 +50,7 @@ public final class Close implements FrameBody _error = error; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleClose(this, payload, context); @@ -51,5 +63,10 @@ public final class Close implements FrameBody "error=" + _error + '}'; } + + @Override + public Close copy() + { + return new Close(this); + } } - \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Detach.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Detach.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Detach.java index b0f2936..2b96611 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Detach.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Detach.java @@ -28,11 +28,23 @@ import org.apache.qpid.proton.amqp.UnsignedInteger; public final class Detach implements FrameBody { - private UnsignedInteger _handle; private boolean _closed; private ErrorCondition _error; + public Detach() {} + + public Detach(Detach other) + { + this._handle = other._handle; + this._closed = other._closed; + if (other._error != null) + { + this._error = new ErrorCondition(); + this._error.copyFrom(other._error); + } + } + public UnsignedInteger getHandle() { return _handle; @@ -67,7 +79,8 @@ public final class Detach implements FrameBody { _error = error; } - + + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleDetach(this, payload, context); @@ -82,5 +95,10 @@ public final class Detach implements FrameBody ", error=" + _error + '}'; } + + @Override + public Detach copy() + { + return new Detach(this); + } } - \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Disposition.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Disposition.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Disposition.java index a850940..771789b 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Disposition.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Disposition.java @@ -35,6 +35,18 @@ public final class Disposition implements FrameBody private DeliveryState _state; private boolean _batchable; + public Disposition() {} + + public Disposition(Disposition other) + { + this._role = other._role; + this._first = other._first; + this._last = other._last; + this._settled = other._settled; + this._state = other._state; + this._batchable = other._batchable; + } + public Role getRole() { return _role; @@ -104,6 +116,7 @@ public final class Disposition implements FrameBody _batchable = batchable; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleDisposition(this, payload, context); @@ -121,4 +134,10 @@ public final class Disposition implements FrameBody ", batchable=" + _batchable + '}'; } + + @Override + public Disposition copy() + { + return new Disposition(this); + } } http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/EmptyFrame.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/EmptyFrame.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/EmptyFrame.java index bb76ba9..15f0308 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/EmptyFrame.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/EmptyFrame.java @@ -37,4 +37,10 @@ public final class EmptyFrame implements FrameBody { return "Empty Frame"; } + + @Override + public EmptyFrame copy() + { + return new EmptyFrame(); + } } http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/End.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/End.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/End.java index e53a456..7cbd774 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/End.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/End.java @@ -29,6 +29,17 @@ public final class End implements FrameBody { private ErrorCondition _error; + public End() {} + + public End(End other) + { + if (other._error != null) + { + this._error = new ErrorCondition(); + this._error.copyFrom(other._error); + } + } + public ErrorCondition getError() { return _error; @@ -39,6 +50,7 @@ public final class End implements FrameBody _error = error; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleEnd(this, payload, context); @@ -51,5 +63,10 @@ public final class End implements FrameBody "error=" + _error + '}'; } + + @Override + public End copy() + { + return new End(this); + } } - \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Flow.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Flow.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Flow.java index b421a00..a321bd3 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Flow.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Flow.java @@ -23,11 +23,13 @@ package org.apache.qpid.proton.amqp.transport; +import java.util.LinkedHashMap; import java.util.Map; import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.UnsignedInteger; +@SuppressWarnings("rawtypes") public final class Flow implements FrameBody { private UnsignedInteger _nextIncomingId; @@ -42,6 +44,27 @@ public final class Flow implements FrameBody private boolean _echo; private Map _properties; + public Flow() {} + + @SuppressWarnings("unchecked") + public Flow(Flow other) + { + this._nextIncomingId = other._nextIncomingId; + this._incomingWindow = other._incomingWindow; + this._nextOutgoingId = other._nextOutgoingId; + this._outgoingWindow = other._outgoingWindow; + this._handle = other._handle; + this._deliveryCount = other._deliveryCount; + this._linkCredit = other._linkCredit; + this._available = other._available; + this._drain = other._drain; + this._echo = other._echo; + if (other._properties != null) + { + this._properties = new LinkedHashMap<>(other._properties); + } + } + public UnsignedInteger getNextIncomingId() { return _nextIncomingId; @@ -167,6 +190,7 @@ public final class Flow implements FrameBody _properties = properties; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleFlow(this, payload, context); @@ -189,5 +213,10 @@ public final class Flow implements FrameBody ", properties=" + _properties + '}'; } + + @Override + public Flow copy() + { + return new Flow(this); + } } - \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/FrameBody.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/FrameBody.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/FrameBody.java index 90a2431..4fb08c3 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/FrameBody.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/FrameBody.java @@ -40,4 +40,10 @@ public interface FrameBody } <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context); + + /** + * @return a deep copy of this FrameBody. + */ + FrameBody copy(); + } http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Open.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Open.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Open.java index e4ec98b..46283fd 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Open.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Open.java @@ -23,15 +23,16 @@ package org.apache.qpid.proton.amqp.transport; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; + import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.Symbol; import org.apache.qpid.proton.amqp.UnsignedInteger; import org.apache.qpid.proton.amqp.UnsignedShort; -import java.util.Arrays; -import java.util.Map; - - +@SuppressWarnings("rawtypes") public final class Open implements FrameBody { private String _containerId; @@ -45,6 +46,33 @@ public final class Open implements FrameBody private Symbol[] _desiredCapabilities; private Map _properties; + public Open() {} + + @SuppressWarnings("unchecked") + public Open(Open other) + { + this._containerId = other._containerId; + this._hostname = other._hostname; + this._maxFrameSize = other._maxFrameSize; + this._channelMax = other._channelMax; + this._idleTimeOut = other._idleTimeOut; + if (other._outgoingLocales != null) { + this._outgoingLocales = Arrays.copyOf(other._outgoingLocales, other._outgoingLocales.length); + } + if (other._incomingLocales != null) { + this._incomingLocales = Arrays.copyOf(other._incomingLocales, other._incomingLocales.length); + } + if (other._offeredCapabilities != null) { + this._offeredCapabilities = Arrays.copyOf(other._offeredCapabilities, other._offeredCapabilities.length); + } + if (other._desiredCapabilities != null) { + this._desiredCapabilities = Arrays.copyOf(other._desiredCapabilities, other._desiredCapabilities.length); + } + if (other._properties != null) { + this._properties = new LinkedHashMap<>(other._properties); + } + } + public String getContainerId() { return _containerId; @@ -150,6 +178,7 @@ public final class Open implements FrameBody _properties = properties; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleOpen(this, payload, context); @@ -171,5 +200,10 @@ public final class Open implements FrameBody ", properties=" + _properties + '}'; } + + @Override + public Open copy() + { + return new Open(this); + } } - \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Transfer.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Transfer.java b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Transfer.java index 35c421c..b95c476 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Transfer.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/amqp/transport/Transfer.java @@ -40,6 +40,23 @@ public final class Transfer implements FrameBody private boolean _aborted; private boolean _batchable; + public Transfer() {} + + public Transfer(Transfer other) + { + this._handle = other._handle; + this._deliveryId = other._deliveryId; + this._deliveryTag = Binary.copy(other._deliveryTag); + this._messageFormat = other._messageFormat; + this._settled = other._settled; + this._more = other._more; + this._rcvSettleMode = other._rcvSettleMode; + this._state = other._state; + this._resume = other._resume; + this._aborted = other._aborted; + this._batchable = other._batchable; + } + public UnsignedInteger getHandle() { return _handle; @@ -155,6 +172,7 @@ public final class Transfer implements FrameBody _batchable = batchable; } + @Override public <E> void invoke(FrameBodyHandler<E> handler, Binary payload, E context) { handler.handleTransfer(this, payload, context); @@ -177,5 +195,10 @@ public final class Transfer implements FrameBody ", batchable=" + _batchable + '}'; } + + @Override + public Transfer copy() + { + return new Transfer(this); + } } - \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java ---------------------------------------------------------------------- diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java index 1a02cf3..9f68260 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java @@ -145,7 +145,11 @@ public class TransportImpl extends EndpointImpl private List<TransportLayer> _additionalTransportLayers; - private final PartialTransferHandler partialTransferHandler = new PartialTransferHandler(); + // Cached instances used to carry the Performatives to the frame writer without the need to create + // a new instance on each operation that triggers a write + private final Disposition cachedDisposition = new Disposition(); + private final Flow cachedFlow = new Flow(); + private final Transfer cachedTransfer = new Transfer(); /** * Application code should use {@link org.apache.qpid.proton.engine.Transport.Factory#create()} instead @@ -155,7 +159,6 @@ public class TransportImpl extends EndpointImpl this(DEFAULT_MAX_FRAME_SIZE); } - /** * Creates a transport with the given maximum frame size. * Note that the maximumFrameSize also determines the size of the output buffer. @@ -447,7 +450,6 @@ public class TransportImpl extends EndpointImpl transportLink.clearLocalHandle(); transportSession.freeLocalHandle(localHandle); - Detach detach = new Detach(); detach.setHandle(localHandle); detach.setClosed(!link.detached()); @@ -458,7 +460,6 @@ public class TransportImpl extends EndpointImpl detach.setError(localError); } - writeFrame(transportSession.getLocalChannel(), detach, null, null); } @@ -472,19 +473,24 @@ public class TransportImpl extends EndpointImpl private void writeFlow(TransportSession ssn, TransportLink link) { - Flow flow = new Flow(); - flow.setNextIncomingId(ssn.getNextIncomingId()); - flow.setNextOutgoingId(ssn.getNextOutgoingId()); + cachedFlow.setNextIncomingId(ssn.getNextIncomingId()); + cachedFlow.setNextOutgoingId(ssn.getNextOutgoingId()); ssn.updateIncomingWindow(); - flow.setIncomingWindow(ssn.getIncomingWindowSize()); - flow.setOutgoingWindow(ssn.getOutgoingWindowSize()); + cachedFlow.setIncomingWindow(ssn.getIncomingWindowSize()); + cachedFlow.setOutgoingWindow(ssn.getOutgoingWindowSize()); + cachedFlow.setProperties(null); if (link != null) { - flow.setHandle(link.getLocalHandle()); - flow.setDeliveryCount(link.getDeliveryCount()); - flow.setLinkCredit(link.getLinkCredit()); - flow.setDrain(link.getLink().getDrain()); + cachedFlow.setHandle(link.getLocalHandle()); + cachedFlow.setDeliveryCount(link.getDeliveryCount()); + cachedFlow.setLinkCredit(link.getLinkCredit()); + cachedFlow.setDrain(link.getLink().getDrain()); + } else { + cachedFlow.setHandle(null); + cachedFlow.setDeliveryCount(null); + cachedFlow.setLinkCredit(null); + cachedFlow.setDrain(false); } - writeFrame(ssn.getLocalChannel(), flow, null, null); + writeFrame(ssn.getLocalChannel(), cachedFlow, null, null); } private void processSenderFlow() @@ -494,7 +500,6 @@ public class TransportImpl extends EndpointImpl EndpointImpl endpoint = _connectionEndpoint.getTransportHead(); while(endpoint != null) { - if(endpoint instanceof SenderImpl) { SenderImpl sender = (SenderImpl) endpoint; @@ -509,7 +514,6 @@ public class TransportImpl extends EndpointImpl writeFlow(transportSession, transportLink); } - } endpoint = endpoint.transportNext(); @@ -578,35 +582,48 @@ public class TransportImpl extends EndpointImpl tpDelivery = new TransportDelivery(deliveryId, delivery, tpLink); delivery.setTransportDelivery(tpDelivery); - final Transfer transfer = new Transfer(); - transfer.setDeliveryId(deliveryId); - transfer.setDeliveryTag(new Binary(delivery.getTag())); - transfer.setHandle(tpLink.getLocalHandle()); + cachedTransfer.setDeliveryId(deliveryId); + cachedTransfer.setDeliveryTag(new Binary(delivery.getTag())); + cachedTransfer.setHandle(tpLink.getLocalHandle()); + cachedTransfer.setRcvSettleMode(null); + cachedTransfer.setResume(false); // Ensure default is written + cachedTransfer.setAborted(false); // Ensure default is written + cachedTransfer.setBatchable(false); // Ensure default is written if(delivery.getLocalState() != null) { - transfer.setState(delivery.getLocalState()); + cachedTransfer.setState(delivery.getLocalState()); + } + else + { + cachedTransfer.setState(null); } if(delivery.isSettled()) { - transfer.setSettled(Boolean.TRUE); + cachedTransfer.setSettled(Boolean.TRUE); } else { + cachedTransfer.setSettled(Boolean.FALSE); tpSession.addUnsettledOutgoing(deliveryId, delivery); } if(snd.current() == delivery) { - transfer.setMore(true); + cachedTransfer.setMore(true); + } + else + { + // Partial transfers will reset this as needed to true in the FrameWriter + cachedTransfer.setMore(false); } int messageFormat = delivery.getMessageFormat(); if(messageFormat == DeliveryImpl.DEFAULT_MESSAGE_FORMAT) { - transfer.setMessageFormat(UnsignedInteger.ZERO); + cachedTransfer.setMessageFormat(UnsignedInteger.ZERO); } else { - transfer.setMessageFormat(UnsignedInteger.valueOf(messageFormat)); + cachedTransfer.setMessageFormat(UnsignedInteger.valueOf(messageFormat)); } ReadableBuffer payload = delivery.getData(); @@ -614,9 +631,8 @@ public class TransportImpl extends EndpointImpl int pending = payload.remaining(); try { - writeFrame(tpSession.getLocalChannel(), transfer, payload, partialTransferHandler.setTransfer(transfer)); + writeFrame(tpSession.getLocalChannel(), cachedTransfer, payload, () -> cachedTransfer.setMore(true)); } finally { - partialTransferHandler.setTransfer(null); delivery.afterSend(); // Allow for freeing resources after write of buffered data } @@ -627,7 +643,7 @@ public class TransportImpl extends EndpointImpl { session.incrementOutgoingBytes(-pending); - if (!transfer.getMore()) { + if (!cachedTransfer.getMore()) { // Clear the in-progress delivery marker tpLink.setInProgressDelivery(null); @@ -655,26 +671,25 @@ public class TransportImpl extends EndpointImpl if(wasDone && delivery.getLocalState() != null) { TransportDelivery tpDelivery = delivery.getTransportDelivery(); - Disposition disposition = new Disposition(); - disposition.setFirst(tpDelivery.getDeliveryId()); - disposition.setLast(tpDelivery.getDeliveryId()); - disposition.setRole(Role.SENDER); - disposition.setSettled(delivery.isSettled()); + // Use cached object as holder of data for immediate write to the FrameWriter + cachedDisposition.setFirst(tpDelivery.getDeliveryId()); + cachedDisposition.setLast(tpDelivery.getDeliveryId()); + cachedDisposition.setRole(Role.SENDER); + cachedDisposition.setSettled(delivery.isSettled()); + cachedDisposition.setBatchable(false); // Enforce default is written if(delivery.isSettled()) { tpDelivery.settled(); } - disposition.setState(delivery.getLocalState()); + cachedDisposition.setState(delivery.getLocalState()); - writeFrame(tpSession.getLocalChannel(), disposition, null, - null); + writeFrame(tpSession.getLocalChannel(), cachedDisposition, null, null); } return !delivery.isBuffered(); } - private boolean processTransportWorkReceiver(DeliveryImpl delivery, - ReceiverImpl rcv) + private boolean processTransportWorkReceiver(DeliveryImpl delivery, ReceiverImpl rcv) { TransportDelivery tpDelivery = delivery.getTransportDelivery(); SessionImpl session = rcv.getSession(); @@ -684,19 +699,19 @@ public class TransportImpl extends EndpointImpl { boolean settled = delivery.isSettled(); DeliveryState localState = delivery.getLocalState(); - - Disposition disposition = new Disposition(); - disposition.setFirst(tpDelivery.getDeliveryId()); - disposition.setLast(tpDelivery.getDeliveryId()); - disposition.setRole(Role.RECEIVER); - disposition.setSettled(settled); - disposition.setState(localState); + // Use cached object as holder of data for immediate write to the FrameWriter + cachedDisposition.setFirst(tpDelivery.getDeliveryId()); + cachedDisposition.setLast(tpDelivery.getDeliveryId()); + cachedDisposition.setRole(Role.RECEIVER); + cachedDisposition.setSettled(settled); + cachedDisposition.setState(localState); + cachedDisposition.setBatchable(false); // Enforce default is written if(localState == null && settled) { - disposition.setState(delivery.getDefaultDeliveryState()); + cachedDisposition.setState(delivery.getDefaultDeliveryState()); } - writeFrame(tpSession.getLocalChannel(), disposition, null, null); + writeFrame(tpSession.getLocalChannel(), cachedDisposition, null, null); if (settled) { tpDelivery.settled(); @@ -1685,23 +1700,6 @@ public class TransportImpl extends EndpointImpl return "TransportImpl [_connectionEndpoint=" + _connectionEndpoint + ", " + super.toString() + "]"; } - private static class PartialTransferHandler implements Runnable - { - private Transfer _transfer; - - PartialTransferHandler setTransfer(Transfer transfer) - { - this._transfer = transfer; - return this; - } - - @Override - public void run() - { - _transfer.setMore(true); - } - } - /** * Override the default frame handler. Must be called before the transport starts being used * (e.g. {@link #getInputBuffer()}, {@link #getOutputBuffer()}, {@link #ssl(SslDomain)} etc). http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/AttachTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/AttachTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/AttachTest.java new file mode 100644 index 0000000..4cc9137 --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/AttachTest.java @@ -0,0 +1,65 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.apache.qpid.proton.amqp.Symbol; +import org.apache.qpid.proton.amqp.UnsignedInteger; +import org.apache.qpid.proton.amqp.UnsignedLong; +import org.junit.Test; + +public class AttachTest { + + @Test + public void testCopy() { + Attach attach = new Attach(); + + attach.setName("test"); + attach.setHandle(UnsignedInteger.ONE); + attach.setRole(Role.RECEIVER); + attach.setSndSettleMode(SenderSettleMode.MIXED); + attach.setRcvSettleMode(ReceiverSettleMode.SECOND); + attach.setSource(null); + attach.setTarget(new org.apache.qpid.proton.amqp.messaging.Target()); + attach.setUnsettled(null); + attach.setIncompleteUnsettled(false); + attach.setInitialDeliveryCount(UnsignedInteger.valueOf(42)); + attach.setMaxMessageSize(UnsignedLong.valueOf(1024)); + attach.setOfferedCapabilities(new Symbol[] { Symbol.valueOf("anonymous-relay") }); + attach.setDesiredCapabilities(new Symbol[0]); + + final Attach copyOf = attach.copy(); + + assertEquals(attach.getName(), copyOf.getName()); + assertArrayEquals(attach.getDesiredCapabilities(), copyOf.getDesiredCapabilities()); + assertEquals(attach.getHandle(), copyOf.getHandle()); + assertEquals(attach.getRole(), copyOf.getRole()); + assertEquals(attach.getSndSettleMode(), copyOf.getSndSettleMode()); + assertEquals(attach.getRcvSettleMode(), copyOf.getRcvSettleMode()); + assertNull(copyOf.getSource()); + assertNotNull(copyOf.getTarget()); + assertEquals(attach.getUnsettled(), copyOf.getUnsettled()); + assertEquals(attach.getIncompleteUnsettled(), copyOf.getIncompleteUnsettled()); + assertEquals(attach.getMaxMessageSize(), copyOf.getMaxMessageSize()); + assertEquals(attach.getInitialDeliveryCount(), copyOf.getInitialDeliveryCount()); + assertArrayEquals(attach.getOfferedCapabilities(), copyOf.getOfferedCapabilities()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/BeginTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/BeginTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/BeginTest.java new file mode 100644 index 0000000..2ef3b80 --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/BeginTest.java @@ -0,0 +1,58 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.qpid.proton.amqp.Symbol; +import org.apache.qpid.proton.amqp.UnsignedInteger; +import org.apache.qpid.proton.amqp.UnsignedShort; +import org.junit.Test; + +public class BeginTest { + + @Test + public void testCopy() { + Map<Symbol, Object> properties = new HashMap<>(); + properties.put(Symbol.valueOf("x-opt"), "value"); + + Begin begin = new Begin(); + begin.setRemoteChannel(UnsignedShort.valueOf((short) 2)); + begin.setNextOutgoingId(UnsignedInteger.valueOf(10)); + begin.setIncomingWindow(UnsignedInteger.valueOf(11)); + begin.setOutgoingWindow(UnsignedInteger.valueOf(12)); + begin.setHandleMax(UnsignedInteger.valueOf(13)); + begin.setDesiredCapabilities(new Symbol[0]); + begin.setOfferedCapabilities(new Symbol[] { Symbol.valueOf("anonymous-relay") }); + begin.setProperties(properties); + + final Begin copyOf = begin.copy(); + + assertEquals(begin.getRemoteChannel(), copyOf.getRemoteChannel()); + assertEquals(begin.getNextOutgoingId(), copyOf.getNextOutgoingId()); + assertEquals(begin.getIncomingWindow(), copyOf.getIncomingWindow()); + assertEquals(begin.getOutgoingWindow(), copyOf.getOutgoingWindow()); + assertEquals(begin.getHandleMax(), copyOf.getHandleMax()); + assertArrayEquals(begin.getDesiredCapabilities(), copyOf.getDesiredCapabilities()); + assertArrayEquals(begin.getOfferedCapabilities(), copyOf.getOfferedCapabilities()); + assertEquals(begin.getProperties(), copyOf.getProperties()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/CloseTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/CloseTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/CloseTest.java new file mode 100644 index 0000000..80f6f11 --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/CloseTest.java @@ -0,0 +1,36 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.junit.Test; + +public class CloseTest { + + @Test + public void testCopy() { + Close close = new Close(); + Close copyOf = close.copy(); + + assertNull(copyOf.getError()); + close.setError(new ErrorCondition()); + copyOf = close.copy(); + assertNotNull(copyOf.getError()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/DetachTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/DetachTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/DetachTest.java new file mode 100644 index 0000000..7478304 --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/DetachTest.java @@ -0,0 +1,45 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.apache.qpid.proton.amqp.UnsignedInteger; +import org.junit.Test; + +public class DetachTest { + + @Test + public void testCopy() { + Detach detach = new Detach(); + Detach copyOf = detach.copy(); + + assertNull(copyOf.getError()); + + detach.setError(new ErrorCondition()); + detach.setClosed(true); + detach.setHandle(UnsignedInteger.ONE); + + copyOf = detach.copy(); + + assertNotNull(copyOf.getError()); + assertEquals(detach.getClosed(), copyOf.getClosed()); + assertEquals(detach.getHandle(), copyOf.getHandle()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/EndTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/EndTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/EndTest.java new file mode 100644 index 0000000..b0f23dc --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/EndTest.java @@ -0,0 +1,36 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.junit.Test; + +public class EndTest { + + @Test + public void testCopy() { + End end = new End(); + End copyOf = end.copy(); + + assertNull(copyOf.getError()); + end.setError(new ErrorCondition()); + copyOf = end.copy(); + assertNotNull(copyOf.getError()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/FlowTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/FlowTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/FlowTest.java new file mode 100644 index 0000000..fb7fdb3 --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/FlowTest.java @@ -0,0 +1,58 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertEquals; + +import java.util.HashMap; + +import org.apache.qpid.proton.amqp.UnsignedInteger; +import org.junit.Test; + +public class FlowTest { + + @Test + public void test() { + Flow flow = new Flow(); + + flow.setNextIncomingId(UnsignedInteger.valueOf(1)); + flow.setIncomingWindow(UnsignedInteger.valueOf(2)); + flow.setNextOutgoingId(UnsignedInteger.valueOf(3)); + flow.setOutgoingWindow(UnsignedInteger.valueOf(4)); + flow.setHandle(UnsignedInteger.valueOf(5)); + flow.setDeliveryCount(UnsignedInteger.valueOf(6)); + flow.setLinkCredit(UnsignedInteger.valueOf(7)); + flow.setAvailable(UnsignedInteger.valueOf(8)); + flow.setDrain(true); + flow.setEcho(true); + flow.setProperties(new HashMap<>()); + + Flow copyOf = flow.copy(); + + assertEquals(flow.getNextIncomingId(), copyOf.getNextIncomingId()); + assertEquals(flow.getIncomingWindow(), copyOf.getIncomingWindow()); + assertEquals(flow.getNextOutgoingId(), copyOf.getNextOutgoingId()); + assertEquals(flow.getOutgoingWindow(), copyOf.getOutgoingWindow()); + assertEquals(flow.getHandle(), copyOf.getHandle()); + assertEquals(flow.getDeliveryCount(), copyOf.getDeliveryCount()); + assertEquals(flow.getLinkCredit(), copyOf.getLinkCredit()); + assertEquals(flow.getAvailable(), copyOf.getAvailable()); + assertEquals(flow.getDrain(), copyOf.getDrain()); + assertEquals(flow.getEcho(), copyOf.getEcho()); + assertEquals(flow.getProperties(), copyOf.getProperties()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/OpenTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/OpenTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/OpenTest.java new file mode 100644 index 0000000..91ea2b7 --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/OpenTest.java @@ -0,0 +1,59 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.qpid.proton.amqp.Symbol; +import org.apache.qpid.proton.amqp.UnsignedInteger; +import org.apache.qpid.proton.amqp.UnsignedShort; +import org.junit.Test; + +public class OpenTest { + + @Test + public void testCopy() { + Map<Symbol, Object> properties = new HashMap<>(); + properties.put(Symbol.valueOf("x-opt"), "value"); + + Open open = new Open(); + + open.setContainerId("test"); + open.setHostname("host"); + open.setMaxFrameSize(UnsignedInteger.valueOf(42)); + open.setChannelMax(UnsignedShort.MAX_VALUE); + open.setIdleTimeOut(UnsignedInteger.valueOf(111)); + open.setOfferedCapabilities(new Symbol[] { Symbol.valueOf("anonymous-relay") }); + open.setDesiredCapabilities(new Symbol[0]); + open.setProperties(properties); + + Open copyOf = open.copy(); + + assertEquals(open.getContainerId(), copyOf.getContainerId()); + assertEquals(open.getHostname(), copyOf.getHostname()); + assertEquals(open.getMaxFrameSize(), copyOf.getMaxFrameSize()); + assertEquals(open.getChannelMax(), copyOf.getChannelMax()); + assertEquals(open.getIdleTimeOut(), copyOf.getIdleTimeOut()); + assertArrayEquals(open.getDesiredCapabilities(), copyOf.getDesiredCapabilities()); + assertArrayEquals(open.getOfferedCapabilities(), copyOf.getOfferedCapabilities()); + assertEquals(open.getProperties(), copyOf.getProperties()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/TransferTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/TransferTest.java b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/TransferTest.java new file mode 100644 index 0000000..62ebc97 --- /dev/null +++ b/proton-j/src/test/java/org/apache/qpid/proton/amqp/transport/TransferTest.java @@ -0,0 +1,48 @@ +/* + * 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.amqp.transport; + +import static org.junit.Assert.assertEquals; + +import org.apache.qpid.proton.amqp.Binary; +import org.apache.qpid.proton.amqp.UnsignedInteger; +import org.junit.Test; + +public class TransferTest { + + @Test + public void testCopyTransfers() { + Transfer transfer = new Transfer(); + transfer.setHandle(UnsignedInteger.ONE); + transfer.setDeliveryTag(new Binary(new byte[] {0, 1})); + transfer.setMessageFormat(UnsignedInteger.ZERO); + transfer.setDeliveryId(UnsignedInteger.valueOf(127)); + transfer.setAborted(false); + transfer.setBatchable(true); + transfer.setRcvSettleMode(ReceiverSettleMode.SECOND); + + final Transfer copyOf = transfer.copy(); + + assertEquals(transfer.getHandle(), copyOf.getHandle()); + assertEquals(transfer.getMessageFormat(), copyOf.getMessageFormat()); + assertEquals(transfer.getDeliveryTag(), copyOf.getDeliveryTag()); + assertEquals(transfer.getDeliveryId(), copyOf.getDeliveryId()); + assertEquals(transfer.getAborted(), copyOf.getAborted()); + assertEquals(transfer.getBatchable(), copyOf.getBatchable()); + assertEquals(transfer.getRcvSettleMode(), copyOf.getRcvSettleMode()); + } +} http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java b/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java index 8e2d088..5bfba50 100644 --- a/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java +++ b/proton-j/src/test/java/org/apache/qpid/proton/codec/TransferTypeTest.java @@ -16,7 +16,7 @@ */ package org.apache.qpid.proton.codec; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.UnsignedInteger; http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/be6d9209/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java ---------------------------------------------------------------------- diff --git a/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java b/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java index 203274f..0be219c 100644 --- a/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java +++ b/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java @@ -379,7 +379,7 @@ public class TransportImplTest protected void writeFrame(int channel, FrameBody frameBody, ReadableBuffer payload, Runnable onPayloadTooLarge) { super.writeFrame(channel, frameBody, payload, onPayloadTooLarge); - writes.addLast(frameBody); + writes.addLast(frameBody != null ? frameBody.copy() : null); } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org
