You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2007/03/18 00:54:03 UTC
svn commit: r519474 [1/2] - in
/activemq/activemq-cpp/trunk/activemq-cpp/src: main/
main/activemq/connector/ main/activemq/connector/openwire/
main/activemq/connector/openwire/commands/ main/activemq/connector/stomp/
main/activemq/core/ main/cms/ test-...
Author: tabish
Date: Sat Mar 17 16:54:01 2007
New Revision: 519474
URL: http://svn.apache.org/viewvc?view=rev&rev=519474
Log:
http://issues.apache.org/activemq/browse/AMQCPP-78
Refectored the close of ConnectorResources, added the ConnectorResourceListener that allows objects to listen to connector resources for close events.
Removed the ActiveMQSessionResource interface and made everything dependant on the ConnectorResource close events.
Fixed a few memory leaks.
Fixed tests for the new interfaces.
Added:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.h (with props)
Removed:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSessionResource.h
Modified:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/Connector.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConnectorResource.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConsumerInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ProducerInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/SessionInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/TransactionInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConsumerInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireProducerInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireSessionInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireTransactionInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConsumerInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompProducerInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompTransactionInfo.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQProducer.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQProducer.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSession.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSession.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQTransaction.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQTransaction.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/cms/TemporaryQueue.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/cms/TemporaryTopic.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireTempDestinationTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/connector/stomp/StompConnectorTest.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/connector/stomp/StompSessionManagerTest.h
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am Sat Mar 17 16:54:01 2007
@@ -65,6 +65,7 @@
activemq/connector/openwire/utils/OpenwireStringSupport.cpp \
activemq/connector/openwire/marshal/PrimitiveMapMarshaller.cpp \
activemq/connector/ConnectorFactoryMap.cpp \
+ activemq/connector/BaseConnectorResource.cpp \
activemq/network/SocketInputStream.cpp \
activemq/network/ServerSocket.cpp \
activemq/network/TcpSocket.cpp \
@@ -89,7 +90,6 @@
activemq/util/URISupport.cpp
h_sources = \
- activemq/core/ActiveMQSessionResource.h \
activemq/core/ActiveMQProducer.h \
activemq/core/ActiveMQMessage.h \
activemq/core/ActiveMQConnectionData.h \
@@ -166,7 +166,9 @@
activemq/connector/TransactionInfo.h \
activemq/connector/SessionInfo.h \
activemq/connector/ConnectorResource.h \
+ activemq/connector/ConnectorResourceListener.h \
activemq/connector/ConnectorFactory.h \
+ activemq/connector/BaseConnectorResource.h \
activemq/connector/stomp/StompSessionInfo.h \
activemq/connector/stomp/StompTransactionInfo.h \
activemq/connector/stomp/StompSessionManager.h \
Added: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.cpp?view=auto&rev=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.cpp (added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.cpp Sat Mar 17 16:54:01 2007
@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+#include "BaseConnectorResource.h"
+
+#include <activemq/exceptions/NullPointerException.h>
+#include <activemq/connector/Connector.h>
+
+using namespace activemq;
+using namespace activemq::connector;
+using namespace activemq::exceptions;
+
+///////////////////////////////////////////////////////////////////////////////
+BaseConnectorResource::BaseConnectorResource()
+{
+ this->connector = NULL;
+ this->closed = false;
+}
+
+///////////////////////////////////////////////////////////////////////////////
+BaseConnectorResource::BaseConnectorResource( Connector* connector )
+{
+ this->connector = connector;
+ this->closed = false;
+}
+
+///////////////////////////////////////////////////////////////////////////////
+BaseConnectorResource::~BaseConnectorResource()
+{}
+
+///////////////////////////////////////////////////////////////////////////////
+void BaseConnectorResource::close() throw ( cms::CMSException )
+{
+ try {
+
+ if( closed == true ) {
+ return;
+ }
+
+ if( connector != NULL ) {
+ // Close the connector resource
+ connector->closeResource( this );
+ }
+
+ // We are now closed
+ this->closed = true;
+
+ // Notify the listeners
+ std::set< ConnectorResourceListener* >::const_iterator iter =
+ listeners.begin();
+
+ for(; iter != listeners.end(); ++iter ) {
+ try{
+ (*iter)->onConnectorResourceClosed( this );
+ } catch( ... ) {
+ // Absorb
+ }
+ }
+ }
+ AMQ_CATCH_RETHROW( ActiveMQException )
+ AMQ_CATCHALL_THROW( ActiveMQException )
+}
+
+///////////////////////////////////////////////////////////////////////////////
+void BaseConnectorResource::addListener( ConnectorResourceListener* listener ) {
+ listeners.insert( listener );
+}
+
+///////////////////////////////////////////////////////////////////////////////
+void BaseConnectorResource::removeListener( ConnectorResourceListener* listener ) {
+ listeners.erase( listener );
+}
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.h?view=auto&rev=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.h (added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.h Sat Mar 17 16:54:01 2007
@@ -0,0 +1,111 @@
+/*
+ * 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.
+ */
+
+#ifndef _ACTIVEMQ_CONNECTOR_BASECONNECTORRESOURCE_H_
+#define _ACTIVEMQ_CONNECTOR_BASECONNECTORRESOURCE_H_
+
+#include <activemq/connector/ConnectorResource.h>
+#include <activemq/connector/ConnectorResourceListener.h>
+
+#include <set>
+
+namespace activemq{
+namespace connector{
+
+ class Connector;
+
+ /**
+ * Base Class that implements the common functionality of a
+ * <code>ConnectorResource</code>.
+ */
+ class BaseConnectorResource : public ConnectorResource {
+
+ private:
+
+ /** The main Connector to call back when closed */
+ Connector* connector;
+
+ /** Set of ConnectorResourceListeners to call back */
+ std::set< ConnectorResourceListener* > listeners;
+
+ /** Have we been closed already */
+ bool closed;
+
+ public:
+
+ /**
+ * Default Constructor
+ */
+ BaseConnectorResource();
+
+ /**
+ * Constructor - Accepts a Connector to call for Close if this
+ * resource.
+ * @param connector - The Connector to close on.
+ */
+ BaseConnectorResource( Connector* connector );
+
+ virtual ~BaseConnectorResource();
+
+ /**
+ * Handles the close of the Resource which calls back to the
+ * Connector to close this resource then notifies the listeners
+ * that this resource has closed.
+ * @throws CMSException
+ */
+ virtual void close() throw( cms::CMSException );
+
+ /**
+ * @ Returns true if this resource has been closed already
+ */
+ virtual bool isClosed() const {
+ return this->closed;
+ }
+
+ /**
+ * Adds a new listener to this Connector Resource
+ * @param listener - Listener of ConnectorResource events to add
+ */
+ virtual void addListener( ConnectorResourceListener* listener );
+
+ /**
+ * Removes a new listener to this Connector Resource
+ * @param listener - Listener of ConnectorResource events to remove
+ */
+ virtual void removeListener( ConnectorResourceListener* listener );
+
+ /**
+ * Gets the Connector that this resource is associated with
+ * @returns Connector pointer to this resources Connector
+ */
+ virtual Connector* getConnector() const {
+ return this->connector;
+ }
+
+ /**
+ * Sets the Connector that this Resouce is associated with.
+ * @param connector - The Connector the resource is associated with.
+ */
+ virtual void setConnector( Connector* connector ) {
+ this->connector = connector;
+ }
+
+ };
+
+}}
+
+#endif /*_ACTIVEMQ_CONNECTOR_BASECONNECTORRESOURCE_H_*/
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/BaseConnectorResource.h
------------------------------------------------------------------------------
svn:eol-style = native
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/Connector.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/Connector.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/Connector.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/Connector.h Sat Mar 17 16:54:01 2007
@@ -326,11 +326,12 @@
throw ( ConnectorException, exceptions::UnsupportedOperationException ) = 0;
/**
- * Destroys the given connector resource.
- * @param resource the resource to be destroyed.
+ * Closes the given connector resource, caller must still delete
+ * the resource once its been closed.
+ * @param resource the resource to be closed
* @throws ConnectorException
*/
- virtual void destroyResource( ConnectorResource* resource )
+ virtual void closeResource( ConnectorResource* resource )
throw ( ConnectorException ) = 0;
/**
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConnectorResource.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConnectorResource.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConnectorResource.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConnectorResource.h Sat Mar 17 16:54:01 2007
@@ -18,20 +18,54 @@
#ifndef ACTIVEMQ_CONNECTOR_CONNECTORRESOURCE_H_
#define ACTIVEMQ_CONNECTOR_CONNECTORRESOURCE_H_
+#include <cms/Closeable.h>
+#include <activemq/connector/ConnectorResourceListener.h>
+
namespace activemq{
namespace connector{
+ class Connector;
+
/**
- * An object who's lifetime is determined by
- * the connector that created it. All ConnectorResources
- * should be given back to the connector rather than
- * deleting explicitly.
+ * A Connector Resource is an object that is created in the Connector
+ * and must alert the connector to is closing so that the connector
+ * can clean up and resouces that are associated with the resouce.
+ * The lifetime of the ConnectorResource is still managed by the
+ * class that owns it, but it must be close on destruction.
+ * <p>
+ * The Connector Resouce should notify its owner of its close by
+ * calling back to its owner through a registered
+ * <code>ConnectorResouceListener</code> from its owner.
*/
- class ConnectorResource
- {
+ class ConnectorResource : public cms::Closeable {
+
public:
virtual ~ConnectorResource() {}
+
+ /**
+ * Adds a new listener to this Connector Resource
+ * @param listener - Listener of ConnectorResource events to add
+ */
+ virtual void addListener( ConnectorResourceListener* listener ) = 0;
+
+ /**
+ * Removes a new listener to this Connector Resource
+ * @param listener - Listener of ConnectorResource events to remove
+ */
+ virtual void removeListener( ConnectorResourceListener* listener ) = 0;
+
+ /**
+ * Gets the Connector that this resource is associated with
+ * @returns Connector pointer to this resources Connector
+ */
+ virtual Connector* getConnector() const = 0;
+
+ /**
+ * Sets the Connector that this Resouce is associated with.
+ * @param connector - The Connector the resource is associated with.
+ */
+ virtual void setConnector( Connector* connector ) = 0;
};
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConsumerInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConsumerInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConsumerInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ConsumerInfo.h Sat Mar 17 16:54:01 2007
@@ -17,7 +17,7 @@
#ifndef _ACTIVEMQ_CONNECTOR_CONSUMERINFO_H_
#define _ACTIVEMQ_CONNECTOR_CONSUMERINFO_H_
-#include <activemq/connector/ConnectorResource.h>
+#include <activemq/connector/BaseConnectorResource.h>
#include <activemq/connector/SessionInfo.h>
#include <cms/Destination.h>
#include <string>
@@ -25,17 +25,22 @@
namespace activemq{
namespace connector{
- class ConsumerInfo : public ConnectorResource
+ class ConsumerInfo : public BaseConnectorResource
{
public:
- virtual ~ConsumerInfo(void) {}
+ ConsumerInfo() : BaseConnectorResource() {}
+
+ ConsumerInfo( Connector* connector ) :
+ BaseConnectorResource( connector ) {}
+
+ virtual ~ConsumerInfo() {}
/**
* Gets this message consumer's message selector expression.
* @return This Consumer's selector expression or "".
*/
- virtual const std::string& getMessageSelector(void) const = 0;
+ virtual const std::string& getMessageSelector() const = 0;
/**
* Sets this message consumer's message selector expression.
@@ -47,7 +52,7 @@
* Gets the ID that is assigned to this consumer
* @return value of the Consumer Id.
*/
- virtual long long getConsumerId(void) const = 0;
+ virtual long long getConsumerId() const = 0;
/**
* Sets the ID that is assigned to this consumer
@@ -59,7 +64,7 @@
* Gets the Destination that this Consumer is subscribed on
* @return Destination
*/
- virtual const cms::Destination* getDestination(void) const = 0;
+ virtual const cms::Destination* getDestination() const = 0;
/**
* Sets the destination that this Consumer is listening on
@@ -71,7 +76,7 @@
* Gets the Session Info that this consumer is attached too
* @return SessionnInfo pointer
*/
- virtual const SessionInfo* getSessionInfo(void) const = 0;
+ virtual const SessionInfo* getSessionInfo() const = 0;
/**
* Gets the Session Info that this consumer is attached too
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ProducerInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ProducerInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ProducerInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/ProducerInfo.h Sat Mar 17 16:54:01 2007
@@ -19,24 +19,29 @@
#include <cms/Destination.h>
-#include <activemq/connector/ConnectorResource.h>
+#include <activemq/connector/BaseConnectorResource.h>
#include <activemq/connector/SessionInfo.h>
namespace activemq{
namespace connector{
- class ProducerInfo : public ConnectorResource
+ class ProducerInfo : public BaseConnectorResource
{
public:
- virtual ~ProducerInfo(void) {}
+ ProducerInfo() : BaseConnectorResource() {}
+
+ ProducerInfo( Connector* connector ) :
+ BaseConnectorResource( connector ) {}
+
+ virtual ~ProducerInfo() {}
/**
* Retrieves the default destination that this producer
* sends its messages to.
* @return Destionation, owned by this object
*/
- virtual const cms::Destination* getDestination(void) const = 0;
+ virtual const cms::Destination* getDestination() const = 0;
/**
* Sets the Default Destination for this Producer
@@ -48,7 +53,7 @@
* Gets the ID that is assigned to this Producer
* @return value of the Producer Id.
*/
- virtual long long getProducerId(void) const = 0;
+ virtual long long getProducerId() const = 0;
/**
* Sets the ID that is assigned to this Producer
@@ -60,7 +65,7 @@
* Gets the Session Info that this consumer is attached too
* @return SessionnInfo pointer
*/
- virtual const SessionInfo* getSessionInfo(void) const = 0;
+ virtual const SessionInfo* getSessionInfo() const = 0;
/**
* Gets the Session Info that this consumer is attached too
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/SessionInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/SessionInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/SessionInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/SessionInfo.h Sat Mar 17 16:54:01 2007
@@ -17,25 +17,30 @@
#ifndef _ACTIVEMQ_CONNECTOR_SESSIONINFO_H_
#define _ACTIVEMQ_CONNECTOR_SESSIONINFO_H_
-#include <activemq/connector/ConnectorResource.h>
+#include <activemq/connector/BaseConnectorResource.h>
#include <activemq/connector/TransactionInfo.h>
#include <cms/Session.h>
namespace activemq{
namespace connector{
- class SessionInfo : public ConnectorResource
+ class SessionInfo : public BaseConnectorResource
{
public:
- virtual ~SessionInfo(void) {}
+ SessionInfo() : BaseConnectorResource() {}
+
+ SessionInfo( Connector* connector ) :
+ BaseConnectorResource( connector ) {}
+
+ virtual ~SessionInfo(void) {}
/**
* Gets the Connection Id of the Connection that this consumer is
* using to receive its messages.
* @return string value of the connection id
*/
- virtual const std::string& getConnectionId(void) const = 0;
+ virtual const std::string& getConnectionId() const = 0;
/**
* Sets the Connection Id of the Connection that this consumer is
@@ -48,7 +53,7 @@
* Gets the Sessions Id value
* @return id for this session
*/
- virtual long long getSessionId(void) const = 0;
+ virtual long long getSessionId() const = 0;
/**
* Sets the Session Id for this Session
@@ -66,7 +71,7 @@
* Gets the Ack Mode of this Session
* @return Ack Mode
*/
- virtual cms::Session::AcknowledgeMode getAckMode(void) const = 0;
+ virtual cms::Session::AcknowledgeMode getAckMode() const = 0;
/**
* Gets the currently active transaction info, if this session is
@@ -74,14 +79,15 @@
* getAckMode and see if the session is transacted.
* @return Transaction Id of current Transaction
*/
- virtual const TransactionInfo* getTransactionInfo(void) const = 0;
+ virtual const TransactionInfo* getTransactionInfo() const = 0;
/**
* Sets the current transaction info for this session, this is nit
* used when the session is not transacted.
* @param transaction Transaction Id
*/
- virtual void setTransactionInfo( const TransactionInfo* transaction ) = 0;
+ virtual void setTransactionInfo(
+ const TransactionInfo* transaction ) = 0;
};
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/TransactionInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/TransactionInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/TransactionInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/TransactionInfo.h Sat Mar 17 16:54:01 2007
@@ -17,24 +17,29 @@
#ifndef _ACTIVEMQ_CONNECTOR_TRANSACTIONINFO_H_
#define _ACTIVEMQ_CONNECTOR_TRANSACTIONINFO_H_
-#include <activemq/connector/ConnectorResource.h>
+#include <activemq/connector/BaseConnectorResource.h>
namespace activemq{
namespace connector{
class SessionInfo;
- class TransactionInfo : public ConnectorResource
+ class TransactionInfo : public BaseConnectorResource
{
public:
- virtual ~TransactionInfo(void) {}
+ TransactionInfo() : BaseConnectorResource() {}
+
+ TransactionInfo( Connector* connector ) :
+ BaseConnectorResource( connector ) {}
+
+ virtual ~TransactionInfo() {}
/**
* Gets the Transction Id
* @return unsigned int Id
*/
- virtual long long getTransactionId(void) const = 0;
+ virtual long long getTransactionId() const = 0;
/**
* Sets the Transction Id
@@ -46,7 +51,7 @@
* Gets the Session Info that this transaction is attached too
* @return SessionnInfo pointer
*/
- virtual const SessionInfo* getSessionInfo(void) const = 0;
+ virtual const SessionInfo* getSessionInfo() const = 0;
/**
* Gets the Session Info that this transaction is attached too
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.cpp Sat Mar 17 16:54:01 2007
@@ -329,7 +329,7 @@
sessionId->setConnectionId( connectionInfo.getConnectionId()->getValue() );
sessionId->setValue( getNextSessionId() );
info->setSessionId( sessionId );
- OpenWireSessionInfo* session = new OpenWireSessionInfo();
+ OpenWireSessionInfo* session = new OpenWireSessionInfo( this );
try{
@@ -377,7 +377,7 @@
{
enforceConnected();
- consumer = new OpenWireConsumerInfo();
+ consumer = new OpenWireConsumerInfo( this );
consumer->setSessionInfo( session );
consumerInfo = createConsumerInfo( destination, session );
consumer->setConsumerInfo( consumerInfo );
@@ -430,7 +430,7 @@
{
enforceConnected();
- consumer = new OpenWireConsumerInfo();
+ consumer = new OpenWireConsumerInfo( this );
consumer->setSessionInfo( session );
consumerInfo = createConsumerInfo( topic, session );
consumer->setConsumerInfo( consumerInfo );
@@ -706,7 +706,7 @@
{
enforceConnected();
- producer = new OpenWireProducerInfo();
+ producer = new OpenWireProducerInfo( this );
producer->setSessionInfo( session );
producerInfo = new commands::ProducerInfo();
@@ -1042,7 +1042,7 @@
enforceConnected();
OpenWireTransactionInfo* transaction =
- new OpenWireTransactionInfo();
+ new OpenWireTransactionInfo( this );
// Place Transaction Data in session for later use as well as
// the session in the Transaction Data
@@ -1246,7 +1246,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-void OpenWireConnector::destroyResource( ConnectorResource* resource )
+void OpenWireConnector::closeResource( ConnectorResource* resource )
throw ( ConnectorException )
{
try
@@ -1273,7 +1273,6 @@
// Unstarted Consumers can just be deleted.
if( consumer->isStarted() == false ) {
- delete resource;
return;
}
@@ -1287,7 +1286,7 @@
dynamic_cast<OpenWireSessionInfo*>(resource);
dataStructure = session->getSessionInfo()->getSessionId();
} else if( typeid( *resource ) == typeid( OpenWireTransactionInfo ) ) {
- delete resource;
+ // Nothing to do for Transaction Info's
return;
} else if( tempDestination != NULL ) {
// User deletes these
@@ -1302,9 +1301,6 @@
// Dispose of this data structure at the broker.
disposeOf( dataStructure );
-
- // No matter what we end it here.
- delete resource;
}
catch( ConnectorException& ex ) {
delete resource;
@@ -1498,6 +1494,9 @@
// The broker did not return an error - this is good.
// Just discard the response.
delete response;
+
+ // Now that its setup, link it to this Connector
+ tempDestination->setConnector( this );
}
AMQ_CATCH_RETHROW( ConnectorException )
AMQ_CATCHALL_THROW( OpenWireConnectorException )
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConnector.h Sat Mar 17 16:54:01 2007
@@ -539,11 +539,12 @@
throw ( ConnectorException, exceptions::UnsupportedOperationException );
/**
- * Destroys the given connector resource.
- * @param resource the resource to be destroyed.
+ * Closes the given connector resource, caller must still delete
+ * the resource once its been closed.
+ * @param resource the resource to be closed
* @throws ConnectorException
*/
- virtual void destroyResource( ConnectorResource* resource )
+ virtual void closeResource( ConnectorResource* resource )
throw ( ConnectorException );
/**
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConsumerInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConsumerInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConsumerInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireConsumerInfo.h Sat Mar 17 16:54:01 2007
@@ -41,12 +41,17 @@
public:
- OpenWireConsumerInfo() {
+ OpenWireConsumerInfo( Connector* connector ) :
+ ConsumerInfo( connector ) {
+
this->session = NULL;
this->started = false;
}
- virtual ~OpenWireConsumerInfo() {}
+ virtual ~OpenWireConsumerInfo() {
+ this->close();
+ delete consumerInfo;
+ }
/**
* Gets this message consumer's message selector expression.
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireProducerInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireProducerInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireProducerInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireProducerInfo.h Sat Mar 17 16:54:01 2007
@@ -41,20 +41,25 @@
public:
- OpenWireProducerInfo() {
+ OpenWireProducerInfo( Connector* connector ) :
+ ProducerInfo( connector ) {
+
this->disableMessageIds = false;
this->producerInfo = NULL;
this->session = NULL;
}
- virtual ~OpenWireProducerInfo() {}
+ virtual ~OpenWireProducerInfo() {
+ this->close();
+ delete producerInfo;
+ }
/**
* Retrieves the default destination that this producer
* sends its messages to.
* @return Destionation, owned by this object
*/
- virtual const cms::Destination* getDestination(void) const {
+ virtual const cms::Destination* getDestination() const {
if( this->producerInfo != NULL ) {
return this->producerInfo->getDestination()->getCMSDestination();
}
@@ -79,7 +84,7 @@
* Gets the ID that is assigned to this Producer
* @return value of the Producer Id.
*/
- virtual long long getProducerId(void) const {
+ virtual long long getProducerId() const {
if( this->producerInfo != NULL ) {
return (unsigned int)
this->producerInfo->getProducerId()->getValue();
@@ -102,7 +107,7 @@
* Gets the Session Info that this consumer is attached too
* @return SessionnInfo pointer
*/
- virtual const SessionInfo* getSessionInfo(void) const {
+ virtual const SessionInfo* getSessionInfo() const {
return session;
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireSessionInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireSessionInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireSessionInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireSessionInfo.h Sat Mar 17 16:54:01 2007
@@ -45,12 +45,15 @@
public:
- OpenWireSessionInfo() {
+ OpenWireSessionInfo( Connector* connector ) :
+ SessionInfo( connector ) {
+
ackMode = cms::Session::AUTO_ACKNOWLEDGE;
transaction = NULL;
sessionInfo = NULL;
}
virtual ~OpenWireSessionInfo() {
+ this->close();
delete sessionInfo;
}
@@ -59,7 +62,7 @@
* using to receive its messages.
* @return string value of the connection id
*/
- virtual const std::string& getConnectionId(void) const{
+ virtual const std::string& getConnectionId() const{
if( sessionInfo != NULL ) {
connectionId = sessionInfo->getSessionId()->getConnectionId();
}
@@ -80,7 +83,7 @@
* Gets the Sessions Id value
* @return id for this session
*/
- virtual long long getSessionId(void) const {
+ virtual long long getSessionId() const {
if( sessionInfo != NULL ) {
return (unsigned int)sessionInfo->getSessionId()->getValue();
}
@@ -110,7 +113,7 @@
* Gets the Ack Mode of this Session
* @return Ack Mode
*/
- virtual cms::Session::AcknowledgeMode getAckMode(void) const {
+ virtual cms::Session::AcknowledgeMode getAckMode() const {
return ackMode;
}
@@ -120,7 +123,7 @@
* getAckMode and see if the session is transacted.
* @return Transaction Id of current Transaction
*/
- virtual const TransactionInfo* getTransactionInfo(void) const {
+ virtual const TransactionInfo* getTransactionInfo() const {
return transaction;
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireTransactionInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireTransactionInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireTransactionInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/OpenWireTransactionInfo.h Sat Mar 17 16:54:01 2007
@@ -38,12 +38,15 @@
public:
- OpenWireTransactionInfo() {
+ OpenWireTransactionInfo( Connector* connector ) :
+ TransactionInfo( connector ) {
+
transactionInfo = NULL;
session = NULL;
}
virtual ~OpenWireTransactionInfo() {
+ this->close();
delete transactionInfo;
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.cpp Sat Mar 17 16:54:01 2007
@@ -41,30 +41,10 @@
////////////////////////////////////////////////////////////////////////////////
ActiveMQTempDestination::~ActiveMQTempDestination()
{
- try{
- close();
- }
- AMQ_CATCH_NOTHROW( ActiveMQException )
- AMQ_CATCHALL_NOTHROW( )
}
////////////////////////////////////////////////////////////////////////////////
unsigned char ActiveMQTempDestination::getDataStructureType() const
{
return ActiveMQTempDestination::ID_ACTIVEMQTEMPDESTINATION;
-}
-
-////////////////////////////////////////////////////////////////////////////////
-void ActiveMQTempDestination::close() throw ( cms::CMSException ) {
-
- try {
-
- // Give the Connector a chance to clean up this temp Destination
- if( connector != NULL ) {
- connector->destroyResource( this );
- }
-
- }
- AMQ_CATCH_RETHROW( ActiveMQException )
- AMQ_CATCHALL_THROW( ActiveMQException )
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempDestination.h Sat Mar 17 16:54:01 2007
@@ -25,7 +25,7 @@
#include <activemq/connector/openwire/commands/ActiveMQDestination.h>
#include <activemq/connector/Connector.h>
-#include <activemq/connector/ConnectorResource.h>
+#include <activemq/connector/BaseConnectorResource.h>
#include <cms/Closeable.h>
#include <vector>
#include <string>
@@ -36,8 +36,7 @@
namespace commands{
class ActiveMQTempDestination : public ActiveMQDestination,
- public ConnectorResource,
- public cms::Closeable
+ public BaseConnectorResource
{
protected:
@@ -91,33 +90,6 @@
*/
virtual bool equals( const DataStructure* value ) const {
return ActiveMQDestination::equals( value );
- }
-
- public: // Closeable
-
- /**
- * Closes this object and deallocates the appropriate resources.
- * The object is generally no longer usable after calling close.
- * @throws CMSException
- */
- virtual void close() throw( cms::CMSException );
-
- public: // ActiveMQTempDestination
-
- /**
- * Gets the Connector that this object will call when its closed
- * @returns pointer to a Connector, or NULL if not set.
- */
- const connector::Connector* getConnector() const {
- return this->connector;
- }
-
- /**
- * Sets the Connector that this object will call when its closed
- * @param connector - pointer to a Connector or NULL to clear.
- */
- virtual void setConnector( connector::Connector* connector ) {
- this->connector = connector;
}
};
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.cpp Sat Mar 17 16:54:01 2007
@@ -35,6 +35,10 @@
////////////////////////////////////////////////////////////////////////////////
ActiveMQTempQueue::~ActiveMQTempQueue()
{
+ try {
+ this->close();
+ }
+ AMQ_CATCHALL_NOTHROW()
}
////////////////////////////////////////////////////////////////////////////////
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempQueue.h Sat Mar 17 16:54:01 2007
@@ -163,6 +163,14 @@
return this->getPhysicalName();
}
+ /**
+ * Destroy's the Temp Destination at the Broker
+ * @throws CMSException
+ */
+ virtual void destroy() throw ( cms::CMSException ) {
+ close();
+ }
+
};
}}}}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.cpp Sat Mar 17 16:54:01 2007
@@ -35,6 +35,10 @@
////////////////////////////////////////////////////////////////////////////////
ActiveMQTempTopic::~ActiveMQTempTopic()
{
+ try {
+ this->close();
+ }
+ AMQ_CATCHALL_NOTHROW()
}
////////////////////////////////////////////////////////////////////////////////
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTempTopic.h Sat Mar 17 16:54:01 2007
@@ -160,6 +160,14 @@
return this->getPhysicalName();
}
+ /**
+ * Destroy's the Temp Destination at the Broker
+ * @throws CMSException
+ */
+ virtual void destroy() throw ( cms::CMSException ) {
+ close();
+ }
+
};
}}}}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.cpp Sat Mar 17 16:54:01 2007
@@ -229,7 +229,7 @@
// Connected so we now create the SessionManager
sessionManager = new StompSessionManager(
- connected->getSessionId(), transport );
+ connected->getSessionId(), this, transport );
// Give our message listener to the session manager it will
// notify all the interested clients
@@ -341,7 +341,7 @@
{
enforceConnected();
- ProducerInfo* producer = new StompProducerInfo();
+ ProducerInfo* producer = new StompProducerInfo( this );
producer->setDestination( destination );
producer->setProducerId( getNextProducerId() );
@@ -541,7 +541,7 @@
{
enforceConnected();
- TransactionInfo* transaction = new StompTransactionInfo();
+ TransactionInfo* transaction = new StompTransactionInfo( this );
transaction->setTransactionId( getNextTransactionId() );
@@ -720,7 +720,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::destroyResource( ConnectorResource* resource )
+void StompConnector::closeResource( ConnectorResource* resource )
throw ( ConnectorException )
{
try
@@ -756,9 +756,6 @@
throw ex;
}
}
-
- // All went well - finish by deleting the resource.
- delete resource;
}
AMQ_CATCH_RETHROW( ConnectorException )
AMQ_CATCHALL_THROW( ConnectorException );
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConnector.h Sat Mar 17 16:54:01 2007
@@ -456,11 +456,12 @@
throw ( ConnectorException, exceptions::UnsupportedOperationException );
/**
- * Destroys the given connector resource.
- * @param resource resource the resource to be destroyed.
+ * Closes the given connector resource, caller must still delete
+ * the resource once its been closed.
+ * @param resource the resource to be closed
* @throws ConnectorException
*/
- virtual void destroyResource( ConnectorResource* resource )
+ virtual void closeResource( ConnectorResource* resource )
throw ( ConnectorException );
/**
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConsumerInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConsumerInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConsumerInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompConsumerInfo.h Sat Mar 17 16:54:01 2007
@@ -18,6 +18,8 @@
#ifndef _ACTIVEMQ_CONNECTOR_STOMP_STOMPCONSUMERINFO_H_
#define _ACTIVEMQ_CONNECTOR_STOMP_STOMPCONSUMERINFO_H_
+#include <activemq/connector/ConsumerInfo.h>
+
namespace activemq{
namespace connector{
namespace stomp{
@@ -40,20 +42,33 @@
public:
- StompConsumerInfo(void) {
+ StompConsumerInfo() : ConsumerInfo() {
+
+ selector = "";
+ consumerId = 0;
+ destination = NULL;
+ session = NULL;
+ }
+
+ StompConsumerInfo( Connector* connector ) :
+ ConsumerInfo( connector ) {
+
selector = "";
consumerId = 0;
destination = NULL;
session = NULL;
}
- virtual ~StompConsumerInfo(void) { delete destination; }
+ virtual ~StompConsumerInfo() {
+ this->close();
+ delete destination;
+ }
/**
* Gets this message consumer's message selector expression.
* @return This Consumer's selector expression or "".
*/
- virtual const std::string& getMessageSelector(void) const {
+ virtual const std::string& getMessageSelector() const {
return selector;
}
@@ -69,7 +84,7 @@
* Gets the ID that is assigned to this consumer
* @return value of the Consumer Id.
*/
- virtual long long getConsumerId(void) const {
+ virtual long long getConsumerId() const {
return consumerId;
}
@@ -85,7 +100,7 @@
* Gets the Destination that this Consumer is subscribed on
* @return Destination this consumer is attached to
*/
- virtual const cms::Destination* getDestination(void) const {
+ virtual const cms::Destination* getDestination() const {
return destination;
}
@@ -101,7 +116,7 @@
* Gets the Session Info that this consumer is attached too
* @return SessionnInfo pointer
*/
- virtual const SessionInfo* getSessionInfo(void) const {
+ virtual const SessionInfo* getSessionInfo() const {
return session;
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompProducerInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompProducerInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompProducerInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompProducerInfo.h Sat Mar 17 16:54:01 2007
@@ -43,13 +43,27 @@
public:
- StompProducerInfo(void) {
+ StompProducerInfo() : ProducerInfo() {
+
+ this->producerId = 0;
+ this->disableMessageIds = false;
+ this->session = NULL;
+ this->destination = NULL;
+ }
+
+ StompProducerInfo( Connector* connector ) :
+ ProducerInfo( connector ) {
+
this->producerId = 0;
this->disableMessageIds = false;
this->session = NULL;
this->destination = NULL;
}
- virtual ~StompProducerInfo(void) { delete destination; }
+
+ virtual ~StompProducerInfo(void) {
+ this->close();
+ delete destination;
+ }
/**
* Retrieves the default destination that this producer
@@ -65,11 +79,11 @@
* @param destination reference to a destination, copied internally
*/
virtual void setDestination( const cms::Destination* destination ) {
-
+
// Delete the previous destination if it exists.
delete this->destination;
this->destination = NULL;
-
+
if( destination != NULL ) {
this->destination = destination->clone();
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionInfo.h Sat Mar 17 16:54:01 2007
@@ -25,7 +25,7 @@
namespace connector{
namespace stomp{
- class StompSessionInfo : public connector::SessionInfo
+ class StompSessionInfo : public SessionInfo
{
private:
@@ -44,21 +44,31 @@
public:
- StompSessionInfo(void)
- {
+ StompSessionInfo() : SessionInfo() {
+
+ sessionId = 0;
+ ackMode = cms::Session::AUTO_ACKNOWLEDGE;
+ transaction = NULL;
+ }
+
+ StompSessionInfo( Connector* connector ) :
+ SessionInfo( connector ) {
+
sessionId = 0;
ackMode = cms::Session::AUTO_ACKNOWLEDGE;
transaction = NULL;
}
- virtual ~StompSessionInfo(void) {}
+ virtual ~StompSessionInfo() {
+ this->close();
+ }
/**
* Gets the Connection Id of the Connection that this consumer is
* using to receive its messages.
* @return string value of the connection id
*/
- virtual const std::string& getConnectionId(void) const{
+ virtual const std::string& getConnectionId() const{
return connectionId;
}
@@ -75,7 +85,7 @@
* Gets the Sessions Id value
* @return id for this session
*/
- virtual long long getSessionId(void) const {
+ virtual long long getSessionId() const {
return sessionId;
}
@@ -99,7 +109,7 @@
* Gets the Ack Mode of this Session
* @return Ack Mode
*/
- virtual cms::Session::AcknowledgeMode getAckMode(void) const {
+ virtual cms::Session::AcknowledgeMode getAckMode() const {
return ackMode;
}
@@ -109,7 +119,7 @@
* getAckMode and see if the session is transacted.
* @return Transaction Id of current Transaction
*/
- virtual const TransactionInfo* getTransactionInfo(void) const {
+ virtual const TransactionInfo* getTransactionInfo() const {
return transaction;
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.cpp Sat Mar 17 16:54:01 2007
@@ -40,6 +40,7 @@
////////////////////////////////////////////////////////////////////////////////
StompSessionManager::StompSessionManager( const std::string& connectionId,
+ Connector* connector,
Transport* transport )
{
if( transport == NULL )
@@ -50,6 +51,7 @@
}
this->transport = transport;
+ this->connector = connector;
this->connectionId = connectionId;
this->nextSessionId = 0;
this->nextConsumerId = 0;
@@ -95,7 +97,7 @@
{
try
{
- SessionInfo* session = new StompSessionInfo();
+ SessionInfo* session = new StompSessionInfo( connector );
// Init data
session->setAckMode( ackMode );
@@ -201,7 +203,7 @@
}
// Initialize a new Consumer info Message
- ConsumerInfo* consumer = new StompConsumerInfo();
+ ConsumerInfo* consumer = new StompConsumerInfo( connector );
consumer->setConsumerId( getNextConsumerId() );
consumer->setDestination( destination );
@@ -369,7 +371,7 @@
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_NOLOCAL );
- if( destProperties.hasProperty( noLocalStr ) )
+ if( destProperties.hasProperty( noLocalStr ) )
{
command.setNoLocal(
Boolean::parseBoolean(
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompSessionManager.h Sat Mar 17 16:54:01 2007
@@ -23,6 +23,7 @@
#include <activemq/transport/Transport.h>
#include <activemq/concurrent/Mutex.h>
#include <activemq/connector/ConnectorException.h>
+#include <activemq/connector/Connector.h>
#include <activemq/connector/stomp/StompCommandListener.h>
#include <activemq/connector/ConsumerMessageListener.h>
#include <activemq/connector/stomp/commands/SubscribeCommand.h>
@@ -73,11 +74,16 @@
// The global connection id
std::string connectionId;
+ // Connector that we are working for
+ connector::Connector* connector;
+
public:
- StompSessionManager( const std::string& connectionId,
+ StompSessionManager( const std::string& connectionId,
+ connector::Connector* connector,
transport::Transport* transport );
- virtual ~StompSessionManager(void);
+
+ virtual ~StompSessionManager();
/**
* Creates a new Session and returns a SessionInfo object whose
@@ -184,13 +190,13 @@
* Gets the Next Session Id
* @return unique session id
*/
- virtual long long getNextSessionId(void);
+ virtual long long getNextSessionId();
/**
* Gets the Next Session Id
* @return unique session id
*/
- virtual long long getNextConsumerId(void);
+ virtual long long getNextConsumerId();
};
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompTransactionInfo.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompTransactionInfo.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompTransactionInfo.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/stomp/StompTransactionInfo.h Sat Mar 17 16:54:01 2007
@@ -24,7 +24,7 @@
namespace connector{
namespace stomp{
- class StompTransactionInfo : public connector::TransactionInfo
+ class StompTransactionInfo : public TransactionInfo
{
private:
@@ -36,15 +36,22 @@
public:
- /**
- * Default Constructor
- */
- StompTransactionInfo(void) {
+ StompTransactionInfo() : TransactionInfo() {
+
transactionId = 0;
session = NULL;
}
- virtual ~StompTransactionInfo(void) {}
+ StompTransactionInfo( Connector* connector ) :
+ TransactionInfo( connector ) {
+
+ transactionId = 0;
+ session = NULL;
+ }
+
+ virtual ~StompTransactionInfo() {
+ this->close();
+ }
/**
* Gets the Transction Id
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp Sat Mar 17 16:54:01 2007
@@ -255,11 +255,6 @@
synchronized( &activeSessions ) {
activeSessions.remove( session );
}
-
- // Destroy this sessions resources
- getConnectionData()->
- getConnector()->destroyResource( session->getSessionInfo() );
-
}
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.cpp?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.cpp Sat Mar 17 16:54:01 2007
@@ -18,6 +18,7 @@
#include <activemq/exceptions/NullPointerException.h>
#include <activemq/exceptions/InvalidStateException.h>
+#include <activemq/exceptions/IllegalArgumentException.h>
#include <activemq/core/ActiveMQSession.h>
#include <activemq/core/ActiveMQMessage.h>
#include <cms/ExceptionListener.h>
@@ -26,6 +27,7 @@
using namespace cms;
using namespace activemq;
using namespace activemq::core;
+using namespace activemq::connector;
using namespace activemq::exceptions;
using namespace activemq::concurrent;
@@ -39,13 +41,16 @@
__FILE__, __LINE__,
"ActiveMQConsumer::ActiveMQConsumer - Init with NULL Session");
}
-
+
// Init Producer Data
this->session = session;
this->consumerInfo = consumerInfo;
this->listenerThread = NULL;
this->listener = NULL;
this->closed = false;
+
+ // Listen for our resource to close
+ this->consumerInfo->addListener( this );
}
////////////////////////////////////////////////////////////////////////////////
@@ -54,69 +59,70 @@
try
{
close();
+
+ delete consumerInfo;
}
AMQ_CATCH_NOTHROW( ActiveMQException )
AMQ_CATCHALL_NOTHROW( )
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQConsumer::close()
+void ActiveMQConsumer::close()
throw ( cms::CMSException )
{
try
{
if( !closed ) {
-
- closed = true;
-
+
// Identifies any errors encountered during shutdown.
bool haveException = false;
- ActiveMQException error;
-
- // Dispose of the Consumer Info, this should stop us from getting
- // any more messages. This may result in message traffic
- // going to the connector. If the socket is broken, this
- // will result in an exception, in which case we catch it
- // and continue to shutdown normally.
- try{
- session->onDestroySessionResource( this );
- } catch( ActiveMQException& ex ){
- if( !haveException ){
- ex.setMark( __FILE__, __LINE__ );
- error = ex;
- haveException = true;
+ ActiveMQException error;
+
+ // Close the ConsumerInfo
+ if( !consumerInfo->isClosed() ) {
+ try{
+ // We don't want a callback now
+ this->consumerInfo->removeListener( this );
+ this->consumerInfo->close();
+ } catch ( ActiveMQException& ex ){
+ if( !haveException ){
+ ex.setMark( __FILE__, __LINE__ );
+ error = ex;
+ haveException = true;
+ }
}
}
-
+
+ closed = true;
+
// Stop the asynchronous message processin thread if it's
// running.
try{
stopThread();
} catch ( ActiveMQException& ex ){
- if( !haveException ){
- ex.setMark( __FILE__, __LINE__ );
+ if( !haveException ){
+ ex.setMark( __FILE__, __LINE__ );
error = ex;
haveException = true;
}
}
-
+
// Purge all the pending messages
try{
purgeMessages();
} catch ( ActiveMQException& ex ){
- if( !haveException ){
- ex.setMark( __FILE__, __LINE__ );
+ if( !haveException ){
+ ex.setMark( __FILE__, __LINE__ );
error = ex;
haveException = true;
}
}
-
+
// If we encountered an error, propagate it.
if( haveException ){
error.setMark( __FILE__, __LINE__ );
throw error;
}
-
}
}
AMQ_CATCH_RETHROW( ActiveMQException )
@@ -124,7 +130,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-std::string ActiveMQConsumer::getMessageSelector() const
+std::string ActiveMQConsumer::getMessageSelector() const
throw ( cms::CMSException )
{
try
@@ -156,7 +162,7 @@
{
msgQueue.wait();
}
-
+
// This will only happen when this object is being
// closed in another thread context - kind of
// scary.
@@ -164,7 +170,7 @@
throw ActiveMQException( __FILE__, __LINE__,
"Consumer is being closed in another thread" );
}
-
+
// Fetch the Message then copy it so it can be handed off
// to the user.
cms::Message* message = msgQueue.pop();
@@ -172,7 +178,7 @@
// The Message is cleaned up here if the Session is not
// transacted, otherwise we let the transaction clean up
- // this message as it will have already been ack'd and
+ // this message as it will have already been ack'd and
// stored for later redelivery.
destroyMessage( message );
@@ -186,7 +192,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-cms::Message* ActiveMQConsumer::receive( int millisecs )
+cms::Message* ActiveMQConsumer::receive( int millisecs )
throw ( cms::CMSException )
{
try
@@ -202,7 +208,7 @@
{
// Check for empty, and wait if its not
if( !closed && msgQueue.empty() ){
-
+
msgQueue.wait(millisecs);
// if its still empty...bail
@@ -210,7 +216,7 @@
return NULL;
}
}
-
+
// This will only happen when this object is being
// closed in another thread context - kind of
// scary.
@@ -226,7 +232,7 @@
// The Message is cleaned up here if the Session is not
// transacted, otherwise we let the transaction clean up
- // this message as it will have already been ack'd and
+ // this message as it will have already been ack'd and
// stored for later redelivery.
destroyMessage( message );
@@ -240,7 +246,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-cms::Message* ActiveMQConsumer::receiveNoWait()
+cms::Message* ActiveMQConsumer::receiveNoWait()
throw ( cms::CMSException )
{
try
@@ -263,14 +269,14 @@
// The Message is cleaned up here if the Session is not
// transacted, otherwise we let the transaction clean up
- // this message as it will have already been ack'd and
+ // this message as it will have already been ack'd and
// stored for later redelivery.
destroyMessage( message );
return result;
}
}
-
+
return NULL;
}
AMQ_CATCH_RETHROW( ActiveMQException )
@@ -293,12 +299,12 @@
{
this->listener = listener;
}
-
+
// Start the thread if it isn't already running.
// If it is already running, this method will wake the thread up
// to notify it that there is a message listener, so that it may
// get rid of backed up messages.
- startThread();
+ startThread();
}
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
@@ -337,7 +343,7 @@
synchronized( &msgQueue )
{
-
+
// Gaurd against spurious wakeup or race to sync lock
// also if the listner has been unregistered we don't
// have anyone to notify, so we wait till a new one is
@@ -350,23 +356,23 @@
}
msgQueue.wait();
}
-
+
// don't want to process messages if we are shutting down.
if( closed )
{
return;
}
-
+
// Dispatch the message
message = msgQueue.pop();
}
-
+
// Notify the listener
- notifyListener( message );
-
+ notifyListener( message );
+
// The Message is cleaned up here if the Session is not
// transacted, otherwise we let the transaction clean up
- // this message as it will have already been ack'd and
+ // this message as it will have already been ack'd and
// stored for later redelivery.
destroyMessage( message );
}
@@ -374,7 +380,7 @@
catch( ... )
{
cms::ExceptionListener* listener = session->getExceptionListener();
-
+
if( listener != NULL )
{
listener->onException( ActiveMQException(
@@ -382,11 +388,11 @@
"ActiveMQConsumer::run - "
"MessageListener threw an unknown Exception, recovering..." ) );
}
- }
+ }
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQConsumer::dispatch( ActiveMQMessage* message )
+void ActiveMQConsumer::dispatch( ActiveMQMessage* message )
throw ( cms::CMSException )
{
try
@@ -399,23 +405,23 @@
// stop now, don't queue
return;
}
-
- // If the Session is in ClientAcknowledge mode, then we set the
+
+ // If the Session is in ClientAcknowledge mode, then we set the
// handler in the message to this object and send it out. Otherwise
// we ack it here for all the other Modes.
if( session->getAcknowledgeMode() == Session::CLIENT_ACKNOWLEDGE ) {
-
+
// Register ourself so that we can handle the Message's
// acknowledge method.
message->setAckHandler( this );
-
+
} else {
session->acknowledge( this, message );
}
// No listener, so we queue it
synchronized( &msgQueue ) {
-
+
msgQueue.push( dynamic_cast< cms::Message* >( message ) );
msgQueue.notifyAll();
}
@@ -434,7 +440,7 @@
while( !msgQueue.empty() )
{
// destroy these messages if this is not a transacted
- // session, if it is then the tranasction will clean
+ // session, if it is then the tranasction will clean
// the messages up.
destroyMessage( msgQueue.pop() );
}
@@ -458,33 +464,33 @@
}
this->dispatch( message );
- }
+ }
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
}
////////////////////////////////////////////////////////////////////////////////
void ActiveMQConsumer::notifyListener( Message* message ) throw ( ActiveMQException ){
-
+
try
{
MessageListener* listener = NULL;
synchronized( &listenerLock )
{
- listener = getMessageListener();
+ listener = getMessageListener();
}
if(listener != NULL)
{
listener->onMessage( message );
}
- }
+ }
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
}
////////////////////////////////////////////////////////////////////////////////
void ActiveMQConsumer::destroyMessage( Message* message ) throw ( ActiveMQException ){
-
+
try
{
/**
@@ -495,42 +501,42 @@
{
delete message;
}
- }
+ }
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
}
////////////////////////////////////////////////////////////////////////////////
void ActiveMQConsumer::startThread() throw ( ActiveMQException ) {
-
+
try
{
// Start the thread, if it's not already started.
if( listenerThread == NULL )
{
- listenerThread = new Thread( this );
- listenerThread->start();
+ listenerThread = new Thread( this );
+ listenerThread->start();
}
-
+
// notify the Queue so that any pending messages get delivered
synchronized( &msgQueue )
{
msgQueue.notifyAll();
}
- }
+ }
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
}
////////////////////////////////////////////////////////////////////////////////
void ActiveMQConsumer::stopThread() throw ( ActiveMQException ) {
-
+
try
{
// if the thread is running signal it to quit and then
// wait for run to return so thread can die
if( listenerThread != NULL )
- {
+ {
synchronized( &msgQueue )
{
// Force a wakeup if run is in a wait.
@@ -542,8 +548,35 @@
delete listenerThread;
listenerThread = NULL;
}
- }
+ }
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
}
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQConsumer::onConnectorResourceClosed(
+ const ConnectorResource* resource ) throw ( cms::CMSException ) {
+
+ try{
+
+ if( closed )
+ {
+ throw InvalidStateException(
+ __FILE__, __LINE__,
+ "ActiveMQConsumer::onConnectorResourceClosed - "
+ "Producer Already Closed");
+ }
+
+ if( resource != consumerInfo ) {
+ throw IllegalArgumentException(
+ __FILE__, __LINE__,
+ "ActiveMQConsumer::onConnectorResourceClosed - "
+ "Unknown object passed to this callback");
+ }
+
+ // If our producer isn't closed already, then lets close
+ this->close();
+ }
+ AMQ_CATCH_RETHROW( ActiveMQException )
+ AMQ_CATCHALL_THROW( ActiveMQException )
+}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.h?view=diff&rev=519474&r1=519473&r2=519474
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConsumer.h Sat Mar 17 16:54:01 2007
@@ -23,10 +23,10 @@
#include <cms/CMSException.h>
#include <activemq/connector/ConsumerInfo.h>
+#include <activemq/connector/ConnectorResourceListener.h>
#include <activemq/util/Queue.h>
#include <activemq/core/ActiveMQAckHandler.h>
#include <activemq/core/ActiveMQMessageListener.h>
-#include <activemq/core/ActiveMQSessionResource.h>
#include <activemq/concurrent/Runnable.h>
#include <activemq/concurrent/Mutex.h>
@@ -35,36 +35,36 @@
class ActiveMQSession;
- class ActiveMQConsumer :
+ class ActiveMQConsumer :
public cms::MessageConsumer,
public ActiveMQAckHandler,
public concurrent::Runnable,
public ActiveMQMessageListener,
- public ActiveMQSessionResource
+ public connector::ConnectorResourceListener
{
private:
-
+
// The session that owns this Consumer
ActiveMQSession* session;
-
+
// The Consumer info for this Consumer
connector::ConsumerInfo* consumerInfo;
-
+
// The Message Listener for this Consumer
cms::MessageListener* listener;
-
+
// Lock to protect us from dispatching to a dead listener
concurrent::Mutex listenerLock;
-
+
// Message Queue
util::Queue<cms::Message*> msgQueue;
-
+
// Thread to notif a listener if one is added
concurrent::Thread* listenerThread;
-
+
// Boolean that indicates if the consumer has been closed
bool closed;
-
+
public:
/**
@@ -85,7 +85,7 @@
* @throws CMSException
*/
virtual void close() throw ( cms::CMSException );
-
+
/**
* Synchronously Receive a Message
* @return new message
@@ -129,9 +129,9 @@
* @return This Consumer's selector expression or "".
* @throws cms::CMSException
*/
- virtual std::string getMessageSelector() const
+ virtual std::string getMessageSelector() const
throw ( cms::CMSException );
-
+
/**
* Method called to acknowledge the message passed
* @param message the Message to Acknowlegde
@@ -145,26 +145,26 @@
* is registered with a Thread and started. This function reads from
* the message queue and dispatches calls to the MessageConsumer that
* is registered with this class.
- *
+ *
* It is a error for a MessageListener to throw an exception in their
* onMessage method, but if it does happen this function will get any
* registered exception listener from the session and notify it.
- */
+ */
virtual void run();
public: // ActiveMQMessageListener Methods
-
+
/**
* Called asynchronously when a new message is received, the message
* that is passed is now the property of the callee, and the caller
* will disavowe all knowledge of the message, i.e Callee must delete.
* @param message object pointer
*/
- virtual void onActiveMQMessage( ActiveMQMessage* message )
+ virtual void onActiveMQMessage( ActiveMQMessage* message )
throw ( exceptions::ActiveMQException );
-
+
public: // ActiveMQSessionResource
-
+
/**
* Retrieve the Connector resource that is associated with
* this Session resource.
@@ -175,7 +175,7 @@
}
public: // ActiveMQConsumer Methods
-
+
/**
* Called to dispatch a message to this consumer, this is usually
* called from the context of another thread. This will enqueue a
@@ -184,40 +184,51 @@
* @param message cms::Message pointer to the message to dispatch
* @throws cms::CMSException
*/
- virtual void dispatch( ActiveMQMessage* message )
+ virtual void dispatch( ActiveMQMessage* message )
throw ( cms::CMSException );
/**
* Get the Consumer information for this consumer
- * @return Pointer to a Consumer Info Object
+ * @return Pointer to a Consumer Info Object
*/
virtual connector::ConsumerInfo* getConsumerInfo() {
return consumerInfo;
}
+ protected: // ConnectorResourceListener
+
+ /**
+ * When a Connector Resouce is closed it will notify any registered
+ * Listeners of its close so that they can take the appropriate
+ * action.
+ * @param resource - The ConnectorResource that was closed.
+ */
+ virtual void onConnectorResourceClosed(
+ const connector::ConnectorResource* resource ) throw ( cms::CMSException );
+
protected:
-
+
/**
* Purges all messages currently in the queue. This can be as a
* result of a rollback, or of the consumer being shutdown.
*/
virtual void purgeMessages() throw (exceptions::ActiveMQException);
-
+
/**
* Destroys the message if the session is transacted, otherwise
* does nothing.
* @param message the message to destroy
*/
- virtual void destroyMessage( cms::Message* message )
+ virtual void destroyMessage( cms::Message* message )
throw (exceptions::ActiveMQException);
/**
* Notifies the listener of a message.
* @param message the message to pass to the listener
*/
- void notifyListener( cms::Message* message )
+ void notifyListener( cms::Message* message )
throw (exceptions::ActiveMQException);
-
+
/**
* Starts the message processing thread to receive messages
* asynchronously. This thread is started when setMessageListener
@@ -225,7 +236,7 @@
* consumer asynchronously instead of synchronously (receive).
*/
void startThread() throw (exceptions::ActiveMQException);
-
+
/**
* Stops the asynchronous message processing thread if it's started.
*/