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 2012/05/04 23:04:54 UTC
svn commit: r1334183 [1/4] - in
/activemq/activemq-cpp/trunk/activemq-cpp/src: main/
main/activemq/commands/ main/activemq/core/ main/activemq/core/kernels/
main/activemq/state/ main/activemq/util/ main/cms/ main/decaf/lang/
main/decaf/util/ main/decaf...
Author: tabish
Date: Fri May 4 21:04:52 2012
New Revision: 1334183
URL: http://svn.apache.org/viewvc?rev=1334183&view=rev
Log:
fix for:
https://issues.apache.org/jira/browse/AMQCPP-399
https://issues.apache.org/jira/browse/AMQCPP-395
https://issues.apache.org/jira/browse/AMQCPP-397
https://issues.apache.org/jira/browse/AMQCPP-398
Added:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/AdvisoryConsumer.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/AdvisoryConsumer.h (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/util/ActiveMQMessageTransformation.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/util/ActiveMQMessageTransformation.h (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/util/AdvisorySupport.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/util/AdvisorySupport.h (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/util/ActiveMQMessageTransformationTest.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/util/ActiveMQMessageTransformationTest.h (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/util/AdvisorySupportTest.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/util/AdvisorySupportTest.h (with props)
Modified:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ConsumerId.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnectionFactory.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnectionFactory.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/DispatchData.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/Dispatcher.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/kernels/ActiveMQConsumerKernel.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/kernels/ActiveMQConsumerKernel.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/kernels/ActiveMQProducerKernel.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/kernels/ActiveMQProducerKernel.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/kernels/ActiveMQSessionKernel.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/kernels/ActiveMQSessionKernel.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/state/ConnectionState.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/cms/Message.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/lang/Comparable.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/lang/Pointer.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/Comparator.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/ConcurrentStlMap.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test/Makefile.am
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/commands/ActiveMQTempQueueTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/commands/ActiveMQTempTopicTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/core/ActiveMQConnectionFactoryTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/state/ConnectionStateTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/testRegistry.cpp
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?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am Fri May 4 21:04:52 2012
@@ -97,6 +97,7 @@ cc_sources = \
activemq/core/ActiveMQXAConnection.cpp \
activemq/core/ActiveMQXAConnectionFactory.cpp \
activemq/core/ActiveMQXASession.cpp \
+ activemq/core/AdvisoryConsumer.cpp \
activemq/core/FifoMessageDispatchChannel.cpp \
activemq/core/PrefetchPolicy.cpp \
activemq/core/RedeliveryPolicy.cpp \
@@ -145,7 +146,9 @@ cc_sources = \
activemq/transport/tcp/SslTransportFactory.cpp \
activemq/transport/tcp/TcpTransport.cpp \
activemq/transport/tcp/TcpTransportFactory.cpp \
+ activemq/util/ActiveMQMessageTransformation.cpp \
activemq/util/ActiveMQProperties.cpp \
+ activemq/util/AdvisorySupport.cpp \
activemq/util/CMSExceptionSupport.cpp \
activemq/util/CompositeData.cpp \
activemq/util/IdGenerator.cpp \
@@ -592,6 +595,7 @@ h_sources = \
activemq/core/ActiveMQXAConnection.h \
activemq/core/ActiveMQXAConnectionFactory.h \
activemq/core/ActiveMQXASession.h \
+ activemq/core/AdvisoryConsumer.h \
activemq/core/DispatchData.h \
activemq/core/Dispatcher.h \
activemq/core/FifoMessageDispatchChannel.h \
@@ -659,7 +663,9 @@ h_sources = \
activemq/transport/tcp/SslTransportFactory.h \
activemq/transport/tcp/TcpTransport.h \
activemq/transport/tcp/TcpTransportFactory.h \
+ activemq/util/ActiveMQMessageTransformation.h \
activemq/util/ActiveMQProperties.h \
+ activemq/util/AdvisorySupport.h \
activemq/util/CMSExceptionSupport.h \
activemq/util/CompositeData.h \
activemq/util/Config.h \
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.cpp?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.cpp Fri May 4 21:04:52 2012
@@ -24,10 +24,15 @@
#include <activemq/commands/ActiveMQTempTopic.h>
#include <activemq/commands/ActiveMQTempQueue.h>
-using namespace std;
+#include <decaf/util/StringTokenizer.h>
+#include <decaf/util/StlSet.h>
+
+#include <apr_strings.h>
+
using namespace activemq;
using namespace activemq::util;
using namespace activemq::commands;
+using namespace decaf::util;
using namespace decaf::lang;
using namespace decaf::lang::exceptions;
@@ -51,16 +56,35 @@ const std::string ActiveMQDestination::T
const std::string ActiveMQDestination::TEMP_TOPIC_QUALIFED_PREFIX = "temp-topic://";
////////////////////////////////////////////////////////////////////////////////
+namespace {
+
+ const std::string trim(const std::string& input, const std::string& ws = " \t")
+ {
+ const size_t beginStr = input.find_first_not_of(ws);
+ if (beginStr == std::string::npos)
+ {
+ // no content
+ return "";
+ }
+
+ const size_t endStr = input.find_last_not_of(ws);
+ const size_t range = endStr - beginStr + 1;
+
+ return input.substr(beginStr, range);
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
ActiveMQDestination::ActiveMQDestination() :
BaseDataStructure(), exclusive(false), ordered(false), advisory(false), orderedTarget(DEFAULT_ORDERED_TARGET), physicalName(), options() {
}
////////////////////////////////////////////////////////////////////////////////
-ActiveMQDestination::ActiveMQDestination( const std::string& physicalName ) :
+ActiveMQDestination::ActiveMQDestination(const std::string& physicalName) :
BaseDataStructure(), exclusive(false), ordered(false), advisory(false), orderedTarget(DEFAULT_ORDERED_TARGET), physicalName(), options() {
- this->setPhysicalName( physicalName );
+ this->setPhysicalName(physicalName);
}
////////////////////////////////////////////////////////////////////////////////
@@ -68,71 +92,67 @@ ActiveMQDestination::~ActiveMQDestinatio
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQDestination::setPhysicalName( const std::string& physicalName ) {
+void ActiveMQDestination::setPhysicalName(const std::string& physicalName) {
this->physicalName = physicalName;
size_t pos = physicalName.find_first_of('?');
- if( pos != string::npos ) {
-
- std::string optstring = physicalName.substr( pos + 1 );
- this->physicalName = physicalName.substr( 0, pos );
- URISupport::parseQuery( optstring, &options.getProperties() );
+ if (pos != std::string::npos) {
+ std::string optstring = physicalName.substr(pos + 1);
+ this->physicalName = physicalName.substr(0, pos);
+ URISupport::parseQuery(optstring, &options.getProperties());
}
- this->advisory = physicalName.find_first_of( ADVISORY_PREFIX ) == 0;
+ this->advisory = physicalName.find_first_of(ADVISORY_PREFIX) == 0;
+ this->compositeDestinations.clear();
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQDestination::copyDataStructure( const DataStructure* src ) {
+void ActiveMQDestination::copyDataStructure(const DataStructure* src) {
// Copy the data of the base class or classes
BaseDataStructure::copyDataStructure( src );
- const ActiveMQDestination* srcPtr =
- dynamic_cast<const ActiveMQDestination*>( src );
-
- if( srcPtr == NULL || src == NULL ) {
+ const ActiveMQDestination* srcPtr = dynamic_cast<const ActiveMQDestination*> (src);
+ if (srcPtr == NULL || src == NULL) {
throw decaf::lang::exceptions::NullPointerException(
- __FILE__, __LINE__,
- "BrokerId::copyDataStructure - src is NULL or invalid" );
+ __FILE__, __LINE__, "BrokerId::copyDataStructure - src is NULL or invalid");
}
- this->setPhysicalName( srcPtr->getPhysicalName() );
- this->setAdvisory( srcPtr->isAdvisory() );
- this->setOrdered( srcPtr->isOrdered() );
- this->setExclusive( srcPtr->isExclusive() );
- this->setOrderedTarget( srcPtr->getOrderedTarget() );
- this->options.copy( &srcPtr->getOptions() );
+ this->setPhysicalName(srcPtr->getPhysicalName());
+ this->setAdvisory(srcPtr->isAdvisory());
+ this->setOrdered(srcPtr->isOrdered());
+ this->setExclusive(srcPtr->isExclusive());
+ this->setOrderedTarget(srcPtr->getOrderedTarget());
+ this->options.copy(&srcPtr->getOptions());
}
////////////////////////////////////////////////////////////////////////////////
std::string ActiveMQDestination::toString() const {
- switch( this->getDestinationType() ) {
-
+ switch (this->getDestinationType()) {
case cms::Destination::TOPIC:
- return std::string( "topic://" ) + this->getPhysicalName();
+ return std::string("topic://") + this->getPhysicalName();
case cms::Destination::TEMPORARY_TOPIC:
- return std::string( "temp-topic://" ) + this->getPhysicalName();
+ return std::string("temp-topic://") + this->getPhysicalName();
case cms::Destination::TEMPORARY_QUEUE:
- return std::string( "temp-queue://" ) + this->getPhysicalName();
+ return std::string("temp-queue://") + this->getPhysicalName();
default:
- return std::string( "queue://" ) + this->getPhysicalName();
+ return std::string("queue://") + this->getPhysicalName();
}
}
////////////////////////////////////////////////////////////////////////////////
-bool ActiveMQDestination::equals( const DataStructure* value ) const {
+bool ActiveMQDestination::equals(const DataStructure* value) const {
- if( this == value ) {
+ if (this == value) {
return true;
}
- const ActiveMQDestination* valuePtr = dynamic_cast<const ActiveMQDestination*>( value );
+ const ActiveMQDestination* valuePtr = dynamic_cast<const ActiveMQDestination*> (value);
- if( valuePtr == NULL || value == NULL ) {
+ if (valuePtr == NULL || value == NULL) {
return false;
}
@@ -145,18 +165,17 @@ unsigned char ActiveMQDestination::getDa
}
////////////////////////////////////////////////////////////////////////////////
-std::string ActiveMQDestination::getClientId(
- const ActiveMQDestination* destination ) {
+std::string ActiveMQDestination::getClientId(const ActiveMQDestination* destination) {
std::string answer = "";
- if( destination != NULL && destination->isTemporary() ) {
+ if (destination != NULL && destination->isTemporary()) {
std::string name = destination->getPhysicalName();
- size_t start = name.find( TEMP_PREFIX );
- if( start != std::string::npos ) {
+ size_t start = name.find(TEMP_PREFIX);
+ if (start != std::string::npos) {
start += TEMP_PREFIX.length();
- size_t stop = name.rfind( TEMP_POSTFIX );
- if( stop > start && stop < name.length() ) {
- answer = name.substr( start, stop-start );
+ size_t stop = name.rfind(TEMP_POSTFIX);
+ if (stop > start && stop < name.length()) {
+ answer = name.substr(start, stop - start);
}
}
}
@@ -164,40 +183,114 @@ std::string ActiveMQDestination::getClie
}
////////////////////////////////////////////////////////////////////////////////
-Pointer<ActiveMQDestination> ActiveMQDestination::createDestination( int type, const std::string& name ) {
+Pointer<ActiveMQDestination> ActiveMQDestination::createDestination(int type, const std::string& name) {
Pointer<ActiveMQDestination> result;
- if( name.find( QUEUE_QUALIFIED_PREFIX ) == 0 ) {
- result.reset( new ActiveMQQueue( name.substr( QUEUE_QUALIFIED_PREFIX.length() ) ) );
+ if (name.find(QUEUE_QUALIFIED_PREFIX) == 0) {
+ result.reset(new ActiveMQQueue(name.substr(QUEUE_QUALIFIED_PREFIX.length())));
return result;
- } else if( name.find( TOPIC_QUALIFIED_PREFIX ) == 0 ) {
- result.reset( new ActiveMQTopic( name.substr( TOPIC_QUALIFIED_PREFIX.length() ) ) );
+ } else if (name.find(TOPIC_QUALIFIED_PREFIX) == 0) {
+ result.reset(new ActiveMQTopic(name.substr(TOPIC_QUALIFIED_PREFIX.length())));
return result;
- } else if( name.find( TEMP_QUEUE_QUALIFED_PREFIX ) == 0 ) {
- result.reset( new ActiveMQTempQueue( name.substr( TEMP_QUEUE_QUALIFED_PREFIX.length() ) ) );
+ } else if (name.find(TEMP_QUEUE_QUALIFED_PREFIX) == 0) {
+ result.reset(new ActiveMQTempQueue(name.substr(TEMP_QUEUE_QUALIFED_PREFIX.length())));
return result;
- } else if( name.find( TEMP_TOPIC_QUALIFED_PREFIX ) == 0 ) {
- result.reset( new ActiveMQTempTopic( name.substr( TEMP_TOPIC_QUALIFED_PREFIX.length() ) ) );
+ } else if (name.find(TEMP_TOPIC_QUALIFED_PREFIX) == 0) {
+ result.reset(new ActiveMQTempTopic(name.substr(TEMP_TOPIC_QUALIFED_PREFIX.length())));
return result;
}
- switch( type ) {
+ switch (type) {
case cms::Destination::QUEUE:
- result.reset( new ActiveMQQueue( name ) );
+ result.reset(new ActiveMQQueue(name));
return result;
case cms::Destination::TOPIC:
- result.reset( new ActiveMQTopic( name ) );
+ result.reset(new ActiveMQTopic(name));
return result;
case cms::Destination::TEMPORARY_QUEUE:
- result.reset( new ActiveMQTempQueue( name ) );
+ result.reset(new ActiveMQTempQueue(name));
return result;
case cms::Destination::TEMPORARY_TOPIC:
- result.reset( new ActiveMQTempTopic( name ) );
+ result.reset(new ActiveMQTempTopic(name));
return result;
default:
throw IllegalArgumentException(
- __FILE__, __LINE__,
- "Invalid default destination type: %d", type );
+ __FILE__, __LINE__, "Invalid default destination type: %d", type);
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+std::string ActiveMQDestination::getDestinationTypeAsString() const {
+ switch (getDestinationType()) {
+ case cms::Destination::QUEUE:
+ return "Queue";
+ case cms::Destination::TOPIC:
+ return "Topic";
+ case cms::Destination::TEMPORARY_QUEUE:
+ return "TempQueue";
+ case cms::Destination::TEMPORARY_TOPIC:
+ return "TempTopic";
+ default:
+ throw new IllegalArgumentException(
+ __FILE__, __LINE__, "Invalid destination type: " + getDestinationType());
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+decaf::util::ArrayList< Pointer<ActiveMQDestination> > ActiveMQDestination::getCompositeDestinations() const {
+
+ if (!this->compositeDestinations.isEmpty()) {
+ return this->compositeDestinations;
+ }
+
+ if (this->isComposite()) {
+
+ StlSet<std::string> components;
+ StringTokenizer iter(this->physicalName, ActiveMQDestination::COMPOSITE_SEPARATOR);
+ while (iter.hasMoreTokens()) {
+ std::string name = trim(iter.nextToken());
+ if (name.length() == 0) {
+ continue;
+ }
+ components.add(name);
+ }
+
+ Pointer< Iterator<std::string> > iterator(components.iterator());
+ while (iterator->hasNext()) {
+ compositeDestinations.add(createDestination(iterator->next()));
+ }
+ }
+
+ return this->compositeDestinations;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+int ActiveMQDestination::compareTo(const ActiveMQDestination& value) const {
+
+ if (this == &value) {
+ return 0;
+ }
+
+ int valueComp = apr_strnatcmp(this->getPhysicalName().c_str(), value.getPhysicalName().c_str());
+ if (valueComp != 0) {
+ return valueComp;
}
+
+ return 0;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+bool ActiveMQDestination::equals(const ActiveMQDestination& value) const {
+ return this->getPhysicalName() == value.getPhysicalName();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+bool ActiveMQDestination::operator==(const ActiveMQDestination& value) const {
+ return this->getPhysicalName() == value.getPhysicalName();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+bool ActiveMQDestination::operator<(const ActiveMQDestination& value) const {
+ return this->compareTo(value) < 0;
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.h?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQDestination.h Fri May 4 21:04:52 2012
@@ -28,6 +28,8 @@
#include <activemq/util/ActiveMQProperties.h>
#include <cms/Destination.h>
#include <decaf/lang/Pointer.h>
+#include <decaf/lang/Comparable.h>
+#include <decaf/util/ArrayList.h>
#include <vector>
#include <string>
@@ -38,7 +40,8 @@ namespace commands{
using decaf::lang::Pointer;
- class AMQCPP_API ActiveMQDestination : public BaseDataStructure {
+ class AMQCPP_API ActiveMQDestination : public BaseDataStructure,
+ public decaf::lang::Comparable<ActiveMQDestination> {
protected:
struct DestinationFilter{
@@ -84,6 +87,7 @@ namespace commands{
bool exclusive;
bool ordered;
bool advisory;
+ mutable decaf::util::ArrayList< Pointer<ActiveMQDestination> > compositeDestinations;
std::string orderedTarget;
std::string physicalName;
@@ -93,16 +97,18 @@ namespace commands{
const static unsigned char ID_ACTIVEMQDESTINATION = 0;
+ typedef decaf::lang::PointerComparator<ActiveMQDestination> COMPARATOR;
+
private:
- ActiveMQDestination( const ActiveMQDestination& );
- ActiveMQDestination& operator= ( const ActiveMQDestination& );
+ ActiveMQDestination(const ActiveMQDestination&);
+ ActiveMQDestination& operator=(const ActiveMQDestination&);
public:
ActiveMQDestination();
- ActiveMQDestination( const std::string& physicalName );
+ ActiveMQDestination(const std::string& physicalName);
virtual ~ActiveMQDestination() throw();
@@ -110,22 +116,21 @@ namespace commands{
return NULL;
}
- virtual void copyDataStructure( const DataStructure* src );
+ virtual void copyDataStructure(const DataStructure* src);
- virtual bool equals( const DataStructure* value ) const;
+ virtual bool equals(const DataStructure* value) const;
virtual unsigned char getDataStructureType() const;
virtual std::string toString() const;
+ public:
+
/**
* Fetch this destination's physical name
* @returns const string containing the name
*/
- virtual const std::string& getPhysicalName() const {
- return this->physicalName;
- }
- virtual std::string& getPhysicalName() {
+ virtual std::string getPhysicalName() const {
return this->physicalName;
}
@@ -133,7 +138,7 @@ namespace commands{
* Set this destination's physical name
* @returns const string containing the name
*/
- virtual void setPhysicalName( const std::string& physicalName );
+ virtual void setPhysicalName(const std::string& physicalName);
/**
* @return Returns the advisory.
@@ -145,7 +150,7 @@ namespace commands{
/**
* @param advisory The advisory to set.
*/
- virtual void setAdvisory( bool advisory ){
+ virtual void setAdvisory(bool advisory) {
this->advisory = advisory;
}
@@ -197,7 +202,7 @@ namespace commands{
/**
* @param ordered The ordered to set.
*/
- virtual void setOrdered( bool ordered ) {
+ virtual void setOrdered(bool ordered) {
this->ordered = ordered;
}
@@ -211,49 +216,23 @@ namespace commands{
/**
* @param orderedTarget The orderedTarget to set.
*/
- virtual void setOrderedTarget( const std::string& orderedTarget) {
+ virtual void setOrderedTarget(const std::string& orderedTarget) {
this->orderedTarget = orderedTarget;
}
- public: // Statics
-
- /**
- * Create a temporary name from the clientId
- *
- * @param clientId
- * @return
- */
- 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
- *
- * @param destination
- * @return the clientId or null if not a temporary destination
+ * Returns the Type of Destination that this object represents
+ * @returns int type qualifier.
*/
- static std::string getClientId( const ActiveMQDestination* destination );
+ virtual cms::Destination::DestinationType getDestinationType() const = 0;
/**
- * Creates a Destination given the String Name to use and a Type.
+ * Returns the type of Destination that this object represents as a string, the
+ * available string values are, "Queue", "Topic", "TempQueue" and "TempTopic".
*
- * @param type - The Type of Destination to Create
- * @param name - The Name to use in the creation of the Destination
- *
- * @return Pointer to a new ActiveMQDestination instance.
+ * @returns The string value that represents the type of this destination.
*/
- static Pointer<ActiveMQDestination> createDestination( int type, const std::string& name );
-
- public:
-
- /**
- * Returns the Type of Destination that this object represents
- * @returns int type qualifier.
- */
- virtual cms::Destination::DestinationType getDestinationType() const = 0;
+ std::string getDestinationTypeAsString() const;
/**
* Returns true if a temporary Destination
@@ -290,15 +269,22 @@ namespace commands{
* destinations.
*/
virtual bool isComposite() const {
- return physicalName.find_first_of( COMPOSITE_SEPARATOR ) > 0;
+ return physicalName.find_first_of(COMPOSITE_SEPARATOR) > 0;
}
/**
+ * Returns an ArrayList containing all the ActiveMQDestinations that comprise this
+ * Composite destination, if this destination is composite, otherwise it returns an
+ * empty list.
+ */
+ decaf::util::ArrayList< Pointer<ActiveMQDestination> > getCompositeDestinations() const;
+
+ /**
* @return true if the destination matches multiple possible destinations
*/
virtual bool isWildcard() const {
- return physicalName.find_first_of( DestinationFilter::ANY_CHILD ) != std::string::npos ||
- physicalName.find_first_of( DestinationFilter::ANY_DESCENDENT ) != std::string::npos;
+ return physicalName.find_first_of(DestinationFilter::ANY_CHILD) != std::string::npos||
+ physicalName.find_first_of(DestinationFilter::ANY_DESCENDENT) != std::string::npos;
}
/**
@@ -316,6 +302,57 @@ namespace commands{
return NULL;
}
+ /**
+ * Create a new Destination that's of the same type as this one but with
+ * the given destination name.
+ *
+ * @param name
+ * The name
+ *
+ * @return Pointer to a new ActiveMQDestination.
+ */
+ Pointer<ActiveMQDestination> createDestination(const std::string& name) const {
+ return ActiveMQDestination::createDestination(getDestinationType(), name);
+ }
+
+ virtual int compareTo(const ActiveMQDestination& value) const;
+
+ virtual bool equals(const ActiveMQDestination& value) const;
+
+ virtual bool operator==(const ActiveMQDestination& value) const;
+
+ virtual bool operator<(const ActiveMQDestination& value) const;
+
+ public: // Statics
+
+ /**
+ * Create a temporary name from the clientId
+ *
+ * @param clientId
+ * @return
+ */
+ 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
+ *
+ * @param destination
+ * @return the clientId or null if not a temporary destination
+ */
+ static std::string getClientId(const ActiveMQDestination* destination);
+
+ /**
+ * Creates a Destination given the String Name to use and a Type.
+ *
+ * @param type - The Type of Destination to Create
+ * @param name - The Name to use in the creation of the Destination
+ *
+ * @return Pointer to a new ActiveMQDestination instance.
+ */
+ static Pointer<ActiveMQDestination> createDestination(int type, const std::string& name);
+
};
}}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h Fri May 4 21:04:52 2012
@@ -37,8 +37,8 @@
namespace activemq {
namespace commands {
- template< typename T >
- class AMQCPP_API ActiveMQMessageTemplate : public T, public Message {
+ template<typename T>
+ class AMQCPP_API ActiveMQMessageTemplate: public T, public Message {
private:
std::auto_ptr<wireformat::openwire::utils::MessagePropertyInterceptor> propertiesInterceptor;
@@ -46,18 +46,17 @@ namespace commands {
public:
ActiveMQMessageTemplate() : commands::Message(), propertiesInterceptor() {
- this->propertiesInterceptor.reset(
- new wireformat::openwire::utils::MessagePropertyInterceptor(
- this, &this->getMessageProperties() ) );
+ this->propertiesInterceptor.reset(new wireformat::openwire::utils::MessagePropertyInterceptor(this, &this->getMessageProperties()));
}
- virtual ~ActiveMQMessageTemplate() throw() {}
+ virtual ~ActiveMQMessageTemplate() throw () {
+ }
- public: // cms::Message related methods
+ public:
virtual void acknowledge() const {
- try{
- this->getAckHandler()->acknowledgeMessage( this );
+ try {
+ this->getAckHandler()->acknowledgeMessage(this);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
@@ -67,238 +66,238 @@ namespace commands {
this->setReadOnlyProperties(true);
}
- virtual bool equals( const DataStructure* value ) const {
- try{
+ virtual bool equals(const DataStructure* value) const {
+ try {
- if( this == value ) {
+ if (this == value) {
return true;
}
const ActiveMQMessageTemplate<T>* object =
- dynamic_cast< const ActiveMQMessageTemplate<T>* >( value );
+ dynamic_cast<const ActiveMQMessageTemplate<T>*> (value);
- if( object == NULL ) {
+ if (object == NULL) {
return false;
}
decaf::lang::Pointer<MessageId> thisMsgId = this->getMessageId();
decaf::lang::Pointer<MessageId> otherMsgId = object->getMessageId();
- return thisMsgId != NULL && otherMsgId != NULL && otherMsgId->equals( thisMsgId.get() );
+ return thisMsgId != NULL && otherMsgId != NULL && otherMsgId->equals(thisMsgId.get());
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
virtual void clearBody() {
- try{
- this->setContent( std::vector<unsigned char>() );
- this->setReadOnlyBody( false );
+ try {
+ this->setContent(std::vector<unsigned char>());
+ this->setReadOnlyBody(false);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
virtual void clearProperties() {
- try{
+ try {
this->getMessageProperties().clear();
- this->setReadOnlyProperties( false );
+ this->setReadOnlyProperties(false);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
virtual std::vector<std::string> getPropertyNames() const {
- try{
+ try {
return getMessageProperties().keySet();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual bool propertyExists( const std::string& name ) const {
- try{
- return getMessageProperties().containsKey( name );
+ virtual bool propertyExists(const std::string& name) const {
+ try {
+ return getMessageProperties().containsKey(name);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual bool getBooleanProperty( const std::string& name ) const {
+ virtual bool getBooleanProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getBooleanProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getBooleanProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual unsigned char getByteProperty( const std::string& name ) const {
+ virtual unsigned char getByteProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getByteProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getByteProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual double getDoubleProperty( const std::string& name ) const {
+ virtual double getDoubleProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getDoubleProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getDoubleProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual float getFloatProperty( const std::string& name ) const {
+ virtual float getFloatProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getFloatProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getFloatProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual int getIntProperty( const std::string& name ) const {
+ virtual int getIntProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getIntProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getIntProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual long long getLongProperty( const std::string& name ) const {
+ virtual long long getLongProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getLongProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getLongProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual short getShortProperty( const std::string& name ) const {
+ virtual short getShortProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getShortProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getShortProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual std::string getStringProperty( const std::string& name ) const {
+ virtual std::string getStringProperty(const std::string& name) const {
- try{
- return this->propertiesInterceptor->getStringProperty( name );
- } catch( decaf::lang::exceptions::UnsupportedOperationException& ex ) {
- throw activemq::util::CMSExceptionSupport::createMessageFormatException( ex );
+ try {
+ return this->propertiesInterceptor->getStringProperty(name);
+ } catch (decaf::lang::exceptions::UnsupportedOperationException& ex) {
+ throw activemq::util::CMSExceptionSupport::createMessageFormatException(ex);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setBooleanProperty( const std::string& name, bool value ) {
+ virtual void setBooleanProperty(const std::string& name, bool value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setBooleanProperty( name, value );
+ try {
+ this->propertiesInterceptor->setBooleanProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setByteProperty( const std::string& name, unsigned char value ) {
+ virtual void setByteProperty(const std::string& name, unsigned char value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setByteProperty( name, value );
+ try {
+ this->propertiesInterceptor->setByteProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setDoubleProperty( const std::string& name, double value ) {
+ virtual void setDoubleProperty(const std::string& name, double value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setDoubleProperty( name, value );
+ try {
+ this->propertiesInterceptor->setDoubleProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setFloatProperty( const std::string& name, float value ) {
+ virtual void setFloatProperty(const std::string& name, float value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setFloatProperty( name, value );
+ try {
+ this->propertiesInterceptor->setFloatProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setIntProperty( const std::string& name, int value ) {
+ virtual void setIntProperty(const std::string& name, int value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setIntProperty( name, value );
+ try {
+ this->propertiesInterceptor->setIntProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setLongProperty( const std::string& name, long long value ) {
+ virtual void setLongProperty(const std::string& name, long long value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setLongProperty( name, value );
+ try {
+ this->propertiesInterceptor->setLongProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setShortProperty( const std::string& name, short value ) {
+ virtual void setShortProperty(const std::string& name, short value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setShortProperty( name, value );
+ try {
+ this->propertiesInterceptor->setShortProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setStringProperty( const std::string& name, const std::string& value ) {
+ virtual void setStringProperty(const std::string& name, const std::string& value) {
- if( name == "" ) {
- throw cms::CMSException( "Message Property names must not be empty", NULL );
+ if (name == "") {
+ throw cms::CMSException("Message Property names must not be empty", NULL);
}
failIfReadOnlyProperties();
- try{
- this->propertiesInterceptor->setStringProperty( name, value );
+ try {
+ this->propertiesInterceptor->setStringProperty(name, value);
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
@@ -307,31 +306,30 @@ namespace commands {
return this->getCorrelationId();
}
- virtual void setCMSCorrelationID( const std::string& correlationId ) {
- this->setCorrelationId( correlationId );
+ virtual void setCMSCorrelationID(const std::string& correlationId) {
+ this->setCorrelationId(correlationId);
}
virtual int getCMSDeliveryMode() const {
return !this->isPersistent();
}
- virtual void setCMSDeliveryMode( int mode ) {
- this->setPersistent( mode == (int)cms::DeliveryMode::PERSISTENT );
+ virtual void setCMSDeliveryMode(int mode) {
+ this->setPersistent(mode == (int) cms::DeliveryMode::PERSISTENT);
}
virtual const cms::Destination* getCMSDestination() const {
- return dynamic_cast<const cms::Destination*>( this->getDestination().get() );
+ return dynamic_cast<const cms::Destination*> (this->getDestination().get());
}
- virtual void setCMSDestination( const cms::Destination* destination ) {
+ virtual void setCMSDestination(const cms::Destination* destination) {
- try{
- if( destination != NULL ) {
- this->setDestination(
- decaf::lang::Pointer<ActiveMQDestination>(
- dynamic_cast<ActiveMQDestination*>( destination->clone() ) ) );
+ try {
+ if (destination != NULL) {
+ this->setDestination(decaf::lang::Pointer<ActiveMQDestination>(
+ dynamic_cast<ActiveMQDestination*> (destination->clone())));
} else {
- this->getDestination().reset( NULL );
+ this->getDestination().reset(NULL);
}
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -341,46 +339,47 @@ namespace commands {
return this->getExpiration();
}
- virtual void setCMSExpiration( long long expireTime ) {
- this->setExpiration( expireTime );
+ virtual void setCMSExpiration(long long expireTime) {
+ this->setExpiration(expireTime);
}
virtual std::string getCMSMessageID() const {
- try{
- return wireformat::openwire::marshal::BaseDataStreamMarshaller::toString( this->getMessageId().get() );
+ try {
+ return wireformat::openwire::marshal::BaseDataStreamMarshaller::toString(this->getMessageId().get());
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
- virtual void setCMSMessageID( const std::string& id AMQCPP_UNUSED ) {}
+ virtual void setCMSMessageID(const std::string& id AMQCPP_UNUSED ) {
+ }
virtual int getCMSPriority() const {
return this->getPriority();
}
- virtual void setCMSPriority( int priority ) {
- this->setPriority( (unsigned char)priority );
+ virtual void setCMSPriority(int priority) {
+ this->setPriority((unsigned char) priority);
}
virtual bool getCMSRedelivered() const {
return this->getRedeliveryCounter() != 0;
}
- virtual void setCMSRedelivered( bool redelivered AMQCPP_UNUSED ) {}
+ virtual void setCMSRedelivered(bool redelivered AMQCPP_UNUSED ) {
+ }
virtual const cms::Destination* getCMSReplyTo() const {
- return dynamic_cast< const cms::Destination* >( this->getReplyTo().get() );
+ return dynamic_cast<const cms::Destination*> (this->getReplyTo().get());
}
- virtual void setCMSReplyTo( const cms::Destination* destination ) {
+ virtual void setCMSReplyTo(const cms::Destination* destination) {
- try{
- if( destination != NULL ) {
- this->setReplyTo(
- decaf::lang::Pointer<ActiveMQDestination>(
- dynamic_cast<ActiveMQDestination*>( destination->clone() ) ) );
+ try {
+ if (destination != NULL) {
+ this->setReplyTo(decaf::lang::Pointer<ActiveMQDestination>(
+ dynamic_cast<ActiveMQDestination*> (destination->clone())));
} else {
- this->setReplyTo( decaf::lang::Pointer<ActiveMQDestination>() );
+ this->setReplyTo(decaf::lang::Pointer<ActiveMQDestination>());
}
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -390,43 +389,41 @@ namespace commands {
return this->getTimestamp();
}
- virtual void setCMSTimestamp( long long timeStamp ) {
- this->setTimestamp( timeStamp );
+ virtual void setCMSTimestamp(long long timeStamp) {
+ this->setTimestamp(timeStamp);
}
virtual std::string getCMSType() const {
return this->getType();
}
- virtual void setCMSType( const std::string& type ) {
- this->setType( type );
+ virtual void setCMSType(const std::string& type) {
+ this->setType(type);
}
protected:
void failIfWriteOnlyBody() const {
- if( !this->isReadOnlyBody() ){
- throw cms::MessageNotReadableException(
- "message is in write-only mode and cannot be read from", NULL );
+ if (!this->isReadOnlyBody()) {
+ throw cms::MessageNotReadableException("message is in write-only mode and cannot be read from", NULL);
}
}
void failIfReadOnlyBody() const {
- if( this->isReadOnlyBody() ) {
- throw cms::MessageNotWriteableException(
- "Message Body is Read-Only.", NULL );
+ if (this->isReadOnlyBody()) {
+ throw cms::MessageNotWriteableException("Message Body is Read-Only.", NULL);
}
}
void failIfReadOnlyProperties() const {
- if( this->isReadOnlyProperties() ) {
- throw cms::MessageNotWriteableException(
- "Message Properties are Read-Only.", NULL );
+ if (this->isReadOnlyProperties()) {
+ throw cms::MessageNotWriteableException("Message Properties are Read-Only.", NULL);
}
}
};
-}}
+}
+}
#endif /* _ACTIVEMQ_COMMANDS_ACTIVEMQMESSAGETEMPLATE_H_ */
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.cpp?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.cpp Fri May 4 21:04:52 2012
@@ -20,19 +20,22 @@
#include <activemq/core/ActiveMQConnection.h>
#include <activemq/util/CMSExceptionSupport.h>
+#include <decaf/lang/Integer.h>
+
using namespace std;
using namespace activemq;
using namespace activemq::exceptions;
using namespace activemq::commands;
+using namespace decaf::lang;
////////////////////////////////////////////////////////////////////////////////
ActiveMQTempDestination::ActiveMQTempDestination() :
- ActiveMQDestination(), Closeable(), connection( NULL ) {
+ ActiveMQDestination(), Closeable(), connection(NULL), connectionId(), sequenceId(0) {
}
////////////////////////////////////////////////////////////////////////////////
-ActiveMQTempDestination::ActiveMQTempDestination( const std::string& name ) :
- ActiveMQDestination( name ), Closeable(), connection( NULL ) {
+ActiveMQTempDestination::ActiveMQTempDestination(const std::string& name) :
+ ActiveMQDestination(name), Closeable(), connection(NULL) {
}
////////////////////////////////////////////////////////////////////////////////
@@ -45,6 +48,30 @@ unsigned char ActiveMQTempDestination::g
}
////////////////////////////////////////////////////////////////////////////////
+void ActiveMQTempDestination::setPhysicalName(const std::string& physicalName) {
+
+ ActiveMQDestination::setPhysicalName(physicalName);
+ this->connectionId = "";
+ this->sequenceId = 0;
+
+ size_t pos = physicalName.find_last_of(":");
+ if (pos != std::string::npos) {
+ std::string seqStr = physicalName.substr(pos + 1);
+ if (!seqStr.empty()) {
+
+ try {
+ this->sequenceId = Integer::parseInt(seqStr);
+ } catch (decaf::lang::exceptions::NumberFormatException& e) {
+ // Not the expected format so ignore.
+ }
+
+ // The rest should be the connection id.
+ this->connectionId = physicalName.substr(0, pos);
+ }
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
std::string ActiveMQTempDestination::toString() const {
return ActiveMQDestination::toString();
}
@@ -52,8 +79,15 @@ std::string ActiveMQTempDestination::toS
////////////////////////////////////////////////////////////////////////////////
void ActiveMQTempDestination::close() {
try {
- if( this->connection != NULL ) {
- this->connection->destroyDestination( this );
+ if (this->connection != NULL) {
+ Pointer<ActiveMQTempDestination> thisPtr(this);
+ try {
+ this->connection->deleteTempDestination(thisPtr);
+ thisPtr.release();
+ } catch(ActiveMQException& ex) {
+ thisPtr.release();
+ throw;
+ }
}
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.h?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempDestination.h Fri May 4 21:04:52 2012
@@ -46,19 +46,26 @@ namespace commands{
*/
core::ActiveMQConnection* connection;
+ /**
+ * The Connection Id of the Connection that created this Temporary Destination.
+ */
+ std::string connectionId;
+
+ int sequenceId;
+
public:
const static unsigned char ID_ACTIVEMQTEMPDESTINATION = 0;
private:
- ActiveMQTempDestination( const ActiveMQTempDestination& );
- ActiveMQTempDestination& operator= ( const ActiveMQTempDestination& );
+ ActiveMQTempDestination(const ActiveMQTempDestination&);
+ ActiveMQTempDestination& operator=(const ActiveMQTempDestination&);
public:
ActiveMQTempDestination();
- ActiveMQTempDestination( const std::string& name );
+ ActiveMQTempDestination(const std::string& name);
virtual ~ActiveMQTempDestination() throw();
virtual unsigned char getDataStructureType() const;
@@ -67,18 +74,20 @@ namespace commands{
return NULL;
}
- virtual void copyDataStructure( const DataStructure* src ) {
- ActiveMQDestination::copyDataStructure( src );
+ virtual void copyDataStructure(const DataStructure* src) {
+ ActiveMQDestination::copyDataStructure(src);
}
virtual std::string toString() const;
- virtual bool equals( const DataStructure* value ) const {
- return ActiveMQDestination::equals( value );
+ virtual bool equals(const DataStructure* value) const {
+ return ActiveMQDestination::equals(value);
}
virtual void close();
+ virtual void setPhysicalName(const std::string& physicalName);
+
/**
* Sets the Parent Connection that is notified when this destination is
* destroyed.
@@ -86,10 +95,25 @@ namespace commands{
* @param connection
* The parent connection to be used to destroy this destination if closed..
*/
- void setConnection( core::ActiveMQConnection* connection ) {
+ void setConnection(core::ActiveMQConnection* connection) {
this->connection = connection;
}
+ /**
+ * Retrieves the Parent Connection that created this Connection.
+ *
+ * @returns pointer to a Connection if one was set, false otherwise.
+ */
+ core::ActiveMQConnection* getConnection() const {
+ return this->connection;
+ }
+
+ /**
+ * @returns the connection Id of the Connection that created this temporary destination.
+ */
+ std::string getConnectionId() const {
+ return this->connectionId;
+ }
};
}}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.cpp?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.cpp Fri May 4 21:04:52 2012
@@ -28,16 +28,11 @@ ActiveMQTempQueue::ActiveMQTempQueue() :
}
////////////////////////////////////////////////////////////////////////////////
-ActiveMQTempQueue::ActiveMQTempQueue( const std::string& name ) :
- ActiveMQTempDestination( name )
+ActiveMQTempQueue::ActiveMQTempQueue(const std::string& name) : ActiveMQTempDestination( name )
{}
////////////////////////////////////////////////////////////////////////////////
ActiveMQTempQueue::~ActiveMQTempQueue() throw() {
- try {
- this->close();
- }
- AMQ_CATCHALL_NOTHROW()
}
////////////////////////////////////////////////////////////////////////////////
@@ -47,14 +42,17 @@ unsigned char ActiveMQTempQueue::getData
////////////////////////////////////////////////////////////////////////////////
ActiveMQTempQueue* ActiveMQTempQueue::cloneDataStructure() const {
- std::auto_ptr<ActiveMQTempQueue> message( new ActiveMQTempQueue() );
- message->copyDataStructure( this );
- return message.release();
+ std::auto_ptr<ActiveMQTempQueue> copy(new ActiveMQTempQueue());
+ copy->copyDataStructure(this);
+ copy->connection = this->connection;
+ copy->connectionId = this->connectionId;
+ copy->sequenceId = this->sequenceId;
+ return copy.release();
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQTempQueue::copyDataStructure( const DataStructure* src ) {
- ActiveMQTempDestination::copyDataStructure( src );
+void ActiveMQTempQueue::copyDataStructure(const DataStructure* src) {
+ ActiveMQTempDestination::copyDataStructure(src);
}
////////////////////////////////////////////////////////////////////////////////
@@ -63,14 +61,14 @@ std::string ActiveMQTempQueue::toString(
}
////////////////////////////////////////////////////////////////////////////////
-bool ActiveMQTempQueue::equals( const DataStructure* value ) const {
- return ActiveMQDestination::equals( value );
+bool ActiveMQTempQueue::equals(const DataStructure* value) const {
+ return ActiveMQDestination::equals(value);
}
////////////////////////////////////////////////////////////////////////////////
-bool ActiveMQTempQueue::equals( const cms::Destination& value ) const {
- const ActiveMQDestination* dest = dynamic_cast<const ActiveMQDestination*>( &value );
- return ActiveMQDestination::equals( dest );
+bool ActiveMQTempQueue::equals(const cms::Destination& value) const {
+ const ActiveMQDestination* dest = dynamic_cast<const ActiveMQDestination*> (&value);
+ return ActiveMQDestination::equals(dest);
}
////////////////////////////////////////////////////////////////////////////////
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.h?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempQueue.h Fri May 4 21:04:52 2012
@@ -41,24 +41,24 @@ namespace commands{
private:
- ActiveMQTempQueue( const ActiveMQTempQueue& );
- ActiveMQTempQueue& operator= ( const ActiveMQTempQueue& );
+ ActiveMQTempQueue(const ActiveMQTempQueue&);
+ ActiveMQTempQueue& operator=(const ActiveMQTempQueue&);
public:
ActiveMQTempQueue();
- ActiveMQTempQueue( const std::string& name );
- virtual ~ActiveMQTempQueue() throw();
+ ActiveMQTempQueue(const std::string& name);
+ virtual ~ActiveMQTempQueue() throw ();
virtual unsigned char getDataStructureType() const;
virtual ActiveMQTempQueue* cloneDataStructure() const;
- virtual void copyDataStructure( const DataStructure* src );
+ virtual void copyDataStructure(const DataStructure* src);
virtual std::string toString() const;
- virtual bool equals( const DataStructure* value ) const;
+ virtual bool equals(const DataStructure* value) const;
virtual const cms::Destination* getCMSDestination() const {
return this;
@@ -71,20 +71,18 @@ namespace commands{
}
virtual cms::Destination* clone() const {
- return dynamic_cast<cms::Destination*>(
- this->cloneDataStructure() );
+ return dynamic_cast<cms::Destination*> (this->cloneDataStructure());
}
- virtual void copy( const cms::Destination& source ) {
- this->copyDataStructure(
- dynamic_cast<const DataStructure*>( &source ) );
+ virtual void copy(const cms::Destination& source) {
+ this->copyDataStructure(dynamic_cast<const DataStructure*> (&source));
}
virtual const cms::CMSProperties& getCMSProperties() const {
return this->getOptions();
}
- virtual bool equals( const cms::Destination& other ) const;
+ virtual bool equals(const cms::Destination& other) const;
public:
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.cpp?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.cpp Fri May 4 21:04:52 2012
@@ -27,16 +27,11 @@ ActiveMQTempTopic::ActiveMQTempTopic() :
}
////////////////////////////////////////////////////////////////////////////////
-ActiveMQTempTopic::ActiveMQTempTopic( const std::string& name ) :
- ActiveMQTempDestination( name )
-{}
+ActiveMQTempTopic::ActiveMQTempTopic(const std::string& name) : ActiveMQTempDestination(name) {
+}
////////////////////////////////////////////////////////////////////////////////
-ActiveMQTempTopic::~ActiveMQTempTopic() throw() {
- try {
- this->close();
- }
- AMQ_CATCHALL_NOTHROW()
+ActiveMQTempTopic::~ActiveMQTempTopic() throw () {
}
////////////////////////////////////////////////////////////////////////////////
@@ -46,14 +41,18 @@ unsigned char ActiveMQTempTopic::getData
////////////////////////////////////////////////////////////////////////////////
ActiveMQTempTopic* ActiveMQTempTopic::cloneDataStructure() const {
- std::auto_ptr<ActiveMQTempTopic> message( new ActiveMQTempTopic() );
- message->copyDataStructure( this );
- return message.release();
+ std::auto_ptr<ActiveMQTempTopic> copy(new ActiveMQTempTopic());
+ copy->copyDataStructure(this);
+ copy->connection = this->connection;
+ copy->connectionId = this->connectionId;
+ copy->sequenceId = this->sequenceId;
+
+ return copy.release();
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQTempTopic::copyDataStructure( const DataStructure* src ) {
- ActiveMQTempDestination::copyDataStructure( src );
+void ActiveMQTempTopic::copyDataStructure(const DataStructure* src) {
+ ActiveMQTempDestination::copyDataStructure(src);
}
////////////////////////////////////////////////////////////////////////////////
@@ -62,19 +61,19 @@ std::string ActiveMQTempTopic::toString(
}
////////////////////////////////////////////////////////////////////////////////
-bool ActiveMQTempTopic::equals( const DataStructure* value ) const {
- return ActiveMQDestination::equals( value );
+bool ActiveMQTempTopic::equals(const DataStructure* value) const {
+ return ActiveMQDestination::equals(value);
}
////////////////////////////////////////////////////////////////////////////////
-bool ActiveMQTempTopic::equals( const cms::Destination& value ) const {
- const ActiveMQDestination* dest = dynamic_cast<const ActiveMQDestination*>( &value );
- return ActiveMQDestination::equals( dest );
+bool ActiveMQTempTopic::equals(const cms::Destination& value) const {
+ const ActiveMQDestination* dest = dynamic_cast<const ActiveMQDestination*> (&value);
+ return ActiveMQDestination::equals(dest);
}
////////////////////////////////////////////////////////////////////////////////
void ActiveMQTempTopic::destroy() {
- try{
+ try {
close();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.h?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTempTopic.h Fri May 4 21:04:52 2012
@@ -41,50 +41,49 @@ namespace commands{
private:
- ActiveMQTempTopic( const ActiveMQTempTopic& );
- ActiveMQTempTopic& operator= ( const ActiveMQTempTopic& );
+ ActiveMQTempTopic(const ActiveMQTempTopic&);
+ ActiveMQTempTopic& operator=(const ActiveMQTempTopic&);
public:
ActiveMQTempTopic();
- ActiveMQTempTopic( const std::string& name );
- virtual ~ActiveMQTempTopic() throw();
+ ActiveMQTempTopic(const std::string& name);
+ virtual ~ActiveMQTempTopic() throw ();
virtual unsigned char getDataStructureType() const;
virtual ActiveMQTempTopic* cloneDataStructure() const;
- virtual void copyDataStructure( const DataStructure* src );
+ virtual void copyDataStructure(const DataStructure* src);
virtual std::string toString() const;
- virtual bool equals( const DataStructure* value ) const;
+ virtual bool equals(const DataStructure* value) const;
virtual const cms::Destination* getCMSDestination() const {
return this;
}
- public: // CMS Destination
+ public:
+ // CMS Destination
virtual cms::Destination::DestinationType getDestinationType() const {
return cms::Destination::TEMPORARY_TOPIC;
}
virtual cms::Destination* clone() const {
- return dynamic_cast<cms::Destination*>(
- this->cloneDataStructure() );
+ return dynamic_cast<cms::Destination*> (this->cloneDataStructure());
}
- virtual void copy( const cms::Destination& source ) {
- this->copyDataStructure(
- dynamic_cast<const DataStructure*>( &source ) );
+ virtual void copy(const cms::Destination& source) {
+ this->copyDataStructure(dynamic_cast<const DataStructure*> (&source));
}
virtual const cms::CMSProperties& getCMSProperties() const {
return this->getOptions();
}
- virtual bool equals( const cms::Destination& other ) const;
+ virtual bool equals(const cms::Destination& other) const;
public: // CMS Temporary Topic
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ConsumerId.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ConsumerId.cpp?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ConsumerId.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ConsumerId.cpp Fri May 4 21:04:52 2012
@@ -40,7 +40,7 @@ using namespace decaf::lang::exceptions;
*/
////////////////////////////////////////////////////////////////////////////////
-ConsumerId::ConsumerId()
+ConsumerId::ConsumerId()
: BaseDataStructure(), connectionId(""), sessionId(0), value(0), parentId() {
}
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?rev=1334183&r1=1334182&r2=1334183&view=diff
==============================================================================
--- 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 Fri May 4 21:04:52 2012
@@ -22,6 +22,7 @@
#include <activemq/core/ActiveMQSession.h>
#include <activemq/core/ActiveMQProducer.h>
#include <activemq/core/ActiveMQConstants.h>
+#include <activemq/core/AdvisoryConsumer.h>
#include <activemq/core/kernels/ActiveMQSessionKernel.h>
#include <activemq/core/policies/DefaultPrefetchPolicy.h>
#include <activemq/core/policies/DefaultRedeliveryPolicy.h>
@@ -101,6 +102,10 @@ namespace core{
Pointer<ActiveMQProducerKernel>,
commands::ProducerId::COMPARATOR > ProducerMap;
+ typedef decaf::util::concurrent::ConcurrentStlMap< Pointer<commands::ActiveMQTempDestination>,
+ Pointer<commands::ActiveMQTempDestination>,
+ commands::ActiveMQTempDestination::COMPARATOR > TempDestinationMap;
+
public:
static util::IdGenerator CONNECTION_ID_GENERATOR;
@@ -112,6 +117,7 @@ namespace core{
Pointer<ExecutorService> executor;
util::LongSequenceGenerator sessionIds;
+ util::LongSequenceGenerator consumerIdGenerator;
util::LongSequenceGenerator tempDestinationIds;
util::LongSequenceGenerator localTransactionIds;
@@ -128,6 +134,7 @@ namespace core{
bool alwaysSyncSend;
bool useAsyncSend;
bool messagePrioritySupported;
+ bool watchTopicAdvisories;
bool useCompression;
int compressionLevel;
unsigned int sendTimeout;
@@ -144,6 +151,7 @@ namespace core{
Pointer<commands::WireFormatInfo> brokerWireFormatInfo;
Pointer<CountDownLatch> transportInterruptionProcessingComplete;
Pointer<CountDownLatch> brokerInfoReceived;
+ Pointer<AdvisoryConsumer> advisoryConsumer;
Pointer<Exception> firstFailureError;
@@ -153,41 +161,47 @@ namespace core{
decaf::util::concurrent::CopyOnWriteArrayList< Pointer<ActiveMQSessionKernel> > activeSessions;
decaf::util::concurrent::CopyOnWriteArrayList<transport::TransportListener*> transportListeners;
+ TempDestinationMap activeTempDestinations;
+
ConnectionConfig() : properties(),
transport(),
clientIdGenerator(),
scheduler(),
sessionIds(),
+ consumerIdGenerator(),
tempDestinationIds(),
localTransactionIds(),
brokerURL(""),
- clientIDSet( false ),
- isConnectionInfoSentToBroker( false ),
- userSpecifiedClientID( false ),
+ clientIDSet(false),
+ isConnectionInfoSentToBroker(false),
+ userSpecifiedClientID(false),
ensureConnectionInfoSentMutex(),
mutex(),
- dispatchAsync( true ),
- alwaysSyncSend( false ),
- useAsyncSend( false ),
- messagePrioritySupported( true ),
- useCompression( false ),
- compressionLevel( -1 ),
- sendTimeout( 0 ),
- closeTimeout( 15000 ),
- producerWindowSize( 0 ),
- defaultPrefetchPolicy( NULL ),
- defaultRedeliveryPolicy( NULL ),
- exceptionListener( NULL ),
+ dispatchAsync(true),
+ alwaysSyncSend(false),
+ useAsyncSend(false),
+ messagePrioritySupported(true),
+ watchTopicAdvisories(true),
+ useCompression(false),
+ compressionLevel(-1),
+ sendTimeout(0),
+ closeTimeout(15000),
+ producerWindowSize(0),
+ defaultPrefetchPolicy(NULL),
+ defaultRedeliveryPolicy(NULL),
+ exceptionListener(NULL),
connectionInfo(),
brokerInfo(),
brokerWireFormatInfo(),
transportInterruptionProcessingComplete(),
brokerInfoReceived(),
+ advisoryConsumer(),
firstFailureError(),
dispatchers(),
activeProducers(),
activeSessions(),
- transportListeners() {
+ transportListeners(),
+ activeTempDestinations() {
this->defaultPrefetchPolicy.reset(new DefaultPrefetchPolicy());
this->defaultRedeliveryPolicy.reset(new DefaultRedeliveryPolicy());
@@ -523,6 +537,16 @@ void ActiveMQConnection::close() {
}
}
+ // As TemporaryQueue and TemporaryTopic instances are bound
+ // to a connection we should just delete them after the connection
+ // is closed to free up memory
+ std::vector< Pointer<ActiveMQTempDestination> > values = this->config->activeTempDestinations.values();
+ std::vector< Pointer<ActiveMQTempDestination> >::iterator iterator = values.begin();
+ for(; iterator != values.end(); ++iterator) {
+ Pointer<ActiveMQTempDestination> dest = *iterator;
+ dest->close();
+ }
+
try {
if (this->config->executor != NULL) {
this->config->executor->shutdown();
@@ -829,10 +853,7 @@ void ActiveMQConnection::onCommand(const
} else if (command->isConnectionError()) {
Pointer<ConnectionError> connectionError = command.dynamicCast<ConnectionError>();
- Pointer<BrokerError> brokerError = connectionError->getException();
- if (brokerError != NULL) {
- this->onAsyncException(brokerError->createExceptionObject());
- }
+ this->config->executor->execute(new ConnectionErrorRunnable(this, connectionError));
} else if (command->isConsumerControl()) {
this->onConsumerControl(command);
@@ -907,6 +928,19 @@ void ActiveMQConnection::onAsyncExceptio
}
////////////////////////////////////////////////////////////////////////////////
+void ActiveMQConnection::onClientInternalException(const decaf::lang::Exception& ex) {
+
+ if ( !closed.get() && !closing.get() ) {
+
+ if (this->config->exceptionListener != NULL) {
+ this->config->executor->execute(new OnAsyncExceptionRunnable(this, ex));
+ }
+
+ // TODO Turn this into an invocation on a special ClientInternalExceptionListener
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
void ActiveMQConnection::transportInterrupted() {
this->config->transportInterruptionProcessingComplete.reset(
@@ -1032,6 +1066,12 @@ void ActiveMQConnection::ensureConnectio
syncRequest(this->config->connectionInfo);
this->config->isConnectionInfoSentToBroker = true;
+
+ Pointer<SessionId> sessionId(new SessionId(this->config->connectionInfo->getConnectionId().get(), -1));
+ Pointer<ConsumerId> consumerId(new ConsumerId(*sessionId, this->config->consumerIdGenerator.getNextSequenceId()));
+ if (this->config->watchTopicAdvisories) {
+ this->config->advisoryConsumer.reset(new AdvisoryConsumer(this, consumerId));
+ }
}
}
AMQ_CATCH_RETHROW( ActiveMQException )
@@ -1116,15 +1156,15 @@ void ActiveMQConnection::signalInterrupt
Pointer<CountDownLatch> cdl = this->config->transportInterruptionProcessingComplete;
- if( cdl->getCount() == 0 ) {
+ if (cdl->getCount() == 0) {
- this->config->transportInterruptionProcessingComplete.reset( NULL );
+ this->config->transportInterruptionProcessingComplete.reset(NULL);
FailoverTransport* failoverTransport =
- dynamic_cast<FailoverTransport*>( this->config->transport->narrow( typeid( FailoverTransport ) ) );
+ dynamic_cast<FailoverTransport*>(this->config->transport->narrow(typeid(FailoverTransport)));
- if( failoverTransport != NULL ) {
+ if (failoverTransport != NULL) {
failoverTransport->setConnectionInterruptProcessingComplete(
- this->config->connectionInfo->getConnectionId() );
+ this->config->connectionInfo->getConnectionId());
}
}
}
@@ -1344,3 +1384,99 @@ const decaf::util::Properties& ActiveMQC
ExecutorService* ActiveMQConnection::getExecutor() const {
return this->config->executor.get();
}
+
+////////////////////////////////////////////////////////////////////////////////
+bool ActiveMQConnection::isWatchTopicAdvisories() const {
+ return this->config->watchTopicAdvisories;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQConnection::setWatchTopicAdvisories(bool value) {
+ this->config->watchTopicAdvisories = value;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQConnection::addTempDestination(Pointer<ActiveMQTempDestination> destination) {
+ this->config->activeTempDestinations.put(destination, destination);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQConnection::removeTempDestination(Pointer<ActiveMQTempDestination> destination) {
+ this->config->activeTempDestinations.remove(destination);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQConnection::deleteTempDestination(Pointer<ActiveMQTempDestination> destination) {
+
+ try {
+
+ if (destination == NULL) {
+ throw NullPointerException(__FILE__, __LINE__, "Destination passed was NULL");
+ }
+
+ checkClosedOrFailed();
+ ensureConnectionInfoSent();
+
+ Pointer< Iterator< Pointer<ActiveMQSessionKernel> > > iterator(this->config->activeSessions.iterator());
+ while (iterator->hasNext()) {
+ Pointer<ActiveMQSessionKernel> session = iterator->next();
+ if (session->isInUse(destination)) {
+ throw new ActiveMQException(__FILE__, __LINE__, "A consumer is consuming from the temporary destination");
+ }
+ }
+
+ this->config->activeTempDestinations.remove(destination);
+
+ Pointer<DestinationInfo> command(new DestinationInfo());
+
+ command->setConnectionId(this->config->connectionInfo->getConnectionId());
+ command->setOperationType(ActiveMQConstants::DESTINATION_REMOVE_OPERATION);
+ command->setDestination(Pointer<ActiveMQDestination> (destination->cloneDataStructure()));
+
+ // Send the message to the broker.
+ syncRequest(command);
+ }
+ AMQ_CATCH_RETHROW( NullPointerException )
+ AMQ_CATCH_RETHROW( decaf::lang::exceptions::IllegalStateException )
+ AMQ_CATCH_RETHROW( ActiveMQException )
+ AMQ_CATCH_EXCEPTION_CONVERT( Exception, ActiveMQException )
+ AMQ_CATCHALL_THROW( ActiveMQException )
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQConnection::cleanUpTempDestinations() {
+
+ if (this->config->activeTempDestinations.isEmpty()) {
+ return;
+ }
+
+ std::vector< Pointer<ActiveMQTempDestination> > values = this->config->activeTempDestinations.values();
+ std::vector< Pointer<ActiveMQTempDestination> >::iterator iterator = values.begin();
+ for(; iterator != values.end(); ++iterator) {
+ Pointer<ActiveMQTempDestination> dest = *iterator;
+
+ try {
+
+ // Only delete this temporary destination if it was created from this connection, since the
+ // advisory consumer tracks all temporary destinations there can be others in our mapping that
+ // this connection did not create.
+ std::string thisConnectionId = this->config->connectionInfo->getConnectionId() != NULL ?
+ this->config->connectionInfo->getConnectionId()->toString() : "";
+ if (dest->getConnectionId() == thisConnectionId) {
+ this->deleteTempDestination(dest);
+ }
+
+ } catch(Exception& ex) {
+ }
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+bool ActiveMQConnection::isDeleted(Pointer<ActiveMQTempDestination> destination) const {
+
+ if (this->config->advisoryConsumer == NULL) {
+ return false;
+ }
+
+ return !this->config->activeTempDestinations.containsKey(destination);
+}