You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2018/01/11 15:37:18 UTC
qpid-proton-j git commit: PROTON-1739: add a test to ensure the
behaviour
Repository: qpid-proton-j
Updated Branches:
refs/heads/master 6fa847b1e -> d9eea7465
PROTON-1739: add a test to ensure the behaviour
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/d9eea746
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton-j/tree/d9eea746
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton-j/diff/d9eea746
Branch: refs/heads/master
Commit: d9eea74658329d65fb8aec1a21f2daf979b1f83e
Parents: 6fa847b
Author: Robbie Gemmell <ro...@apache.org>
Authored: Thu Jan 11 15:27:53 2018 +0000
Committer: Robbie Gemmell <ro...@apache.org>
Committed: Thu Jan 11 15:27:53 2018 +0000
----------------------------------------------------------------------
.../proton/engine/impl/TransportImplTest.java | 141 +++++++++++++++++++
1 file changed, 141 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton-j/blob/d9eea746/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 ff5a5b4..b474802 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
@@ -290,6 +290,14 @@ public class TransportImplTest
private class MockTransportImpl extends TransportImpl
{
+ public MockTransportImpl() {
+ super();
+ }
+
+ public MockTransportImpl(int maxFrameSize) {
+ super(maxFrameSize);
+ }
+
LinkedList<FrameBody> writes = new LinkedList<FrameBody>();
@Override
protected void writeFrame(int channel, FrameBody frameBody,
@@ -960,6 +968,139 @@ public class TransportImplTest
assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 3, transport.writes.size());
}
+ @Test
+ public void testEmittedSessionIncomingWindow()
+ {
+ doSessionIncomingWindowTestImpl(false, false);
+ doSessionIncomingWindowTestImpl(true, false);
+ doSessionIncomingWindowTestImpl(false, true);
+ doSessionIncomingWindowTestImpl(true, true);
+ }
+
+ private void doSessionIncomingWindowTestImpl(boolean setFrameSize, boolean setSessionCapacity) {
+ MockTransportImpl transport;
+ if(setFrameSize) {
+ transport = new MockTransportImpl(5*1024);
+ } else {
+ transport = new MockTransportImpl();
+ }
+
+ Connection connection = Proton.connection();
+ transport.bind(connection);
+
+ connection.open();
+
+ Session session = connection.session();
+ int sessionCapacity = 0;
+ if(setSessionCapacity) {
+ sessionCapacity = 100*1024;
+ session.setIncomingCapacity(sessionCapacity);
+ }
+
+ pumpMockTransport(transport);
+
+ assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 1, transport.writes.size());
+ assertTrue("Unexpected frame type", transport.writes.get(0) instanceof Open);
+ // Provide an Open response
+ transport.handleFrame(new TransportFrame(0, new Open(), null));
+
+ // Open session and verify emitted incoming window
+ session.open();
+ pumpMockTransport(transport);
+
+ assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 2, transport.writes.size());
+ assertTrue("Unexpected frame type", transport.writes.get(1) instanceof Begin);
+ Begin sentBegin = (Begin) transport.writes.get(1);
+
+ assertEquals("Unexpected session capacity", sessionCapacity, session.getIncomingCapacity());
+
+ int expectedWindowSize = 2147483647;
+ if(setSessionCapacity && setFrameSize) {
+ expectedWindowSize = (100*1024) / (5*1024); // capacity / frameSize
+ }
+
+ assertEquals("Unexpected session window", UnsignedInteger.valueOf(expectedWindowSize), sentBegin.getIncomingWindow());
+
+ // Open receiver
+ String linkName = "myReceiver";
+ Receiver receiver = session.receiver(linkName);
+ receiver.open();
+
+ pumpMockTransport(transport);
+
+ assertTrue("Unexpected frame type", transport.writes.get(2) instanceof Attach);
+ assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 3, transport.writes.size());
+
+ // Provide an begin+attach response
+ Begin beginResponse = new Begin();
+ beginResponse.setRemoteChannel(UnsignedShort.valueOf((short) 0));
+ beginResponse.setNextOutgoingId(UnsignedInteger.ONE);
+ beginResponse.setIncomingWindow(UnsignedInteger.valueOf(1024));
+ beginResponse.setOutgoingWindow(UnsignedInteger.valueOf(1024));
+ transport.handleFrame(new TransportFrame(0, beginResponse, null));
+
+ Attach attach = new Attach();
+ attach.setHandle(UnsignedInteger.ZERO);
+ attach.setRole(Role.SENDER);
+ attach.setName(linkName);
+ attach.setInitialDeliveryCount(UnsignedInteger.ZERO);
+ transport.handleFrame(new TransportFrame(0, attach, null));
+
+ pumpMockTransport(transport);
+
+ assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 3, transport.writes.size());
+
+ // Flow some credit, verify emitted incoming window remains the same
+ receiver.flow(1);
+
+ pumpMockTransport(transport);
+
+ assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 4, transport.writes.size());
+ assertTrue("Unexpected frame type", transport.writes.get(3) instanceof Flow);
+ Flow sentFlow = (Flow) transport.writes.get(3);
+
+ assertEquals("Unexpected session window", UnsignedInteger.valueOf(expectedWindowSize), sentFlow.getIncomingWindow());
+
+ // Provide a transfer, don't consume it, flow more credit, verify the emitted
+ // incoming window (should reduce 1 if capacity and frame size set)
+ String deliveryTag = "tag1";
+ String messageContent = "content1";
+ handleTransfer(transport, 1, deliveryTag, messageContent);
+
+ assertTrue("Unexpected session byte count", session.getIncomingBytes() > 0);
+
+ receiver.flow(1);
+
+ pumpMockTransport(transport);
+
+ assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 5, transport.writes.size());
+ assertTrue("Unexpected frame type", transport.writes.get(4) instanceof Flow);
+ sentFlow = (Flow) transport.writes.get(4);
+
+ if(setSessionCapacity && setFrameSize) {
+ expectedWindowSize = expectedWindowSize -1;
+ }
+ assertEquals("Unexpected session window", UnsignedInteger.valueOf(expectedWindowSize), sentFlow.getIncomingWindow());
+
+ // Consume the transfer then flow more credit, verify the emitted
+ // incoming window (should increase 1 if capacity and frame size set)
+ verifyDelivery(receiver, deliveryTag, messageContent);
+ assertEquals("Unexpected session byte count", 0, session.getIncomingBytes());
+
+ receiver.flow(1);
+
+ pumpMockTransport(transport);
+
+ assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 6, transport.writes.size());
+ assertTrue("Unexpected frame type", transport.writes.get(5) instanceof Flow);
+ sentFlow = (Flow) transport.writes.get(5);
+
+ if(setSessionCapacity && setFrameSize) {
+ expectedWindowSize = expectedWindowSize +1;
+ }
+ assertEquals("Unexpected session window", UnsignedInteger.valueOf(expectedWindowSize), sentFlow.getIncomingWindow());
+ }
+
/**
* Verify that no Attach frame is emitted by the Transport should a Receiver
* be opened after the Close frame was sent.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org