diff options
Diffstat (limited to 'java/client/src')
22 files changed, 102 insertions, 125 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index be240cc39e..ca8ba5e619 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -28,7 +28,7 @@ import org.apache.qpid.common.AMQPFilterTypes; import org.apache.qpid.client.failover.FailoverSupport; import org.apache.qpid.client.message.*; import org.apache.qpid.client.protocol.AMQProtocolHandler; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.util.FlowControllingBlockingQueue; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.*; diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java index d855e97204..dba44a3115 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java @@ -1,8 +1,29 @@ +/* + * + * 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.client.handler; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.qpid.client.state.AMQStateManager; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.BasicMessageConsumer; import org.apache.qpid.AMQException; import org.apache.qpid.framing.ExchangeBoundOkBody; @@ -26,10 +47,10 @@ public class BasicCancelOkMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { _logger.debug("New BasicCancelOk method received"); BasicCancelOkBody body = (BasicCancelOkBody) evt.getMethod(); - evt.getProtocolSession().confirmConsumerCancelled(evt.getChannelId(), body.consumerTag); + protocolSession.confirmConsumerCancelled(evt.getChannelId(), body.consumerTag); } } diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java index 6e12899204..f036f88d50 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java @@ -23,9 +23,10 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.framing.BasicDeliverBody; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.message.UnprocessedMessage; public class BasicDeliverMethodHandler implements StateAwareMethodListener @@ -39,12 +40,12 @@ public class BasicDeliverMethodHandler implements StateAwareMethodListener return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { final UnprocessedMessage msg = new UnprocessedMessage(); msg.deliverBody = (BasicDeliverBody) evt.getMethod(); msg.channelId = evt.getChannelId(); _logger.debug("New JmsDeliver method received"); - evt.getProtocolSession().unprocessedMessageReceived(msg); + protocolSession.unprocessedMessageReceived(msg); } } diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/BasicReturnMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/BasicReturnMethodHandler.java index 8785a96396..72923c5cae 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/BasicReturnMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/BasicReturnMethodHandler.java @@ -23,7 +23,8 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.client.message.UnprocessedMessage; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.qpid.framing.BasicReturnBody; @@ -39,7 +40,7 @@ public class BasicReturnMethodHandler implements StateAwareMethodListener return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { _logger.debug("New JmsBounce method received"); final UnprocessedMessage msg = new UnprocessedMessage(); @@ -47,6 +48,6 @@ public class BasicReturnMethodHandler implements StateAwareMethodListener msg.bounceBody = (BasicReturnBody) evt.getMethod(); msg.channelId = evt.getChannelId(); - evt.getProtocolSession().unprocessedMessageReceived(msg); + protocolSession.unprocessedMessageReceived(msg); } -}
\ No newline at end of file +} diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java index dbc1512b2f..17869154c2 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java @@ -27,7 +27,8 @@ import org.apache.qpid.AMQInvalidSelectorException; import org.apache.qpid.client.AMQNoConsumersException; import org.apache.qpid.client.AMQNoRouteException; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.qpid.framing.AMQFrame; @@ -46,7 +47,7 @@ public class ChannelCloseMethodHandler implements StateAwareMethodListener return _handler; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { _logger.debug("ChannelClose method received"); ChannelCloseBody method = (ChannelCloseBody) evt.getMethod(); @@ -62,7 +63,7 @@ public class ChannelCloseMethodHandler implements StateAwareMethodListener // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. // Be aware of possible changes to parameter order as versions change. AMQFrame frame = ChannelCloseOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0); - evt.getProtocolSession().writeFrame(frame); + protocolSession.writeFrame(frame); if (errorCode != AMQConstant.REPLY_SUCCESS.getCode()) { _logger.error("Channel close received with errorCode " + errorCode + ", and reason " + reason); @@ -86,6 +87,6 @@ public class ChannelCloseMethodHandler implements StateAwareMethodListener } } - evt.getProtocolSession().channelClosed(evt.getChannelId(), errorCode, String.valueOf(reason)); + protocolSession.channelClosed(evt.getChannelId(), errorCode, String.valueOf(reason)); } } diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseOkMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseOkMethodHandler.java index 383cebbcab..a99c963eb4 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseOkMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseOkMethodHandler.java @@ -21,7 +21,8 @@ package org.apache.qpid.client.handler; import org.apache.qpid.AMQException; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.log4j.Logger; @@ -37,7 +38,7 @@ public class ChannelCloseOkMethodHandler implements StateAwareMethodListener return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { _logger.info("Received channel-close-ok for channel-id " + evt.getChannelId()); diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelFlowOkMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelFlowOkMethodHandler.java index 9a4f7af849..7a13972d8f 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelFlowOkMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelFlowOkMethodHandler.java @@ -22,7 +22,8 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.qpid.framing.ChannelFlowOkBody; @@ -41,7 +42,7 @@ public class ChannelFlowOkMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { ChannelFlowOkBody method = (ChannelFlowOkBody) evt.getMethod(); _logger.debug("Received Channel.Flow-Ok message, active = " + method.active); diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java index bd1be5d629..b9f97f7a46 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java @@ -24,7 +24,8 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.AMQConnectionClosedException; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQState; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; @@ -48,7 +49,7 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { _logger.info("ConnectionClose frame received"); ConnectionCloseBody method = (ConnectionCloseBody) evt.getMethod(); @@ -63,7 +64,7 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. // Be aware of possible changes to parameter order as versions change. - evt.getProtocolSession().writeFrame(ConnectionCloseOkBody.createAMQFrame((short)0, (byte)8, (byte)0)); + protocolSession.writeFrame(ConnectionCloseOkBody.createAMQFrame((short)0, (byte)8, (byte)0)); if (errorCode != 200) { @@ -71,7 +72,7 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener { _logger.info("Authentication Error:"+Thread.currentThread().getName()); - evt.getProtocolSession().closeProtocolSession(); + protocolSession.closeProtocolSession(); //todo this is a bit of a fudge (could be conssidered such as each new connection needs a new state manager or at least a fresh state. stateManager.changeState(AMQState.CONNECTION_NOT_STARTED); @@ -89,7 +90,7 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener // this actually closes the connection in the case where it is not an error. - evt.getProtocolSession().closeProtocolSession(); + protocolSession.closeProtocolSession(); stateManager.changeState(AMQState.CONNECTION_CLOSED); } diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java index b5001a6e64..da903e7c1d 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java @@ -21,7 +21,8 @@ package org.apache.qpid.client.handler; import org.apache.qpid.AMQException; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQState; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; @@ -39,7 +40,7 @@ public class ConnectionOpenOkMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { stateManager.changeState(AMQState.CONNECTION_OPEN); } diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java index 5580d02895..ae622bb30b 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java @@ -22,7 +22,8 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.qpid.framing.ConnectionRedirectBody; @@ -44,7 +45,7 @@ public class ConnectionRedirectMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { _logger.info("ConnectionRedirect frame received"); ConnectionRedirectBody method = (ConnectionRedirectBody) evt.getMethod(); @@ -64,6 +65,6 @@ public class ConnectionRedirectMethodHandler implements StateAwareMethodListener host = host.substring(0, portIndex); } - evt.getProtocolSession().failover(host, port); + protocolSession.failover(host, port); } } diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionSecureMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionSecureMethodHandler.java index 153b641a39..165b83e05d 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionSecureMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionSecureMethodHandler.java @@ -24,9 +24,10 @@ import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQFrame; import org.apache.qpid.framing.ConnectionSecureOkBody; import org.apache.qpid.framing.ConnectionSecureBody; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; import javax.security.sasl.SaslClient; import javax.security.sasl.SaslException; @@ -40,9 +41,9 @@ public class ConnectionSecureMethodHandler implements StateAwareMethodListener return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { - SaslClient client = evt.getProtocolSession().getSaslClient(); + SaslClient client = protocolSession.getSaslClient(); if (client == null) { throw new AMQException("No SASL client set up - cannot proceed with authentication"); @@ -60,7 +61,7 @@ public class ConnectionSecureMethodHandler implements StateAwareMethodListener AMQFrame responseFrame = ConnectionSecureOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0, // AMQP version (major, minor) response); // response - evt.getProtocolSession().writeFrame(responseFrame); + protocolSession.writeFrame(responseFrame); } catch (SaslException e) { diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java index 6f206735fe..d2949779eb 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java @@ -24,7 +24,7 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.common.ClientProperties; import org.apache.qpid.common.QpidProperties; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.security.AMQCallbackHandler; import org.apache.qpid.client.security.CallbackHandlerRegistry; @@ -56,7 +56,7 @@ public class ConnectionStartMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { ConnectionStartBody body = (ConnectionStartBody) evt.getMethod(); @@ -78,25 +78,24 @@ public class ConnectionStartMethodHandler implements StateAwareMethodListener throw new AMQException("No supported security mechanism found, passed: " + new String(body.mechanisms)); } - final AMQProtocolSession ps = evt.getProtocolSession(); byte[] saslResponse; try { SaslClient sc = Sasl.createSaslClient(new String[]{mechanism}, null, "AMQP", "localhost", - null, createCallbackHandler(mechanism, ps)); + null,createCallbackHandler(mechanism, protocolSession)); if (sc == null) { throw new AMQException("Client SASL configuration error: no SaslClient could be created for mechanism " + mechanism + ". Please ensure all factories are registered. See DynamicSaslRegistrar for " + " details of how to register non-standard SASL client providers."); } - ps.setSaslClient(sc); + protocolSession.setSaslClient(sc); saslResponse = (sc.hasInitialResponse() ? sc.evaluateChallenge(new byte[0]) : null); } catch (SaslException e) { - ps.setSaslClient(null); + protocolSession.setSaslClient(null); throw new AMQException("Unable to create SASL client: " + e, e); } @@ -119,14 +118,14 @@ public class ConnectionStartMethodHandler implements StateAwareMethodListener stateManager.changeState(AMQState.CONNECTION_NOT_TUNED); FieldTable clientProperties = FieldTableFactory.newFieldTable(); - clientProperties.setString(ClientProperties.instance.toString(), ps.getClientID()); + clientProperties.setString(ClientProperties.instance.toString(), protocolSession.getClientID()); clientProperties.setString(ClientProperties.product.toString(), QpidProperties.getProductName()); clientProperties.setString(ClientProperties.version.toString(), QpidProperties.getReleaseVersion()); clientProperties.setString(ClientProperties.platform.toString(), getFullSystemInfo()); // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. // Be aware of possible changes to parameter order as versions change. - ps.writeFrame(ConnectionStartOkBody.createAMQFrame(evt.getChannelId(), + protocolSession.writeFrame(ConnectionStartOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0, // AMQP version (major, minor) clientProperties, // clientProperties new AMQShortString(selectedLocale), // locale diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java index 604202a742..3f4e0022ca 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java @@ -23,7 +23,7 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.client.ConnectionTuneParameters; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQState; import org.apache.qpid.client.state.AMQStateManager; @@ -45,13 +45,12 @@ public class ConnectionTuneMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { _logger.debug("ConnectionTune frame received"); ConnectionTuneBody frame = (ConnectionTuneBody) evt.getMethod(); - AMQProtocolSession session = evt.getProtocolSession(); - ConnectionTuneParameters params = session.getConnectionTuneParameters(); + ConnectionTuneParameters params = protocolSession.getConnectionTuneParameters(); if (params == null) { params = new ConnectionTuneParameters(); @@ -60,11 +59,11 @@ public class ConnectionTuneMethodHandler implements StateAwareMethodListener params.setFrameMax(frame.frameMax); params.setChannelMax(frame.channelMax); params.setHeartbeat(Integer.getInteger("amqj.heartbeat.delay", frame.heartbeat)); - session.setConnectionTuneParameters(params); + protocolSession.setConnectionTuneParameters(params); stateManager.changeState(AMQState.CONNECTION_NOT_OPENED); - session.writeFrame(createTuneOkFrame(evt.getChannelId(), params)); - session.writeFrame(createConnectionOpenFrame(evt.getChannelId(), new AMQShortString(session.getAMQConnection().getVirtualHost()), null, true)); + protocolSession.writeFrame(createTuneOkFrame(evt.getChannelId(), params)); + protocolSession.writeFrame(createConnectionOpenFrame(evt.getChannelId(), new AMQShortString(protocolSession.getAMQConnection().getVirtualHost()), null, true)); } protected AMQFrame createConnectionOpenFrame(int channel, AMQShortString path, AMQShortString capabilities, boolean insist) diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ExchangeBoundOkMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ExchangeBoundOkMethodHandler.java index 858726745e..84ad6ca0b1 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/ExchangeBoundOkMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/ExchangeBoundOkMethodHandler.java @@ -19,7 +19,8 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.qpid.framing.ExchangeBoundOkBody; @@ -41,7 +42,7 @@ public class ExchangeBoundOkMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { if (_logger.isDebugEnabled()) { diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/QueueDeleteOkMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/QueueDeleteOkMethodHandler.java index 3271a715a2..a9db28acaa 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/QueueDeleteOkMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/QueueDeleteOkMethodHandler.java @@ -17,9 +17,10 @@ */ package org.apache.qpid.client.handler; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.StateAwareMethodListener; import org.apache.qpid.client.state.AMQStateManager; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.AMQException; import org.apache.qpid.framing.QueueDeleteOkBody; import org.apache.log4j.Logger; @@ -41,7 +42,7 @@ public class QueueDeleteOkMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { if (_logger.isDebugEnabled()) { diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQMethodEvent.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQMethodEvent.java deleted file mode 100644 index 403aa3486e..0000000000 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQMethodEvent.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * - * 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.client.protocol; - -import org.apache.qpid.framing.AMQMethodBody; - -public class AMQMethodEvent -{ - private AMQMethodBody _method; - - private int _channelId; - - private AMQProtocolSession _protocolSession; - - public AMQMethodEvent(int channelId, AMQMethodBody method, AMQProtocolSession protocolSession) - { - _channelId = channelId; - _method = method; - _protocolSession = protocolSession; - } - - public AMQMethodBody getMethod() - { - return _method; - } - - public int getChannelId() - { - return _channelId; - } - - public AMQProtocolSession getProtocolSession() - { - return _protocolSession; - } - - public String toString() - { - StringBuffer buf = new StringBuffer("Method event: "); - buf.append("\nChannel id: ").append(_channelId); - buf.append("\nMethod: ").append(_method); - return buf.toString(); - } -} diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQMethodListener.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQMethodListener.java index 68299033a5..2cbd8f0e32 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQMethodListener.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQMethodListener.java @@ -21,6 +21,8 @@ package org.apache.qpid.client.protocol; import org.apache.qpid.AMQException; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; public interface AMQMethodListener { @@ -34,7 +36,7 @@ public interface AMQMethodListener * to all registered listeners using the error() method (see below) allowing them to * perform cleanup if necessary. */ - boolean methodReceived(AMQMethodEvent evt) throws AMQException; + boolean methodReceived(AMQMethodEvent evt, AMQProtocolSession protocolSession) throws AMQException; /** * Callback when an error has occurred. Allows listeners to clean up. diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index bd60b2c250..220f3c3b69 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -29,6 +29,7 @@ import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.qpid.AMQConnectionClosedException; import org.apache.qpid.AMQDisconnectedException; import org.apache.qpid.AMQException; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.client.AMQSession; import org.apache.qpid.client.failover.FailoverHandler; @@ -310,19 +311,19 @@ public class AMQProtocolHandler extends IoHandlerAdapter _logger.debug("Method frame received: " + frame); } - final AMQMethodEvent evt = new AMQMethodEvent(frame.channel, (AMQMethodBody) frame.bodyFrame, _protocolSession); + final AMQMethodEvent<AMQMethodBody> evt = new AMQMethodEvent<AMQMethodBody>(frame.channel, (AMQMethodBody) frame.bodyFrame); try { - boolean wasAnyoneInterested = _stateManager.methodReceived(evt); + boolean wasAnyoneInterested = _stateManager.methodReceived(evt, _protocolSession); if(!_frameListeners.isEmpty()) { Iterator it = _frameListeners.iterator(); while (it.hasNext()) { final AMQMethodListener listener = (AMQMethodListener) it.next(); - wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested; + wasAnyoneInterested = listener.methodReceived(evt, _protocolSession) || wasAnyoneInterested; } } if (!wasAnyoneInterested) diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java index ca622a98ba..718d256a52 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java @@ -32,6 +32,7 @@ import org.apache.qpid.client.ConnectionTuneParameters; import org.apache.qpid.client.message.UnexpectedBodyReceivedException; import org.apache.qpid.client.message.UnprocessedMessage; import org.apache.qpid.framing.*; +import org.apache.qpid.protocol.AMQProtocolWriter; import org.apache.commons.lang.StringUtils; import javax.jms.JMSException; @@ -45,7 +46,7 @@ import java.util.concurrent.ConcurrentMap; * The underlying protocol session is still available but clients should not * use it to obtain session attributes. */ -public class AMQProtocolSession implements ProtocolVersionList +public class AMQProtocolSession implements AMQProtocolWriter, ProtocolVersionList { protected static final int LAST_WRITE_FUTURE_JOIN_TIMEOUT = 1000 * 60 * 2; diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java b/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java index 21ae3fc71f..4fff4fab00 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java @@ -22,6 +22,8 @@ package org.apache.qpid.client.protocol; import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQMethodBody; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; public abstract class BlockingMethodFrameListener implements AMQMethodListener { @@ -53,7 +55,7 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener * @return true if the listener has dealt with this frame * @throws AMQException */ - public boolean methodReceived(AMQMethodEvent evt) throws AMQException + public boolean methodReceived(AMQMethodEvent evt, AMQProtocolSession protocolSession) throws AMQException { AMQMethodBody method = evt.getMethod(); diff --git a/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java b/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java index 50bd1667f9..c9b7593796 100644 --- a/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java +++ b/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java @@ -22,7 +22,8 @@ package org.apache.qpid.client.state; import org.apache.qpid.AMQException; import org.apache.qpid.client.handler.*; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.protocol.AMQMethodListener; import org.apache.qpid.framing.*; import org.apache.log4j.Logger; @@ -146,12 +147,12 @@ public class AMQStateManager implements AMQMethodListener } } - public boolean methodReceived(AMQMethodEvent evt) throws AMQException + public boolean methodReceived(AMQMethodEvent evt, AMQProtocolSession protocolSession) throws AMQException { StateAwareMethodListener handler = findStateTransitionHandler(_currentState, evt.getMethod()); if (handler != null) { - handler.methodReceived(this, evt); + handler.methodReceived(this, protocolSession, evt); return true; } return false; diff --git a/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java b/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java index 37a0f9f376..f2c8916868 100644 --- a/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java +++ b/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java @@ -20,7 +20,8 @@ */ package org.apache.qpid.client.state; -import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.AMQException; /** @@ -30,5 +31,6 @@ import org.apache.qpid.AMQException; */ public interface StateAwareMethodListener { - void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException; + void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession, + AMQMethodEvent evt) throws AMQException; } |