From 4c5fdd045c6278c344ad5e2b044eabcd357fcf50 Mon Sep 17 00:00:00 2001 From: Robert Gemmell Date: Mon, 20 Apr 2015 10:21:11 +0100 Subject: [PATCH] PROTON-848, PROTON-849: dont store the TransportSession or TransportLink state in maps, use the references set on the associated Session and Link objects. Update channel+link reference handling to behave more like proton-c in order to resolve the resulting test failures. --- .../proton/engine/impl/TransportImpl.java | 18 ++++------- .../proton/engine/impl/TransportLink.java | 19 +++++++++-- .../proton/engine/impl/TransportSession.java | 32 +++++++++++++++++++ tests/python/proton_tests/engine.py | 3 +- 4 files changed, 56 insertions(+), 16 deletions(-) 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 ee190a6b0f..551a699d6b 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 @@ -96,10 +96,6 @@ private static final boolean getBooleanEnv(String name) private TransportInput _inputProcessor; private TransportOutput _outputProcessor; - private Map _transportSessionState = new HashMap(); - private Map> _transportLinkState = new HashMap>(); - - private DecoderImpl _decoder = new DecoderImpl(); private EncoderImpl _encoder = new EncoderImpl(_decoder); @@ -252,12 +248,11 @@ public void bind(Connection conn) @Override public void unbind() { - for (TransportSession ts: _transportSessionState.values()) { + for (TransportSession ts: _localSessions.values()) { ts.unbind(); } - - for (TransportLink tl: _transportLinkState.values()) { - tl.unbind(); + for (TransportSession ts: _remoteSessions.values()) { + ts.unbind(); } put(Event.Type.CONNECTION_UNBOUND, _connectionEndpoint); @@ -846,23 +841,21 @@ private void processBegin() private TransportSession getTransportState(SessionImpl session) { - TransportSession transportSession = _transportSessionState.get(session); + TransportSession transportSession = session.getTransportSession(); if(transportSession == null) { transportSession = new TransportSession(this, session); session.setTransportSession(transportSession); - _transportSessionState.put(session, transportSession); } return transportSession; } private TransportLink getTransportState(LinkImpl link) { - TransportLink transportLink = _transportLinkState.get(link); + TransportLink transportLink = link.getTransportLink(); if(transportLink == null) { transportLink = TransportLink.createTransportLink(link); - _transportLinkState.put(link, transportLink); } return transportLink; } @@ -1248,6 +1241,7 @@ public void handleEnd(End end, Binary payload, Integer channel) { _remoteSessions.remove(channel); transportSession.receivedEnd(); + transportSession.unsetRemoteChannel(); SessionImpl session = transportSession.getSession(); session.setRemoteState(EndpointState.CLOSED); ErrorCondition errorCondition = end.getError(); diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java index 2b60266f38..bdd80b519e 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java @@ -46,9 +46,22 @@ protected TransportLink(T link) static TransportLink createTransportLink(L link) { - return (TransportLink) (link instanceof ReceiverImpl - ? new TransportReceiver((ReceiverImpl)link) - : new TransportSender((SenderImpl)link)); + if (link instanceof ReceiverImpl) + { + ReceiverImpl r = (ReceiverImpl) link; + TransportReceiver tr = new TransportReceiver(r); + r.setTransportLink(tr); + + return (TransportLink) tr; + } + else + { + SenderImpl s = (SenderImpl) link; + TransportSender ts = new TransportSender(s); + s.setTransportLink(ts); + + return (TransportLink) ts; + } } void unbind() diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java index 1b23df7e27..fb537cc785 100644 --- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java +++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java @@ -130,19 +130,46 @@ public boolean isLocalChannelSet() public void unsetLocalChannel() { if (isLocalChannelSet()) { + unsetLocalHandles(); _session.decref(); } _localChannel = -1; } + private void unsetLocalHandles() + { + for(int i = 0; i < _localHandleMap.length; i++) + { + TransportLink tl = _localHandleMap[i]; + if(tl != null) + { + _localHandleMap[i] = null; + tl.clearLocalHandle(); + } + } + } + public void unsetRemoteChannel() { if (isRemoteChannelSet()) { + unsetRemoteHandles(); _session.decref(); } _remoteChannel = -1; } + private void unsetRemoteHandles() + { + for(int i = 0; i < _remoteHandleMap.length; i++) + { + TransportLink tl = _remoteHandleMap[i]; + if(tl != null) + { + _remoteHandleMap[i] = null; + tl.clearRemoteHandle(); + } + } + } public UnsignedInteger getHandleMax() { @@ -334,6 +361,11 @@ public void freeLocalChannel() unsetLocalChannel(); } + public void freeRemoteChannel() + { + unsetRemoteChannel(); + } + private void setRemoteIncomingWindow(UnsignedInteger incomingWindow) { _remoteIncomingWindow = incomingWindow; diff --git a/tests/python/proton_tests/engine.py b/tests/python/proton_tests/engine.py index 99125d9a54..ca42ac0820 100644 --- a/tests/python/proton_tests/engine.py +++ b/tests/python/proton_tests/engine.py @@ -2399,9 +2399,10 @@ def doLeak(self, local, remote): Event.TRANSPORT_CLOSED) self.connection.free() + self.expect(Event.LINK_FINAL, Event.SESSION_FINAL) self.transport.unbind() - self.expect(Event.LINK_FINAL, Event.SESSION_FINAL, Event.CONNECTION_UNBOUND, Event.CONNECTION_FINAL) + self.expect(Event.CONNECTION_UNBOUND, Event.CONNECTION_FINAL) def testLocalRemoteLeak(self): self.doLeak(True, True)