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/02/15 20:42:06 UTC
svn commit: r508103 [1/2] - in
/activemq/activemq-cpp/trunk/activemq-cpp/src: main/activemq/connector/
main/activemq/connector/openwire/
main/activemq/connector/openwire/commands/ main/activemq/connector/stomp/
main/activemq/core/ test/activemq/connect...
Author: tabish
Date: Thu Feb 15 11:42:05 2007
New Revision: 508103
URL: http://svn.apache.org/viewvc?view=rev&rev=508103
Log:
http://issues.apache.org/activemq/browse/AMQCPP-30
Modified:
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/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/commands/ActiveMQDestination.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQQueue.h
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.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTopic.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/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/StompSessionManager.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQProducer.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
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/core/ActiveMQConnectionTest.h
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -30,42 +30,42 @@
public:
virtual ~ConsumerInfo(void) {}
-
+
/**
* Gets this message consumer's message selector expression.
* @return This Consumer's selector expression or "".
*/
virtual const std::string& getMessageSelector(void) const = 0;
-
+
/**
* Sets this message consumer's message selector expression.
* @param selector This Consumer's selector expression or "".
*/
- virtual void setMessageSelector( const std::string& selector ) = 0;
+ virtual void setMessageSelector( const std::string& selector ) = 0;
/**
* Gets the ID that is assigned to this consumer
* @return value of the Consumer Id.
*/
virtual unsigned int getConsumerId(void) const = 0;
-
+
/**
* Sets the ID that is assigned to this consumer
* @return id string value of the Consumer Id.
*/
virtual void setConsumerId( const unsigned int id ) = 0;
-
+
/**
* Gets the Destination that this Consumer is subscribed on
* @return Destination
*/
- virtual const cms::Destination& getDestination(void) const = 0;
-
+ virtual const cms::Destination* getDestination(void) const = 0;
+
/**
* Sets the destination that this Consumer is listening on
* @param destination Destination
*/
- virtual void setDestination( const cms::Destination& destination ) = 0;
+ virtual void setDestination( const cms::Destination* destination ) = 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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -30,26 +30,26 @@
public:
virtual ~ProducerInfo(void) {}
-
+
/**
* 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(void) const = 0;
+
/**
* Sets the Default Destination for this Producer
* @param destination reference to a destination, copied internally
*/
- virtual void setDestination( const cms::Destination& destination ) = 0;
+ virtual void setDestination( const cms::Destination* destination ) = 0;
/**
* Gets the ID that is assigned to this Producer
* @return value of the Producer Id.
*/
virtual unsigned int getProducerId(void) const = 0;
-
+
/**
* Sets the ID that is assigned to this Producer
* @return id string value of the Producer Id.
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -95,20 +95,24 @@
* Gets the Destination that this Consumer is subscribed on
* @return Destination this consumer is attached to
*/
- virtual const cms::Destination& getDestination(void) const {
- //return *(this->consumerInfo->getDestination());
+ virtual const cms::Destination* getDestination(void) const {
+ if( this->consumerInfo != NULL ) {
+ return this->consumerInfo->getDestination()->getCMSDestination();
+ }
+
+ return NULL;
}
/**
* Sets the destination that this Consumer is listening on
* @param destination Destination this consumer is attached to
*/
- virtual void setDestination( const cms::Destination& destination ) {
+ virtual void setDestination( const cms::Destination* destination ) {
if( consumerInfo != NULL ) {
this->consumerInfo->setDestination(
dynamic_cast<commands::ActiveMQDestination*>(
- destination.clone() ) );
+ destination->clone() ) );
}
}
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -49,20 +49,24 @@
* sends its messages to.
* @return Destionation, owned by this object
*/
- virtual const cms::Destination& getDestination(void) const {
- //return *destination;
+ virtual const cms::Destination* getDestination(void) const {
+ if( this->producerInfo != NULL ) {
+ return this->producerInfo->getDestination()->getCMSDestination();
+ }
+
+ return NULL;
}
/**
* Sets the Default Destination for this Producer
* @param destination reference to a destination, copied internally
*/
- virtual void setDestination( const cms::Destination& destination ) {
+ virtual void setDestination( const cms::Destination* destination ) {
if( this->producerInfo != NULL ) {
this->producerInfo->setDestination(
dynamic_cast<commands::ActiveMQDestination*>(
- destination.clone() ) );
+ destination->clone() ) );
}
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQDestination.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQDestination.h?view=diff&rev=508103&r1=508102&r2=508103
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQDestination.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQDestination.h Thu Feb 15 11:42:05 2007
@@ -51,32 +51,32 @@
static const std::string ANY_CHILD;
static const std::string ANY_DESCENDENT;
};
-
+
/**
* prefix for Advisory message destinations
*/
static const std::string ADVISORY_PREFIX;
-
+
/**
* prefix for consumer advisory destinations
*/
static const std::string CONSUMER_ADVISORY_PREFIX;
-
+
/**
* prefix for producer advisory destinations
*/
static const std::string PRODUCER_ADVISORY_PREFIX;
-
+
/**
* prefix for connection advisory destinations
*/
static const std::string CONNECTION_ADVISORY_PREFIX;
-
+
/**
* The default target for ordered destinations
*/
static const std::string DEFAULT_ORDERED_TARGET;
-
+
static const std::string TEMP_PREFIX;
static const std::string TEMP_POSTFIX;
static const std::string COMPOSITE_SEPARATOR;
@@ -86,7 +86,7 @@
bool ordered;
bool advisory;
std::string orderedTarget;
-
+
std::string physicalName;
util::SimpleProperties options;
@@ -132,52 +132,52 @@
virtual bool isAdvisory() const {
return advisory;
}
-
+
/**
* @param advisory The advisory to set.
*/
virtual void setAdvisory( bool advisory ){
this->advisory = advisory;
}
-
+
/**
* @return true if this is a destination for Consumer advisories
*/
virtual bool isConsumerAdvisory() const {
- return isAdvisory() &&
+ return isAdvisory() &&
physicalName.find(CONSUMER_ADVISORY_PREFIX) == 0;
}
-
+
/**
* @return true if this is a destination for Producer advisories
*/
virtual bool isProducerAdvisory() const {
- return isAdvisory() &&
+ return isAdvisory() &&
physicalName.find(PRODUCER_ADVISORY_PREFIX) == 0;
}
-
+
/**
* @return true if this is a destination for Connection advisories
*/
virtual bool isConnectionAdvisory() const {
- return isAdvisory() &&
+ return isAdvisory() &&
physicalName.find(CONNECTION_ADVISORY_PREFIX) == 0;
}
-
+
/**
* @return Returns the exclusive.
*/
virtual bool isExclusive() const {
return exclusive;
}
-
+
/**
* @param exclusive The exclusive to set.
*/
virtual void setExclusive( bool exclusive ) {
this->exclusive = exclusive;
}
-
+
/**
* @return Returns the ordered.
*/
@@ -207,19 +207,19 @@
}
public: // Statics
-
+
/**
* Create a temporary name from the clientId
*
* @param clientId
* @return
*/
- static std::string createTemporaryName(
+ static std::string createTemporaryName(
const std::string& clientId ) {
return TEMP_PREFIX + clientId + TEMP_POSTFIX;
}
-
+
/**
* From a temporary destination find the clientId of the Connection that created it
*
@@ -229,7 +229,7 @@
static std::string getClientId( const ActiveMQDestination* destination );
public:
-
+
/**
* Returns the Type of Destination that this object represents
* @returns int type qualifier.
@@ -239,21 +239,21 @@
/**
* Returns true if a temporary Destination
* @return true/false
- */
+ */
virtual bool isTemporary() const {
return getDestinationType() == cms::Destination::TEMPORARY_TOPIC ||
getDestinationType() == cms::Destination::TEMPORARY_QUEUE;
}
-
+
/**
* Returns true if a Topic Destination
* @return true/false
- */
+ */
virtual bool isTopic() const {
- return getDestinationType() == cms::Destination::TOPIC ||
+ return getDestinationType() == cms::Destination::TOPIC ||
getDestinationType() == cms::Destination::TEMPORARY_TOPIC;
}
-
+
/**
* Returns true if a Queue Destination
* @return true/false
@@ -261,16 +261,16 @@
virtual bool isQueue() const {
return !isTopic();
}
-
+
/**
* Returns true if this destination represents a collection of
- * destinations; allowing a set of destinations to be published to or
+ * destinations; allowing a set of destinations to be published to or
* subscribed from in one CMS operation.
* <p/>
- * If this destination is a composite then you can call
+ * If this destination is a composite then you can call
* {@link #getChildDestinations()}
* to return the list of child destinations.
- * @return true if this destination represents a collection of child
+ * @return true if this destination represents a collection of child
* destinations.
*/
virtual bool isComposite() const {
@@ -284,12 +284,20 @@
return physicalName.find_first_of( DestinationFilter::ANY_CHILD ) != std::string::npos ||
physicalName.find_first_of( DestinationFilter::ANY_DESCENDENT ) != std::string::npos;
}
-
+
/**
* @returns a reference (const) to the options properties for this Dest.
*/
const activemq::util::Properties& getOptions() const {
return options;
+ }
+
+ /**
+ * @returns the cms::Destination interface pointer that the
+ * objects that derive from this class implement.
+ */
+ virtual const cms::Destination* getCMSDestination() const {
+ return NULL;
}
};
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQQueue.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQQueue.h?view=diff&rev=508103&r1=508102&r2=508103
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQQueue.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQQueue.h Thu Feb 15 11:42:05 2007
@@ -93,6 +93,14 @@
return ActiveMQDestination::equals( value );
}
+ /**
+ * @returns the cms::Destination interface pointer that the
+ * objects that derive from this class implement.
+ */
+ virtual const cms::Destination* getCMSDestination() const {
+ return this;
+ }
+
public: // CMS Destination
/**
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -95,6 +95,14 @@
return ActiveMQDestination::equals( value );
}
+ /**
+ * @returns the cms::Destination interface pointer that the
+ * objects that derive from this class implement.
+ */
+ virtual const cms::Destination* getCMSDestination() const {
+ return this;
+ }
+
public: // CMS Destination
/**
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -92,6 +92,14 @@
return ActiveMQDestination::equals( value );
}
+ /**
+ * @returns the cms::Destination interface pointer that the
+ * objects that derive from this class implement.
+ */
+ virtual const cms::Destination* getCMSDestination() const {
+ return this;
+ }
+
public: // CMS Destination
/**
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTopic.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTopic.h?view=diff&rev=508103&r1=508102&r2=508103
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTopic.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/connector/openwire/commands/ActiveMQTopic.h Thu Feb 15 11:42:05 2007
@@ -93,6 +93,14 @@
return ActiveMQDestination::equals( value );
}
+ /**
+ * @returns the cms::Destination interface pointer that the
+ * objects that derive from this class implement.
+ */
+ virtual const cms::Destination* getCMSDestination() const {
+ return this;
+ }
+
public: // CMS Destination
/**
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
+
#include <activemq/connector/stomp/StompConnector.h>
#include <activemq/concurrent/Concurrent.h>
#include <activemq/transport/BrokerError.h>
@@ -51,7 +51,7 @@
LOGCMS_INITIALIZE(logger, StompConnector, "activemq.connector.stomp.StompConnector" )
////////////////////////////////////////////////////////////////////////////////
-StompConnector::StompConnector( Transport* transport,
+StompConnector::StompConnector( Transport* transport,
const util::Properties& properties )
throw ( IllegalArgumentException )
{
@@ -61,7 +61,7 @@
__FILE__, __LINE__,
"StompConnector::StompConnector - Transport cannot be NULL");
}
-
+
this->transport = transport;
this->state = DISCONNECTED;
this->exceptionListener = NULL;
@@ -70,7 +70,7 @@
this->nextProducerId = 1;
this->nextTransactionId = 1;
this->properties.copy( &properties );
-
+
// Observe the transport for events.
this->transport->setCommandListener( this );
this->transport->setTransportExceptionListener( this );
@@ -79,7 +79,7 @@
this->transport->setCommandReader( &reader );
this->transport->setCommandWriter( &writer );
- // Register ourself for those commands that we process
+ // Register ourself for those commands that we process
addCmdListener( CommandConstants::ERROR_CMD, this );
}
@@ -89,7 +89,7 @@
try
{
close();
-
+
delete sessionManager;
}
AMQ_CATCH_NOTHROW( ActiveMQException )
@@ -103,7 +103,7 @@
{
return nextProducerId++;
}
-
+
return 0;
}
@@ -114,7 +114,7 @@
{
return nextTransactionId++;
}
-
+
return 0;
}
@@ -130,7 +130,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::addCmdListener(
+void StompConnector::addCmdListener(
commands::CommandConstants::CommandId commandId,
StompCommandListener* listener )
{
@@ -138,7 +138,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::removeCmdListener(
+void StompConnector::removeCmdListener(
commands::CommandConstants::CommandId commandId )
{
cmdListenerMap.erase( commandId );
@@ -153,17 +153,17 @@
{
if( state == CONNECTED )
{
- throw ActiveMQException(
- __FILE__, __LINE__,
+ throw ActiveMQException(
+ __FILE__, __LINE__,
"StompConnector::start - already started" );
}
-
+
// Start the transport - this establishes the socket.
transport->start();
// Send the connect message to the broker.
- connect();
- }
+ connect();
+ }
}
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException );
@@ -171,18 +171,18 @@
////////////////////////////////////////////////////////////////////////////////
void StompConnector::close() throw( cms::CMSException ){
-
+
try
{
if( state == DISCONNECTED ){
return;
}
-
+
synchronized( &mutex )
{
// Send the disconnect message to the broker.
disconnect();
-
+
transport->close();
}
}
@@ -198,35 +198,35 @@
// Mark this connector as started.
state = CONNECTING;
- // Send the connect command to the broker
+ // Send the connect command to the broker
ConnectCommand cmd;
// Encode User Name and Password and Client ID
string login = getUsername();
if( login.length() > 0 ){
cmd.setLogin( login );
- }
+ }
string password = getPassword();
if( password.length() > 0 ){
cmd.setPassword( password );
- }
+ }
string clientId = getClientId();
if( clientId.length() > 0 ){
cmd.setClientId( clientId );
}
Response* response = transport->request( &cmd );
-
+
if( dynamic_cast< ExceptionResponse* >( response ) != NULL )
{
delete response;
-
+
throw StompConnectorException(
__FILE__, __LINE__,
"StompConnector::connect - Failed on Connect Request" );
}
- ConnectedCommand* connected =
+ ConnectedCommand* connected =
dynamic_cast< ConnectedCommand* >( response );
if( connected == NULL )
@@ -236,7 +236,7 @@
throw StompConnectorException(
__FILE__, __LINE__,
"StompConnector::connect - "
- "Response not a connected response" );
+ "Response not a connected response" );
}
// Connected so we now create the SessionManager
@@ -244,24 +244,24 @@
connected->getSessionId(), transport );
// Give our message listener to the session manager it will
- // notify all the interested clients
+ // notify all the interested clients
sessionManager->setConsumerMessageListener( messageListener );
- // Add the Session Manager as the Command Listener for
- // Message commands so that it can route them to the
+ // Add the Session Manager as the Command Listener for
+ // Message commands so that it can route them to the
// correct consumers.
addCmdListener( CommandConstants::MESSAGE, sessionManager );
-
+
// In Stomp, the client Id is the same as the session id that is
// returned in the Connected response
- properties.setProperty(
- commands::CommandConstants::toString(
+ properties.setProperty(
+ commands::CommandConstants::toString(
commands::CommandConstants::HEADER_CLIENT_ID ),
- connected->getSessionId() );
+ connected->getSessionId() );
// Tag us in the Connected State now.
state = CONNECTED;
-
+
// Clean up
delete response;
}
@@ -292,13 +292,13 @@
////////////////////////////////////////////////////////////////////////////////
SessionInfo* StompConnector::createSession(
- cms::Session::AcknowledgeMode ackMode )
+ cms::Session::AcknowledgeMode ackMode )
throw( ConnectorException )
{
try
{
enforceConnected();
-
+
return sessionManager->createSession( ackMode );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -307,7 +307,7 @@
////////////////////////////////////////////////////////////////////////////////
ConsumerInfo* StompConnector::createConsumer(
- const cms::Destination* destination,
+ const cms::Destination* destination,
SessionInfo* session,
const std::string& selector,
bool noLocal )
@@ -316,8 +316,8 @@
try
{
enforceConnected();
-
- return sessionManager->createConsumer(
+
+ return sessionManager->createConsumer(
destination, session, selector, noLocal );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -326,7 +326,7 @@
////////////////////////////////////////////////////////////////////////////////
ConsumerInfo* StompConnector::createDurableConsumer(
- const cms::Topic* topic,
+ const cms::Topic* topic,
SessionInfo* session,
const std::string& name,
const std::string& selector,
@@ -336,8 +336,8 @@
try
{
enforceConnected();
-
- return sessionManager->createDurableConsumer(
+
+ return sessionManager->createDurableConsumer(
topic, session, name, selector, noLocal );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -346,20 +346,20 @@
////////////////////////////////////////////////////////////////////////////////
ProducerInfo* StompConnector::createProducer(
- const cms::Destination* destination,
+ const cms::Destination* destination,
SessionInfo* session )
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
ProducerInfo* producer = new StompProducerInfo();
-
- producer->setDestination( *destination );
+
+ producer->setDestination( destination );
producer->setProducerId( getNextProducerId() );
producer->setSessionInfo( session );
-
+
return producer;
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -367,14 +367,14 @@
}
////////////////////////////////////////////////////////////////////////////////
-cms::Topic* StompConnector::createTopic( const std::string& name,
+cms::Topic* StompConnector::createTopic( const std::string& name,
SessionInfo* session AMQCPP_UNUSED)
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
return new StompTopic( name );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -382,14 +382,14 @@
}
////////////////////////////////////////////////////////////////////////////////
-cms::Queue* StompConnector::createQueue( const std::string& name,
+cms::Queue* StompConnector::createQueue( const std::string& name,
SessionInfo* session AMQCPP_UNUSED)
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
return new StompQueue( name );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -404,7 +404,7 @@
try
{
throw UnsupportedOperationException(
- __FILE__, __LINE__,
+ __FILE__, __LINE__,
"StompConnector::createTemporaryTopic - No Stomp Support");
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -420,7 +420,7 @@
try
{
throw UnsupportedOperationException(
- __FILE__, __LINE__,
+ __FILE__, __LINE__,
"StompConnector::createTemporaryQueue - No Stomp Support");
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -429,17 +429,17 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::send( cms::Message* message,
- ProducerInfo* producerInfo )
+void StompConnector::send( cms::Message* message,
+ ProducerInfo* producerInfo )
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
const SessionInfo* session = producerInfo->getSessionInfo();
Command* command = dynamic_cast< transport::Command* >( message );
-
+
if( command == NULL )
{
throw StompConnectorException(
@@ -450,7 +450,7 @@
if( session->getAckMode() == cms::Session::SESSION_TRANSACTED )
{
- StompCommand* stompCommand =
+ StompCommand* stompCommand =
dynamic_cast< StompCommand* >( message );
if( stompCommand == NULL )
@@ -460,18 +460,18 @@
"StompConnector::send - "
"Message is not a valid stomp type.");
}
-
+
stompCommand->setTransactionId(
- Integer::toString(
+ Integer::toString(
session->getTransactionInfo()->getTransactionId() ) );
}
-
+
// Send it
transport->oneway( command );
}
catch( CommandIOException& ex ){
transport->close();
- throw ConnectorException( __FILE__, __LINE__,
+ throw ConnectorException( __FILE__, __LINE__,
ex.what() );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -480,15 +480,15 @@
////////////////////////////////////////////////////////////////////////////////
void StompConnector::send( std::list<cms::Message*>& messages,
- ProducerInfo* producerInfo )
+ ProducerInfo* producerInfo )
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
list< cms::Message* >::const_iterator itr = messages.begin();
-
+
for( ; itr != messages.end(); ++itr )
{
this->send( *itr, producerInfo );
@@ -507,10 +507,10 @@
try
{
enforceConnected();
-
+
// Auto to Stomp means don't do anything, so we drop it here
- // for client acknowledge we have to send and ack.
- if( session->getAckMode() == cms::Session::CLIENT_ACKNOWLEDGE ||
+ // for client acknowledge we have to send and ack.
+ if( session->getAckMode() == cms::Session::CLIENT_ACKNOWLEDGE ||
session->getAckMode() == cms::Session::SESSION_TRANSACTED )
{
AckCommand cmd;
@@ -527,17 +527,17 @@
if( session->getAckMode() == cms::Session::SESSION_TRANSACTED )
{
- cmd.setTransactionId(
- Integer::toString(
+ cmd.setTransactionId(
+ Integer::toString(
session->getTransactionInfo()->getTransactionId() ) );
}
- transport->oneway( &cmd );
+ transport->oneway( &cmd );
}
}
catch( CommandIOException& ex ){
transport->close();
- throw ConnectorException( __FILE__, __LINE__,
+ throw ConnectorException( __FILE__, __LINE__,
ex.what() );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -546,22 +546,22 @@
////////////////////////////////////////////////////////////////////////////////
TransactionInfo* StompConnector::startTransaction(
- SessionInfo* session )
+ SessionInfo* session )
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
TransactionInfo* transaction = new StompTransactionInfo();
-
+
transaction->setTransactionId( getNextTransactionId() );
-
+
session->setTransactionInfo( transaction );
BeginCommand cmd;
- cmd.setTransactionId(
+ cmd.setTransactionId(
Integer::toString( transaction->getTransactionId() ) );
transport->oneway( &cmd );
@@ -570,7 +570,7 @@
}
catch( CommandIOException& ex ){
transport->close();
- throw ConnectorException( __FILE__, __LINE__,
+ throw ConnectorException( __FILE__, __LINE__,
ex.what() );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -579,24 +579,24 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::commit( TransactionInfo* transaction,
+void StompConnector::commit( TransactionInfo* transaction,
SessionInfo* session AMQCPP_UNUSED)
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
CommitCommand cmd;
-
- cmd.setTransactionId(
+
+ cmd.setTransactionId(
Integer::toString( transaction->getTransactionId() ) );
-
+
transport->oneway( &cmd );
}
catch( CommandIOException& ex ){
transport->close();
- throw ConnectorException( __FILE__, __LINE__,
+ throw ConnectorException( __FILE__, __LINE__,
ex.what() );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -604,24 +604,24 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::rollback( TransactionInfo* transaction,
+void StompConnector::rollback( TransactionInfo* transaction,
SessionInfo* session AMQCPP_UNUSED)
throw ( ConnectorException )
{
try
{
enforceConnected();
-
+
AbortCommand cmd;
-
- cmd.setTransactionId(
+
+ cmd.setTransactionId(
Integer::toString( transaction->getTransactionId() ) );
-
+
transport->oneway( &cmd );
}
catch( CommandIOException& ex ){
transport->close();
- throw ConnectorException( __FILE__, __LINE__,
+ throw ConnectorException( __FILE__, __LINE__,
ex.what() );
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -637,15 +637,15 @@
try
{
enforceConnected();
-
+
MessageCommand* cmd = new MessageCommand();
-
+
if( transaction != NULL )
{
- cmd->setTransactionId(
+ cmd->setTransactionId(
Integer::toString( transaction->getTransactionId() ) );
}
-
+
return cmd;
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -661,15 +661,15 @@
try
{
enforceConnected();
-
+
BytesMessageCommand* cmd = new BytesMessageCommand();
-
+
if( transaction != NULL )
{
- cmd->setTransactionId(
+ cmd->setTransactionId(
Integer::toString( transaction->getTransactionId() ) );
}
-
+
return cmd;
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -685,15 +685,15 @@
try
{
enforceConnected();
-
+
TextMessageCommand* cmd = new TextMessageCommand;
-
+
if( transaction != NULL )
{
- cmd->setTransactionId(
+ cmd->setTransactionId(
Integer::toString( transaction->getTransactionId() ) );
}
-
+
return cmd;
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -709,7 +709,7 @@
try
{
throw UnsupportedOperationException(
- __FILE__, __LINE__,
+ __FILE__, __LINE__,
"StompConnector::createTemporaryQueue - No Stomp Support");
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -724,7 +724,7 @@
try
{
throw UnsupportedOperationException(
- __FILE__, __LINE__,
+ __FILE__, __LINE__,
"StompConnector::createTemporaryQueue - No Stomp Support");
}
AMQ_CATCH_RETHROW( ConnectorException )
@@ -737,9 +737,9 @@
{
try
{
- ConsumerInfo* consumer =
+ ConsumerInfo* consumer =
dynamic_cast<ConsumerInfo*>(resource);
- SessionInfo* session =
+ SessionInfo* session =
dynamic_cast<SessionInfo*>(resource);
if( consumer != NULL)
@@ -747,10 +747,10 @@
try{
sessionManager->removeConsumer( consumer );
} catch( ConnectorException& ex ){
-
+
// Make sure we delete the resource
delete resource;
-
+
ex.setMark( __FILE__, __LINE__ );
throw ex;
}
@@ -760,10 +760,10 @@
try{
sessionManager->removeSession( session );
} catch( ConnectorException& ex ){
-
+
// Make sure we delete the resource
delete resource;
-
+
ex.setMark( __FILE__, __LINE__ );
throw ex;
}
@@ -790,9 +790,9 @@
"StompConnector::onCommand - Recieved an unknown Command") );
}
- CmdListenerMap::iterator itr =
+ CmdListenerMap::iterator itr =
cmdListenerMap.find( stompCommand->getStompCommandId() );
-
+
if( itr == cmdListenerMap.end() )
{
fire( ConnectorException(
@@ -805,17 +805,17 @@
return; // we are done
}
-
+
// Hand off
- itr->second->onStompCommand( stompCommand );
+ itr->second->onStompCommand( stompCommand );
}
AMQ_CATCH_RETHROW( ConnectorException )
AMQ_CATCHALL_THROW( ConnectorException );
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::onTransportException(
- transport::Transport* source AMQCPP_UNUSED,
+void StompConnector::onTransportException(
+ transport::Transport* source AMQCPP_UNUSED,
const exceptions::ActiveMQException& ex )
{
try
@@ -824,10 +824,10 @@
if( state == DISCONNECTED ){
return;
}
-
+
// We were not closing - log the stack trace.
LOGCMS_WARN(logger, ex.getStackTraceString() );
-
+
// Inform the user of the error.
fire( ex );
}
@@ -836,25 +836,25 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompConnector::onStompCommand( commands::StompCommand* command )
+void StompConnector::onStompCommand( commands::StompCommand* command )
throw ( StompConnectorException )
{
try
- {
- ErrorCommand* error =
+ {
+ ErrorCommand* error =
dynamic_cast<ErrorCommand*>( command );
-
+
if( error != NULL )
{
fire( StompConnectorException(
__FILE__, __LINE__,
- ( string( "StompConnector::onStompCommand - " ) +
+ ( string( "StompConnector::onStompCommand - " ) +
error->getErrorMessage() ).c_str() ) );
-
+
// Shutdown
close();
}
-
+
// command is done here, delete it.
delete command;
}
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -85,16 +85,16 @@
* Gets the Destination that this Consumer is subscribed on
* @return Destination this consumer is attached to
*/
- virtual const cms::Destination& getDestination(void) const {
- return *destination;
+ virtual const cms::Destination* getDestination(void) const {
+ return destination;
}
/**
* Sets the destination that this Consumer is listening on
* @param destination Destination this consumer is attached to
*/
- virtual void setDestination( const cms::Destination& destination ) {
- this->destination = destination.clone();
+ virtual void setDestination( const cms::Destination* destination ) {
+ this->destination = destination->clone();
}
/**
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -52,16 +52,16 @@
* sends its messages to.
* @return Destionation, owned by this object
*/
- virtual const cms::Destination& getDestination(void) const {
- return *destination;
+ virtual const cms::Destination* getDestination(void) const {
+ return destination;
}
/**
* Sets the Default Destination for this Producer
* @param destination reference to a destination, copied internally
*/
- virtual void setDestination( const cms::Destination& destination ) {
- this->destination = destination.clone();
+ virtual void setDestination( const cms::Destination* destination ) {
+ this->destination = destination->clone();
}
/**
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=508103&r1=508102&r2=508103
==============================================================================
--- 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 Thu Feb 15 11:42:05 2007
@@ -39,12 +39,12 @@
using namespace activemq::connector::stomp::commands;
////////////////////////////////////////////////////////////////////////////////
-StompSessionManager::StompSessionManager( const std::string& connectionId,
+StompSessionManager::StompSessionManager( const std::string& connectionId,
Transport* transport )
{
if( transport == NULL )
{
- throw NullPointerException(
+ throw NullPointerException(
__FILE__, __LINE__,
"StompSessionManager::StompSessionManager" );
}
@@ -63,7 +63,7 @@
// map because it is really the job of the consumer to remove itself
// when it is destructed. If it doesn't then we would have problems,
// but it's deleted after this object then we would
- // still have problems.
+ // still have problems.
}
////////////////////////////////////////////////////////////////////////////////
@@ -73,7 +73,7 @@
{
return nextSessionId++;
}
-
+
return 0;
}
@@ -90,18 +90,18 @@
////////////////////////////////////////////////////////////////////////////////
connector::SessionInfo* StompSessionManager::createSession(
- cms::Session::AcknowledgeMode ackMode )
+ cms::Session::AcknowledgeMode ackMode )
throw ( exceptions::ActiveMQException )
{
try
{
SessionInfo* session = new StompSessionInfo();
-
+
// Init data
session->setAckMode( ackMode );
session->setConnectionId( connectionId );
session->setSessionId( getNextSessionId() );
-
+
return session;
}
AMQ_CATCH_RETHROW( ActiveMQException )
@@ -109,16 +109,16 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompSessionManager::removeSession(
+void StompSessionManager::removeSession(
connector::SessionInfo* session AMQCPP_UNUSED)
throw ( exceptions::ActiveMQException )
{
// NO-op
}
-
+
////////////////////////////////////////////////////////////////////////////////
connector::ConsumerInfo* StompSessionManager::createConsumer(
- const cms::Destination* destination,
+ const cms::Destination* destination,
SessionInfo* session,
const std::string& selector,
bool noLocal )
@@ -129,8 +129,8 @@
// Delegate to the createDurableConsumer method, just pass the
// appropriate params so that a regular consumer is created on
// the broker side.
- return createDurableConsumer(
- destination, session, "", selector, noLocal );
+ return createDurableConsumer(
+ destination, session, "", selector, noLocal );
}
AMQ_CATCH_RETHROW( StompConnectorException )
AMQ_CATCHALL_THROW( StompConnectorException )
@@ -138,7 +138,7 @@
////////////////////////////////////////////////////////////////////////////////
connector::ConsumerInfo* StompSessionManager::createDurableConsumer(
- const cms::Destination* destination,
+ const cms::Destination* destination,
SessionInfo* session,
const std::string& name,
const std::string& selector,
@@ -149,12 +149,12 @@
{
synchronized( &mutex )
{
- // Find the right mapping to consumers
- ConsumerMap& consumerMap =
+ // Find the right mapping to consumers
+ ConsumerMap& consumerMap =
destinationMap[ destination->toProviderString() ];
- // We only need to send a sub request if there are no active
- // consumers on this destination.
+ // We only need to send a sub request if there are no active
+ // consumers on this destination.
if( consumerMap.empty() )
{
// Send the request to the Broker
@@ -165,8 +165,8 @@
cmd.setAckMode( CommandConstants::ACK_CLIENT );
}
cmd.setDestination( destination->toProviderString() );
-
- if( noLocal == true )
+
+ if( noLocal == true )
{
cmd.setNoLocal( noLocal );
}
@@ -175,46 +175,46 @@
{
cmd.setSubscriptionName( name );
}
-
+
// Grab any options from the destination and set them
// for this subscription.
setSubscribeOptions( destination, cmd );
-
+
// The Selector is set on the first subscribe on this dest,
// and if another consumer is created on this destination
- // that specifies a selector it will be ignored. While
+ // that specifies a selector it will be ignored. While
// this is not ideal, is the only way to handle the fact
// that activemq stomp doesn't support multiple sessions.
if( selector != "" )
{
cmd.setMessageSelector( selector );
}
-
- // Fire the message
+
+ // Fire the message
try{
transport->oneway( &cmd );
} catch( CommandIOException& ex ){
transport->close();
- throw StompConnectorException( __FILE__, __LINE__,
+ throw StompConnectorException( __FILE__, __LINE__,
ex.what() );
}
}
-
+
// Initialize a new Consumer info Message
ConsumerInfo* consumer = new StompConsumerInfo();
-
+
consumer->setConsumerId( getNextConsumerId() );
- consumer->setDestination( *destination );
+ consumer->setDestination( destination );
consumer->setMessageSelector( selector );
consumer->setSessionInfo( session );
-
- // Store this consumer for later message dispatching.
- consumerMap.insert(
+
+ // Store this consumer for later message dispatching.
+ consumerMap.insert(
make_pair( consumer->getConsumerId(), consumer ) );
-
+
return consumer;
}
-
+
return NULL;
}
AMQ_CATCH_RETHROW( StompConnectorException )
@@ -230,37 +230,37 @@
{
synchronized( &mutex )
{
- DestinationMap::iterator itr =
- destinationMap.find( consumer->getDestination().toProviderString() );
-
+ DestinationMap::iterator itr =
+ destinationMap.find( consumer->getDestination()->toProviderString() );
+
if( itr == destinationMap.end() )
{
// Already removed from the map
return;
}
-
+
ConsumerMap& consumers = itr->second;
-
+
// Remove from the map.
consumers.erase( consumer->getConsumerId() );
-
+
// If there are no more on this destination then we unsubscribe
if( consumers.empty() )
{
UnsubscribeCommand cmd;
-
- cmd.setDestination(
- consumer->getDestination().toProviderString() );
-
+
+ cmd.setDestination(
+ consumer->getDestination()->toProviderString() );
+
// Send the message
try{
transport->oneway( &cmd );
} catch( CommandIOException& ex ){
transport->close();
- throw StompConnectorException( __FILE__, __LINE__,
+ throw StompConnectorException( __FILE__, __LINE__,
ex.what() );
}
- }
+ }
}
}
AMQ_CATCH_RETHROW( StompConnectorException )
@@ -268,7 +268,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-void StompSessionManager::onStompCommand( commands::StompCommand* command )
+void StompSessionManager::onStompCommand( commands::StompCommand* command )
throw ( StompConnectorException )
{
try
@@ -289,10 +289,10 @@
"StompSessionManager::onStompCommand - "
"No Message Listener Registered." );
}
-
+
synchronized( &mutex )
{
- DestinationMap::iterator itr =
+ DestinationMap::iterator itr =
destinationMap.find( message->getCMSDestination()->toProviderString() );
if( itr == destinationMap.end() )
@@ -308,16 +308,16 @@
if( itr->second.size() == 1 )
{
ConsumerInfo* consumerInfo = itr->second.begin()->second;
-
- if( StompSelector::isSelected(
+
+ if( StompSelector::isSelected(
consumerInfo->getMessageSelector(),
message ) )
- {
- ActiveMQMessage* msg =
+ {
+ ActiveMQMessage* msg =
dynamic_cast< ActiveMQMessage* >( message );
messageListener->onConsumerMessage( consumerInfo, msg );
}
-
+
return;
}
@@ -325,21 +325,21 @@
// clone the message for each consumer so they don't destroy each other's
// message.
ConsumerMap::iterator c_itr = itr->second.begin();
-
+
for( ; c_itr != itr->second.end(); ++c_itr )
{
ConsumerInfo* consumerInfo = c_itr->second;
-
- if( StompSelector::isSelected(
+
+ if( StompSelector::isSelected(
consumerInfo->getMessageSelector(),
message ) )
{
- ActiveMQMessage* msg =
+ ActiveMQMessage* msg =
dynamic_cast< ActiveMQMessage* >( message->clone() );
messageListener->onConsumerMessage( consumerInfo, msg );
}
}
-
+
// We got here which means that we sent copies, so remove
// the original.
delete command;
@@ -357,7 +357,7 @@
try
{
// Get the properties of this destination
- const Properties& destProperties = dest->getProperties();
+ const Properties& destProperties = dest->getProperties();
if( destProperties.isEmpty() )
{
@@ -365,18 +365,18 @@
return;
}
- std::string noLocalStr =
+ std::string noLocalStr =
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_NOLOCAL );
if( destProperties.getProperty( noLocalStr, "false" ) == "true" )
{
command.setNoLocal(
- Boolean::parseBoolean(
+ Boolean::parseBoolean(
destProperties.getProperty( noLocalStr ) ) );
}
- std::string selectorStr =
+ std::string selectorStr =
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_SELECTOR );
@@ -386,69 +386,69 @@
destProperties.getProperty( selectorStr ) );
}
- std::string priorityStr =
+ std::string priorityStr =
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_PRIORITY );
if( destProperties.hasProperty( priorityStr ) )
{
command.setPriority(
- Integer::parseInt(
+ Integer::parseInt(
destProperties.getProperty( priorityStr ) ) );
}
- std::string dispatchAsyncStr =
+ std::string dispatchAsyncStr =
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_DISPATCHASYNC );
-
+
if( destProperties.hasProperty( dispatchAsyncStr ) )
{
command.setDispatchAsync(
- Boolean::parseBoolean(
+ Boolean::parseBoolean(
destProperties.getProperty( dispatchAsyncStr ) ) );
}
- std::string exclusiveStr =
+ std::string exclusiveStr =
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_EXCLUSIVE );
-
+
if( destProperties.hasProperty( exclusiveStr ) )
{
command.setExclusive(
- Boolean::parseBoolean(
+ Boolean::parseBoolean(
destProperties.getProperty( exclusiveStr ) ) );
}
- std::string maxPendingMsgLimitStr =
+ std::string maxPendingMsgLimitStr =
ActiveMQConstants::toString(
ActiveMQConstants::CUNSUMER_MAXPENDINGMSGLIMIT );
-
+
if( destProperties.hasProperty( maxPendingMsgLimitStr ) )
{
command.setMaxPendingMsgLimit(
- Integer::parseInt(
+ Integer::parseInt(
destProperties.getProperty( maxPendingMsgLimitStr ) ) );
}
-
- std::string prefetchSizeStr =
+
+ std::string prefetchSizeStr =
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_PREFECTCHSIZE );
-
+
if( destProperties.hasProperty( prefetchSizeStr ) )
{
command.setPrefetchSize(
- Integer::parseInt(
+ Integer::parseInt(
destProperties.getProperty( prefetchSizeStr ) ) );
}
- std::string retroactiveStr =
+ std::string retroactiveStr =
ActiveMQConstants::toString(
ActiveMQConstants::CONSUMER_RETROACTIVE );
-
+
if( destProperties.hasProperty( retroactiveStr ) )
{
command.setRetroactive(
- Boolean::parseBoolean(
+ Boolean::parseBoolean(
destProperties.getProperty( retroactiveStr ) ) );
}
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQProducer.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQProducer.cpp?view=diff&rev=508103&r1=508102&r2=508103
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQProducer.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQProducer.cpp Thu Feb 15 11:42:05 2007
@@ -38,7 +38,7 @@
__FILE__, __LINE__,
"ActiveMQProducer::ActiveMQProducer - Init with NULL Session" );
}
-
+
// Init Producer Data
this->session = session;
this->producerInfo = producerInfo;
@@ -64,7 +64,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQProducer::close()
+void ActiveMQProducer::close()
throw ( cms::CMSException )
{
try
@@ -72,7 +72,7 @@
if( !closed ) {
// Dispose of the ProducerInfo
session->onDestroySessionResource( this );
-
+
closed = true;
}
}
@@ -81,7 +81,7 @@
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQProducer::send( cms::Message* message )
+void ActiveMQProducer::send( cms::Message* message )
throw ( cms::CMSException )
{
try
@@ -93,16 +93,16 @@
"ActiveMQProducer::send - This Producer is closed" );
}
- send( &producerInfo->getDestination(), message );
+ send( producerInfo->getDestination(), message );
}
AMQ_CATCH_RETHROW( ActiveMQException )
AMQ_CATCHALL_THROW( ActiveMQException )
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQProducer::send( cms::Message* message, int deliveryMode,
- int priority,
- long long timeToLive )
+void ActiveMQProducer::send( cms::Message* message, int deliveryMode,
+ int priority,
+ long long timeToLive )
throw ( cms::CMSException )
{
try
@@ -114,7 +114,7 @@
"ActiveMQProducer::send - This Producer is closed" );
}
- send( &producerInfo->getDestination(), message, deliveryMode,
+ send( producerInfo->getDestination(), message, deliveryMode,
priority, timeToLive );
}
AMQ_CATCH_RETHROW( ActiveMQException )
@@ -143,8 +143,8 @@
////////////////////////////////////////////////////////////////////////////////
void ActiveMQProducer::send( const cms::Destination* destination,
- cms::Message* message, int deliveryMode,
- int priority, long long timeToLive)
+ cms::Message* message, int deliveryMode,
+ int priority, long long timeToLive)
throw ( cms::CMSException )
{
try
@@ -160,7 +160,7 @@
message->setCMSDestination( destination );
message->setCMSDeliveryMode( deliveryMode );
message->setCMSPriority( priority );
-
+
long long expiration = 0LL;
if( !disableTimestamps ) {
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/connector/stomp/StompConnectorTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/connector/stomp/StompConnectorTest.h?view=diff&rev=508103&r1=508102&r2=508103
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/connector/stomp/StompConnectorTest.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/connector/stomp/StompConnectorTest.h Thu Feb 15 11:42:05 2007
@@ -51,10 +51,10 @@
CPPUNIT_TEST_SUITE_END();
public:
-
+
StompConnectorTest() {}
virtual ~StompConnectorTest() {}
-
+
class MyExceptionListener : public cms::ExceptionListener{
public:
int num;
@@ -63,33 +63,33 @@
virtual void onException( const cms::CMSException& ex AMQCPP_UNUSED){
num++;
}
- };
-
+ };
+
class MyCommandListener : public transport::CommandListener{
public:
-
+
transport::Command* cmd;
-
+
public:
-
+
MyCommandListener(){
cmd = NULL;
}
virtual ~MyCommandListener(){}
-
+
virtual void onCommand( transport::Command* command ){
cmd = command;
}
};
-
+
class MyMessageListener : public ConsumerMessageListener{
public:
-
+
std::vector<ConsumerInfo*> consumers;
-
+
public:
virtual ~MyMessageListener(){}
-
+
virtual void onConsumerMessage( ConsumerInfo* consumer,
core::ActiveMQMessage* msg )
{
@@ -105,29 +105,29 @@
StompResponseBuilder responseBuilder("testConnectionId");
transport::DummyTransport transport( &responseBuilder );
util::SimpleProperties properties;
-
+
// Using a pointer for the connector so we ensure the proper destruction
// order of objects - connector before the transport.
StompConnector* connector = new StompConnector( &transport, properties );
-
+
connector->start();
-
+
SessionInfo* info1 = connector->createSession( cms::Session::AUTO_ACKNOWLEDGE );
CPPUNIT_ASSERT( info1->getAckMode() == cms::Session::AUTO_ACKNOWLEDGE );
CPPUNIT_ASSERT( info1->getConnectionId() == connectionId );
-
+
SessionInfo* info2 = connector->createSession( cms::Session::DUPS_OK_ACKNOWLEDGE );
CPPUNIT_ASSERT( info2->getAckMode() == cms::Session::DUPS_OK_ACKNOWLEDGE );
CPPUNIT_ASSERT( info2->getConnectionId() == connectionId );
-
+
SessionInfo* info3 = connector->createSession( cms::Session::CLIENT_ACKNOWLEDGE );
CPPUNIT_ASSERT( info3->getAckMode() == cms::Session::CLIENT_ACKNOWLEDGE );
CPPUNIT_ASSERT( info3->getConnectionId() == connectionId );
-
+
SessionInfo* info4 = connector->createSession( cms::Session::SESSION_TRANSACTED );
CPPUNIT_ASSERT( info4->getAckMode() == cms::Session::SESSION_TRANSACTED );
CPPUNIT_ASSERT( info4->getConnectionId() == connectionId );
-
+
connector->destroyResource( info1 );
connector->destroyResource( info2 );
connector->destroyResource( info3 );
@@ -137,52 +137,52 @@
// of destruction.
delete connector;
}
-
+
void testConsumers()
{
std::string connectionId = "testConnectionId";
StompResponseBuilder responseBuilder("testConnectionId");
transport::DummyTransport transport( &responseBuilder );
util::SimpleProperties properties;
-
+
// Using a pointer for the connector so we ensure the proper destruction
// order of objects - connector before the transport.
StompConnector* connector = new StompConnector( &transport, properties );
-
+
connector->start();
-
+
SessionInfo* info1 = connector->createSession( cms::Session::AUTO_ACKNOWLEDGE );
std::string sel1 = "";
StompTopic dest1( "dummy.topic.1" );
ConsumerInfo* cinfo1 = connector->createConsumer( &dest1, info1, sel1 );
CPPUNIT_ASSERT( cinfo1->getSessionInfo() == info1 );
- CPPUNIT_ASSERT( cinfo1->getDestination().toProviderString() == dest1.toProviderString() );
+ CPPUNIT_ASSERT( cinfo1->getDestination()->toProviderString() == dest1.toProviderString() );
CPPUNIT_ASSERT( cinfo1->getMessageSelector() == sel1 );
-
+
SessionInfo* info2 = connector->createSession( cms::Session::DUPS_OK_ACKNOWLEDGE );
std::string sel2 = "mysel2";
StompTopic dest2( "dummy.topic.2" );
ConsumerInfo* cinfo2 = connector->createConsumer( &dest2, info2, sel2 );
CPPUNIT_ASSERT( cinfo2->getSessionInfo() == info2 );
- CPPUNIT_ASSERT( cinfo2->getDestination().toProviderString() == dest2.toProviderString() );
+ CPPUNIT_ASSERT( cinfo2->getDestination()->toProviderString() == dest2.toProviderString() );
CPPUNIT_ASSERT( cinfo2->getMessageSelector() == sel2 );
-
+
SessionInfo* info3 = connector->createSession( cms::Session::CLIENT_ACKNOWLEDGE );
std::string sel3 = "mysel3";
StompQueue dest3( "dummy.queue.1" );
ConsumerInfo* cinfo3 = connector->createConsumer( &dest3, info3, sel3 );
CPPUNIT_ASSERT( cinfo3->getSessionInfo() == info3 );
- CPPUNIT_ASSERT( cinfo3->getDestination().toProviderString() == dest3.toProviderString() );
+ CPPUNIT_ASSERT( cinfo3->getDestination()->toProviderString() == dest3.toProviderString() );
CPPUNIT_ASSERT( cinfo3->getMessageSelector() == sel3 );
-
+
SessionInfo* info4 = connector->createSession( cms::Session::SESSION_TRANSACTED );
std::string sel4 = "";
StompTopic dest4( "dummy.queue.2" );
ConsumerInfo* cinfo4 = connector->createConsumer( &dest4, info4, sel4 );
CPPUNIT_ASSERT( cinfo4->getSessionInfo() == info4 );
- CPPUNIT_ASSERT( cinfo4->getDestination().toProviderString() == dest4.toProviderString() );
+ CPPUNIT_ASSERT( cinfo4->getDestination()->toProviderString() == dest4.toProviderString() );
CPPUNIT_ASSERT( cinfo4->getMessageSelector() == sel4 );
-
+
connector->destroyResource( cinfo1 );
connector->destroyResource( cinfo2 );
connector->destroyResource( cinfo3 );
@@ -192,7 +192,7 @@
connector->destroyResource( info2 );
connector->destroyResource( info3 );
connector->destroyResource( info4 );
-
+
// Delete the connector here - this assures the propery order
// of destruction.
delete connector;
@@ -204,37 +204,37 @@
StompResponseBuilder responseBuilder("testConnectionId");
transport::DummyTransport transport( &responseBuilder );
util::SimpleProperties properties;
-
+
// Using a pointer for the connector so we ensure the proper destruction
// order of objects - connector before the transport.
StompConnector* connector = new StompConnector( &transport, properties );
-
+
connector->start();
-
+
SessionInfo* info1 = connector->createSession( cms::Session::AUTO_ACKNOWLEDGE );
StompTopic dest1( "dummy.topic.1" );
ProducerInfo* pinfo1 = connector->createProducer( &dest1, info1 );
CPPUNIT_ASSERT( pinfo1->getSessionInfo() == info1 );
- CPPUNIT_ASSERT( pinfo1->getDestination().toProviderString() == dest1.toProviderString() );
-
+ CPPUNIT_ASSERT( pinfo1->getDestination()->toProviderString() == dest1.toProviderString() );
+
SessionInfo* info2 = connector->createSession( cms::Session::DUPS_OK_ACKNOWLEDGE );
StompTopic dest2( "dummy.topic.2" );
ProducerInfo* pinfo2 = connector->createProducer( &dest2, info2 );
CPPUNIT_ASSERT( pinfo2->getSessionInfo() == info2 );
- CPPUNIT_ASSERT( pinfo2->getDestination().toProviderString() == dest2.toProviderString() );
-
+ CPPUNIT_ASSERT( pinfo2->getDestination()->toProviderString() == dest2.toProviderString() );
+
SessionInfo* info3 = connector->createSession( cms::Session::CLIENT_ACKNOWLEDGE );
StompQueue dest3( "dummy.queue.1" );
ProducerInfo* pinfo3 = connector->createProducer( &dest3, info3 );
CPPUNIT_ASSERT( pinfo3->getSessionInfo() == info3 );
- CPPUNIT_ASSERT( pinfo3->getDestination().toProviderString() == dest3.toProviderString() );
-
+ CPPUNIT_ASSERT( pinfo3->getDestination()->toProviderString() == dest3.toProviderString() );
+
SessionInfo* info4 = connector->createSession( cms::Session::SESSION_TRANSACTED );
StompTopic dest4( "dummy.queue.2" );
ProducerInfo* pinfo4 = connector->createProducer( &dest4, info4 );
CPPUNIT_ASSERT( pinfo4->getSessionInfo() == info4 );
- CPPUNIT_ASSERT( pinfo4->getDestination().toProviderString() == dest4.toProviderString() );
-
+ CPPUNIT_ASSERT( pinfo4->getDestination()->toProviderString() == dest4.toProviderString() );
+
connector->destroyResource( pinfo1 );
connector->destroyResource( pinfo2 );
connector->destroyResource( pinfo3 );
@@ -244,79 +244,79 @@
connector->destroyResource( info2 );
connector->destroyResource( info3 );
connector->destroyResource( info4 );
-
+
// Delete the connector here - this assures the propery order
// of destruction.
delete connector;
}
-
+
void testCommand()
{
std::string connectionId = "testConnectionId";
StompResponseBuilder responseBuilder("testConnectionId");
transport::DummyTransport transport( &responseBuilder );
util::SimpleProperties properties;
-
+
// Using a pointer for the connector so we ensure the proper destruction
// order of objects - connector before the transport.
StompConnector* connector = new StompConnector( &transport, properties );
-
+
connector->start();
-
+
StompTopic dest1( "dummy.topic" );
StompTopic dest2( "dummy.topic2" );
-
+
SessionInfo* info1 = connector->createSession( cms::Session::AUTO_ACKNOWLEDGE );
ConsumerInfo* cinfo1 = connector->createConsumer( &dest1, info1, "" );
-
+
SessionInfo* info2 = connector->createSession( cms::Session::DUPS_OK_ACKNOWLEDGE );
ConsumerInfo* cinfo2 = connector->createConsumer( &dest1, info2, "" );
-
+
SessionInfo* info3 = connector->createSession( cms::Session::CLIENT_ACKNOWLEDGE );
ConsumerInfo* cinfo3 = connector->createConsumer( &dest2, info3, "" );
-
+
SessionInfo* info4 = connector->createSession( cms::Session::SESSION_TRANSACTED );
ConsumerInfo* cinfo4 = connector->createConsumer( &dest2, info4, "" );
-
+
MyMessageListener listener;
connector->setConsumerMessageListener( &listener );
-
+
StompFrame* frame = new StompFrame();
frame->setCommand( "MESSAGE" );
- frame->getProperties().setProperty(
+ frame->getProperties().setProperty(
"destination", dest1.toProviderString() );
const char* buffer = "hello world";
frame->setBody( (const unsigned char*)buffer, 12 );
- commands::TextMessageCommand* msg =
+ commands::TextMessageCommand* msg =
new commands::TextMessageCommand( frame ); // deleted by listener
transport.fireCommand( msg );
-
+
CPPUNIT_ASSERT( listener.consumers.size() == 2 );
for( unsigned int ix=0; ix<listener.consumers.size(); ++ix ){
- CPPUNIT_ASSERT( listener.consumers[ix] == cinfo1 ||
+ CPPUNIT_ASSERT( listener.consumers[ix] == cinfo1 ||
listener.consumers[ix] == cinfo2 );
}
-
+
// Clean up the consumers list
listener.consumers.clear();
-
+
frame = new StompFrame();
frame->setCommand( "MESSAGE" );
- frame->getProperties().setProperty(
+ frame->getProperties().setProperty(
"destination", dest2.toProviderString() );
buffer = "hello world";
frame->setBody( (unsigned char*)buffer, 12 );
msg = new commands::TextMessageCommand( frame ); // deleted by listener
transport.fireCommand( msg );
-
+
CPPUNIT_ASSERT( listener.consumers.size() == 2 );
for( unsigned int ix=0; ix<listener.consumers.size(); ++ix ){
- CPPUNIT_ASSERT( listener.consumers[ix] == cinfo3 ||
+ CPPUNIT_ASSERT( listener.consumers[ix] == cinfo3 ||
listener.consumers[ix] == cinfo4 );
}
-
+
connector->destroyResource( cinfo1 );
connector->destroyResource( cinfo2 );
connector->destroyResource( cinfo3 );
@@ -326,50 +326,50 @@
connector->destroyResource( info2 );
connector->destroyResource( info3 );
connector->destroyResource( info4 );
-
+
// Delete the connector here - this assures the propery order
// of destruction.
delete connector;
}
-
+
void testSendingCommands()
{
std::string connectionId = "testConnectionId";
StompResponseBuilder responseBuilder("testConnectionId");
transport::DummyTransport transport( &responseBuilder );
util::SimpleProperties properties;
- StompConnector* connector =
+ StompConnector* connector =
new StompConnector( &transport, properties );
connector->start();
-
+
StompTopic dest1( "dummy.topic.1" );
-
+
MyCommandListener cmdListener;
transport.setOutgoingCommandListener( &cmdListener );
-
+
SessionInfo* info1 = connector->createSession( cms::Session::AUTO_ACKNOWLEDGE );
- ConsumerInfo* cinfo1 = connector->createConsumer( &dest1, info1, "" );
+ ConsumerInfo* cinfo1 = connector->createConsumer( &dest1, info1, "" );
CPPUNIT_ASSERT( cmdListener.cmd != NULL );
-
+
cmdListener.cmd = NULL;
-
+
SessionInfo* info2 = connector->createSession( cms::Session::DUPS_OK_ACKNOWLEDGE );
ConsumerInfo* cinfo2 = connector->createConsumer( &dest1, info2, "" );
CPPUNIT_ASSERT( cmdListener.cmd == NULL );
-
+
cmdListener.cmd = NULL;
-
+
connector->destroyResource( cinfo1 );
CPPUNIT_ASSERT( cmdListener.cmd == NULL );
-
+
cmdListener.cmd = NULL;
-
+
connector->destroyResource( cinfo2 );
CPPUNIT_ASSERT( cmdListener.cmd != NULL );
-
+
connector->destroyResource( info1 );
connector->destroyResource( info2 );
-
+
delete connector;
CPPUNIT_ASSERT( cmdListener.cmd != NULL );
}
@@ -381,25 +381,25 @@
transport::DummyTransport transport( &responseBuilder );
MyExceptionListener exListener;
util::SimpleProperties properties;
-
+
// Using a pointer for the connector so we ensure the proper destruction
// order of objects - connector before the transport.
StompConnector* connector = new StompConnector( &transport, properties );
-
+
connector->setExceptionListener(&exListener);
-
+
connector->start();
-
+
// Initiate an exception from the transport.
transport.fireException( exceptions::ActiveMQException(__FILE__, __LINE__, "test") );
-
+
CPPUNIT_ASSERT( exListener.num == 1 );
// Delete the connector here - this assures the propery order
// of destruction.
delete connector;
}
- };
+ };
}}}