| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| #include "ActiveMQConnectionFactory.h" |
| |
| #include <cms/MessageTransformer.h> |
| #include <decaf/net/URI.h> |
| #include <decaf/util/Properties.h> |
| #include <decaf/util/concurrent/Mutex.h> |
| #include <decaf/lang/Boolean.h> |
| #include <decaf/lang/Integer.h> |
| #include <decaf/lang/Long.h> |
| #include <decaf/lang/Pointer.h> |
| #include <decaf/lang/Math.h> |
| #include <decaf/lang/exceptions/NullPointerException.h> |
| #include <activemq/exceptions/ExceptionDefines.h> |
| #include <activemq/transport/TransportRegistry.h> |
| #include <activemq/core/ActiveMQConnection.h> |
| #include <activemq/core/ActiveMQConstants.h> |
| #include <activemq/core/ActiveMQMessageAudit.h> |
| #include <activemq/core/policies/DefaultPrefetchPolicy.h> |
| #include <activemq/core/policies/DefaultRedeliveryPolicy.h> |
| #include <activemq/util/URISupport.h> |
| #include <activemq/util/CompositeData.h> |
| #include <memory> |
| |
| using namespace std; |
| using namespace activemq; |
| using namespace activemq::util; |
| using namespace activemq::core; |
| using namespace activemq::core::policies; |
| using namespace activemq::exceptions; |
| using namespace activemq::transport; |
| using namespace decaf; |
| using namespace decaf::net; |
| using namespace decaf::util; |
| using namespace decaf::util::concurrent; |
| using namespace decaf::lang; |
| using namespace decaf::lang::exceptions; |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| const std::string ActiveMQConnectionFactory::DEFAULT_URI = "failover:(tcp://localhost:61616)"; |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| namespace activemq{ |
| namespace core{ |
| |
| class FactorySettings { |
| private: |
| |
| FactorySettings(const FactorySettings&); |
| FactorySettings& operator=(const FactorySettings&); |
| |
| public: |
| |
| Mutex configLock; |
| |
| Pointer<Properties> properties; |
| |
| std::string username; |
| std::string password; |
| std::string clientId; |
| |
| URI brokerURI; |
| |
| bool dispatchAsync; |
| bool alwaysSyncSend; |
| bool useAsyncSend; |
| bool sendAcksAsync; |
| bool messagePrioritySupported; |
| bool useCompression; |
| bool useRetroactiveConsumer; |
| bool watchTopicAdvisories; |
| bool checkForDuplicates; |
| bool optimizeAcknowledge; |
| bool exclusiveConsumer; |
| bool transactedIndividualAck; |
| bool nonBlockingRedelivery; |
| bool alwaysSessionAsync; |
| int compressionLevel; |
| unsigned int sendTimeout; |
| unsigned int closeTimeout; |
| unsigned int producerWindowSize; |
| int auditDepth; |
| int auditMaximumProducerNumber; |
| long long optimizeAcknowledgeTimeOut; |
| long long optimizedAckScheduledAckInterval; |
| long long consumerFailoverRedeliveryWaitPeriod; |
| bool consumerExpiryCheckEnabled; |
| |
| cms::ExceptionListener* defaultListener; |
| cms::MessageTransformer* defaultTransformer; |
| std::auto_ptr<PrefetchPolicy> defaultPrefetchPolicy; |
| std::auto_ptr<RedeliveryPolicy> defaultRedeliveryPolicy; |
| |
| FactorySettings() : configLock(), |
| properties(new Properties()), |
| username(), |
| password(), |
| clientId(), |
| brokerURI(ActiveMQConnectionFactory::DEFAULT_URI), |
| dispatchAsync(true), |
| alwaysSyncSend(false), |
| useAsyncSend(false), |
| sendAcksAsync(true), |
| messagePrioritySupported(false), |
| useCompression(false), |
| useRetroactiveConsumer(false), |
| watchTopicAdvisories(true), |
| checkForDuplicates(true), |
| optimizeAcknowledge(false), |
| exclusiveConsumer(false), |
| transactedIndividualAck(false), |
| nonBlockingRedelivery(false), |
| alwaysSessionAsync(true), |
| compressionLevel(-1), |
| sendTimeout(0), |
| closeTimeout(15000), |
| producerWindowSize(0), |
| auditDepth(ActiveMQMessageAudit::DEFAULT_WINDOW_SIZE), |
| auditMaximumProducerNumber(ActiveMQMessageAudit::MAXIMUM_PRODUCER_COUNT), |
| optimizeAcknowledgeTimeOut(300), |
| optimizedAckScheduledAckInterval(0), |
| consumerFailoverRedeliveryWaitPeriod(0), |
| consumerExpiryCheckEnabled(true), |
| defaultListener(NULL), |
| defaultTransformer(NULL), |
| defaultPrefetchPolicy(new DefaultPrefetchPolicy()), |
| defaultRedeliveryPolicy(new DefaultRedeliveryPolicy()) { |
| } |
| |
| void updateConfiguration(const URI& uri) { |
| |
| this->brokerURI = uri; |
| this->properties->clear(); |
| |
| if (uri.getQuery() != "") { |
| // Not a composite URI so this works fine. |
| try { |
| URISupport::parseQuery(uri.getQuery(), properties.get()); |
| } catch (URISyntaxException& ex) { |
| } |
| } else { |
| // Composite URI won't indicate it has a query even if it does. |
| try { |
| CompositeData composite = URISupport::parseComposite(uri); |
| *this->properties = composite.getParameters(); |
| } catch (URISyntaxException& ex) { |
| } |
| } |
| |
| // Check the connection options |
| this->alwaysSyncSend = Boolean::parseBoolean( |
| properties->getProperty(core::ActiveMQConstants::toString( |
| core::ActiveMQConstants::CONNECTION_ALWAYSSYNCSEND), Boolean::toString(alwaysSyncSend))); |
| this->useAsyncSend = Boolean::parseBoolean( |
| properties->getProperty(core::ActiveMQConstants::toString( |
| core::ActiveMQConstants::CONNECTION_USEASYNCSEND), Boolean::toString(useAsyncSend))); |
| this->useCompression = Boolean::parseBoolean( |
| properties->getProperty(core::ActiveMQConstants::toString( |
| core::ActiveMQConstants::CONNECTION_USECOMPRESSION), Boolean::toString(useCompression))); |
| this->compressionLevel = Integer::parseInt( |
| properties->getProperty("connection.compressionLevel", Integer::toString(compressionLevel))); |
| this->messagePrioritySupported = Boolean::parseBoolean( |
| properties->getProperty("connection.messagePrioritySupported", Boolean::toString(messagePrioritySupported))); |
| this->checkForDuplicates = Boolean::parseBoolean( |
| properties->getProperty("connection.checkForDuplicates", Boolean::toString(checkForDuplicates))); |
| this->auditDepth = Integer::parseInt( |
| properties->getProperty("connection.auditDepth", Integer::toString(auditDepth))); |
| this->auditMaximumProducerNumber = Integer::parseInt( |
| properties->getProperty("connection.auditMaximumProducerNumber", Integer::toString(auditMaximumProducerNumber))); |
| this->dispatchAsync = Boolean::parseBoolean( |
| properties->getProperty(core::ActiveMQConstants::toString( |
| core::ActiveMQConstants::CONNECTION_DISPATCHASYNC), Boolean::toString(dispatchAsync))); |
| this->producerWindowSize = Integer::parseInt( |
| properties->getProperty(core::ActiveMQConstants::toString( |
| core::ActiveMQConstants::CONNECTION_PRODUCERWINDOWSIZE), Integer::toString(producerWindowSize))); |
| this->sendTimeout = decaf::lang::Integer::parseInt( |
| properties->getProperty(core::ActiveMQConstants::toString( |
| core::ActiveMQConstants::CONNECTION_SENDTIMEOUT), Integer::toString(sendTimeout))); |
| this->closeTimeout = decaf::lang::Integer::parseInt( |
| properties->getProperty(core::ActiveMQConstants::toString( |
| core::ActiveMQConstants::CONNECTION_CLOSETIMEOUT), Integer::toString(closeTimeout))); |
| this->clientId = properties->getProperty( |
| core::ActiveMQConstants::toString(core::ActiveMQConstants::PARAM_CLIENTID), clientId); |
| this->username = properties->getProperty( |
| core::ActiveMQConstants::toString(core::ActiveMQConstants::PARAM_USERNAME), username); |
| this->password = properties->getProperty( |
| core::ActiveMQConstants::toString(core::ActiveMQConstants::PARAM_PASSWORD), password); |
| this->optimizeAcknowledge = Boolean::parseBoolean( |
| properties->getProperty("connection.optimizeAcknowledge", Boolean::toString(optimizeAcknowledge))); |
| this->exclusiveConsumer = Boolean::parseBoolean( |
| properties->getProperty("connection.exclusiveConsumer", Boolean::toString(exclusiveConsumer))); |
| this->transactedIndividualAck = Boolean::parseBoolean( |
| properties->getProperty("connection.transactedIndividualAck", Boolean::toString(transactedIndividualAck))); |
| this->useRetroactiveConsumer = Boolean::parseBoolean( |
| properties->getProperty("connection.useRetroactiveConsumer", Boolean::toString(useRetroactiveConsumer))); |
| this->sendAcksAsync = Boolean::parseBoolean( |
| properties->getProperty("connection.sendAcksAsync", Boolean::toString(sendAcksAsync))); |
| this->optimizeAcknowledgeTimeOut = Long::parseLong( |
| properties->getProperty("connection.optimizeAcknowledgeTimeOut", Long::toString(optimizeAcknowledgeTimeOut))); |
| this->optimizedAckScheduledAckInterval = Long::parseLong( |
| properties->getProperty("connection.optimizedAckScheduledAckInterval", Long::toString(optimizedAckScheduledAckInterval))); |
| this->consumerFailoverRedeliveryWaitPeriod = Long::parseLong( |
| properties->getProperty("connection.consumerFailoverRedeliveryWaitPeriod", Long::toString(consumerFailoverRedeliveryWaitPeriod))); |
| this->nonBlockingRedelivery = Boolean::parseBoolean( |
| properties->getProperty("connection.nonBlockingRedelivery", Boolean::toString(nonBlockingRedelivery))); |
| this->watchTopicAdvisories = Boolean::parseBoolean( |
| properties->getProperty("connection.watchTopicAdvisories", Boolean::toString(watchTopicAdvisories))); |
| this->alwaysSessionAsync = Boolean::parseBoolean( |
| properties->getProperty("connection.alwaysSessionAsync", Boolean::toString(alwaysSessionAsync))); |
| this->consumerExpiryCheckEnabled = Boolean::parseBoolean( |
| properties->getProperty("connection.consumerExpiryCheckEnabled", Boolean::toString(consumerExpiryCheckEnabled))); |
| |
| this->defaultPrefetchPolicy->configure(*properties); |
| this->defaultRedeliveryPolicy->configure(*properties); |
| } |
| |
| static URI createURI(const std::string& uriString) { |
| try { |
| return URI(uriString); |
| } catch (URISyntaxException& ex) { |
| throw cms::CMSException("Invalid Connection Uri detected."); |
| } |
| } |
| }; |
| |
| }} |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::ConnectionFactory* cms::ConnectionFactory::createCMSConnectionFactory(const std::string& brokerURI) { |
| return new ActiveMQConnectionFactory( brokerURI ); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| ActiveMQConnectionFactory::ActiveMQConnectionFactory() : settings(new FactorySettings()) { |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| ActiveMQConnectionFactory::ActiveMQConnectionFactory(const std::string& uri, |
| const std::string& username, |
| const std::string& password) : settings( new FactorySettings() ) { |
| |
| this->setBrokerURI(FactorySettings::createURI(uri)); |
| |
| // Store login data in the properties |
| if (!username.empty()) { |
| this->settings->username = username; |
| } |
| if (!password.empty()) { |
| this->settings->password = password; |
| } |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| ActiveMQConnectionFactory::ActiveMQConnectionFactory(const decaf::net::URI& uri, |
| const std::string& username, |
| const std::string& password) : settings( new FactorySettings() ) { |
| |
| this->setBrokerURI(uri); |
| |
| // Store login data in the properties |
| if (!username.empty()) { |
| this->settings->username = username; |
| } |
| if (!password.empty()) { |
| this->settings->password = password; |
| } |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| ActiveMQConnectionFactory::~ActiveMQConnectionFactory() { |
| try { |
| delete this->settings; |
| } |
| DECAF_CATCH_NOTHROW(Exception) |
| DECAF_CATCHALL_NOTHROW() |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::Connection* ActiveMQConnectionFactory::createConnection() { |
| return doCreateConnection(settings->brokerURI, settings->username, settings->password, settings->clientId); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::Connection* ActiveMQConnectionFactory::createConnection(const std::string& username, const std::string& password) { |
| return doCreateConnection(settings->brokerURI, username, password, settings->clientId); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::Connection* ActiveMQConnectionFactory::createConnection(const std::string& username, const std::string& password, const std::string& clientId) { |
| return doCreateConnection(settings->brokerURI, username, password, clientId); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::Connection* ActiveMQConnectionFactory::doCreateConnection(const decaf::net::URI& uri, |
| const std::string& username, |
| const std::string& password, |
| const std::string& clientId) { |
| |
| Pointer<Transport> transport; |
| auto_ptr<ActiveMQConnection> connection; |
| |
| try { |
| |
| synchronized(&this->settings->configLock) { |
| |
| this->setBrokerURI(uri); |
| |
| // Store login data in the properties |
| if (!username.empty()) { |
| this->settings->username = username; |
| } |
| if (!password.empty()) { |
| this->settings->password = password; |
| } |
| if (!clientId.empty()) { |
| this->settings->clientId = clientId; |
| } |
| |
| // Use the TransportBuilder to get our Transport |
| transport = TransportRegistry::getInstance().findFactory(uri.getScheme())->create(uri); |
| |
| if (transport == NULL) { |
| throw ActiveMQException(__FILE__, __LINE__, "ActiveMQConnectionFactory::createConnection - " |
| "failed creating new Transport"); |
| } |
| |
| Pointer<Properties> properties(this->settings->properties->clone()); |
| |
| // Create and Return the new connection object. |
| connection.reset(createActiveMQConnection(transport, properties)); |
| |
| // Set all options parsed from the URI. |
| configureConnection(connection.get()); |
| |
| // Now start the connection since all other configuration is done. |
| transport->start(); |
| |
| if (!this->settings->clientId.empty()) { |
| connection->setDefaultClientId(this->settings->clientId); |
| } |
| } |
| |
| return connection.release(); |
| } catch (cms::CMSException& ex) { |
| ex.setMark(__FILE__, __LINE__); |
| throw ex; |
| } catch (activemq::exceptions::ActiveMQException& ex) { |
| ex.setMark(__FILE__, __LINE__); |
| throw ex.convertToCMSException(); |
| } catch (decaf::lang::Exception& ex) { |
| ex.setMark(__FILE__, __LINE__); |
| activemq::exceptions::ActiveMQException amqEx(ex); |
| throw amqEx.convertToCMSException(); |
| } catch (std::exception& ex) { |
| throw cms::CMSException(ex.what(), NULL); |
| } catch (...) { |
| throw cms::CMSException("Caught Unknown Exception", NULL); |
| } |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| ActiveMQConnection* ActiveMQConnectionFactory::createActiveMQConnection( |
| const Pointer<transport::Transport>& transport, |
| const Pointer<decaf::util::Properties>& properties) { |
| |
| return new ActiveMQConnection(transport, properties); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::Connection* ActiveMQConnectionFactory::createConnection(const std::string& uri, |
| const std::string& username, |
| const std::string& password, |
| const std::string& clientId) { |
| ActiveMQConnectionFactory factory; |
| return factory.doCreateConnection(URI(uri), username, password, clientId); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::configureConnection(ActiveMQConnection* connection) { |
| |
| connection->setUsername(this->settings->username); |
| connection->setPassword(this->settings->password); |
| connection->setDispatchAsync(this->settings->dispatchAsync); |
| connection->setAlwaysSyncSend(this->settings->alwaysSyncSend); |
| connection->setUseAsyncSend(this->settings->useAsyncSend); |
| connection->setUseCompression(this->settings->useCompression); |
| connection->setCompressionLevel(this->settings->compressionLevel); |
| connection->setSendTimeout(this->settings->sendTimeout); |
| connection->setCloseTimeout(this->settings->closeTimeout); |
| connection->setProducerWindowSize(this->settings->producerWindowSize); |
| connection->setPrefetchPolicy(this->settings->defaultPrefetchPolicy->clone()); |
| connection->setRedeliveryPolicy(this->settings->defaultRedeliveryPolicy->clone()); |
| connection->setMessagePrioritySupported(this->settings->messagePrioritySupported); |
| connection->setWatchTopicAdvisories(this->settings->watchTopicAdvisories); |
| connection->setCheckForDuplicates(this->settings->checkForDuplicates); |
| connection->setAuditDepth(this->settings->auditDepth); |
| connection->setAuditMaximumProducerNumber(this->settings->auditMaximumProducerNumber); |
| connection->setOptimizeAcknowledge(this->settings->optimizeAcknowledge); |
| connection->setOptimizeAcknowledgeTimeOut(this->settings->optimizeAcknowledgeTimeOut); |
| connection->setOptimizedAckScheduledAckInterval(this->settings->optimizedAckScheduledAckInterval); |
| connection->setSendAcksAsync(this->settings->sendAcksAsync); |
| connection->setExclusiveConsumer(this->settings->exclusiveConsumer); |
| connection->setTransactedIndividualAck(this->settings->transactedIndividualAck); |
| connection->setUseRetroactiveConsumer(this->settings->useRetroactiveConsumer); |
| connection->setNonBlockingRedelivery(this->settings->nonBlockingRedelivery); |
| connection->setConsumerFailoverRedeliveryWaitPeriod(this->settings->consumerFailoverRedeliveryWaitPeriod); |
| connection->setAlwaysSessionAsync(this->settings->alwaysSessionAsync); |
| connection->setConsumerExpiryCheckEnabled(this->settings->consumerExpiryCheckEnabled); |
| |
| if (this->settings->defaultListener) { |
| connection->setExceptionListener(this->settings->defaultListener); |
| } |
| |
| if (this->settings->defaultTransformer) { |
| connection->setMessageTransformer(this->settings->defaultTransformer); |
| } |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setUsername(const std::string& username) { |
| settings->username = username; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| const std::string& ActiveMQConnectionFactory::getUsername() const { |
| return settings->username; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setPassword(const std::string& password) { |
| settings->password = password; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| const std::string& ActiveMQConnectionFactory::getPassword() const { |
| return settings->password; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| std::string ActiveMQConnectionFactory::getClientId() const { |
| return this->settings->clientId; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setClientId(const std::string& clientId) { |
| this->settings->clientId = clientId; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setBrokerURI(const std::string& uri) { |
| this->setBrokerURI(FactorySettings::createURI(uri)); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setBrokerURI(const decaf::net::URI& uri) { |
| synchronized(&this->settings->configLock) { |
| this->settings->updateConfiguration(uri); |
| } |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| const decaf::net::URI& ActiveMQConnectionFactory::getBrokerURI() const { |
| return settings->brokerURI; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setExceptionListener(cms::ExceptionListener* listener) { |
| this->settings->defaultListener = listener; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::ExceptionListener* ActiveMQConnectionFactory::getExceptionListener() const { |
| return this->settings->defaultListener; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setMessageTransformer(cms::MessageTransformer* transformer) { |
| this->settings->defaultTransformer = transformer; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| cms::MessageTransformer* ActiveMQConnectionFactory::getMessageTransformer() const { |
| return this->settings->defaultTransformer; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setPrefetchPolicy(PrefetchPolicy* policy) { |
| this->settings->defaultPrefetchPolicy.reset(policy); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| PrefetchPolicy* ActiveMQConnectionFactory::getPrefetchPolicy() const { |
| return this->settings->defaultPrefetchPolicy.get(); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setRedeliveryPolicy(RedeliveryPolicy* policy) { |
| this->settings->defaultRedeliveryPolicy.reset(policy); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| RedeliveryPolicy* ActiveMQConnectionFactory::getRedeliveryPolicy() const { |
| return this->settings->defaultRedeliveryPolicy.get(); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isDispatchAsync() const { |
| return this->settings->dispatchAsync; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setDispatchAsync(bool value) { |
| this->settings->dispatchAsync = value; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isAlwaysSyncSend() const { |
| return this->settings->alwaysSyncSend; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setAlwaysSyncSend(bool value) { |
| this->settings->alwaysSyncSend = value; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isUseAsyncSend() const { |
| return this->settings->useAsyncSend; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setUseAsyncSend(bool value) { |
| this->settings->useAsyncSend = value; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isSendAcksAsync() const { |
| return this->settings->sendAcksAsync; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setSendAcksAsync(bool sendAcksAsync) { |
| this->settings->sendAcksAsync = sendAcksAsync; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isUseCompression() const { |
| return this->settings->useCompression; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setUseCompression(bool value) { |
| this->settings->useCompression = value; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| int ActiveMQConnectionFactory::getCompressionLevel() const { |
| return this->settings->compressionLevel; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setCompressionLevel(int value) { |
| |
| if (value < 0) { |
| this->settings->compressionLevel = -1; |
| } |
| |
| this->settings->compressionLevel = Math::min(value, 9); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| unsigned int ActiveMQConnectionFactory::getSendTimeout() const { |
| return this->settings->sendTimeout; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setSendTimeout(unsigned int timeout) { |
| this->settings->sendTimeout = timeout; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| unsigned int ActiveMQConnectionFactory::getCloseTimeout() const { |
| return this->settings->closeTimeout; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setCloseTimeout(unsigned int timeout) { |
| this->settings->closeTimeout = timeout; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| unsigned int ActiveMQConnectionFactory::getProducerWindowSize() const { |
| return this->settings->producerWindowSize; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setProducerWindowSize(unsigned int windowSize) { |
| this->settings->producerWindowSize = windowSize; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isMessagePrioritySupported() const { |
| return this->settings->messagePrioritySupported; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setMessagePrioritySupported(bool value) { |
| this->settings->messagePrioritySupported = value; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isWatchTopicAdvisories() const { |
| return this->settings->watchTopicAdvisories; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setWatchTopicAdvisories(bool value) { |
| this->settings->watchTopicAdvisories = value; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| int ActiveMQConnectionFactory::getAuditDepth() const { |
| return this->settings->auditDepth; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setAuditDepth(int auditDepth) { |
| this->settings->auditDepth = auditDepth; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| int ActiveMQConnectionFactory::getAuditMaximumProducerNumber() const { |
| return this->settings->auditMaximumProducerNumber; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setAuditMaximumProducerNumber(int auditMaximumProducerNumber) { |
| this->settings->auditMaximumProducerNumber = auditMaximumProducerNumber; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isCheckForDuplicates() const { |
| return this->settings->checkForDuplicates; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setCheckForDuplicates(bool checkForDuplicates) { |
| this->settings->checkForDuplicates = checkForDuplicates; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isTransactedIndividualAck() const { |
| return this->settings->transactedIndividualAck; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setTransactedIndividualAck(bool transactedIndividualAck) { |
| this->settings->transactedIndividualAck = transactedIndividualAck; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isNonBlockingRedelivery() const { |
| return this->settings->nonBlockingRedelivery; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setNonBlockingRedelivery(bool nonBlockingRedelivery) { |
| this->settings->nonBlockingRedelivery = nonBlockingRedelivery; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isOptimizeAcknowledge() const { |
| return this->settings->optimizeAcknowledge; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setOptimizeAcknowledge(bool optimizeAcknowledge) { |
| this->settings->optimizeAcknowledge = optimizeAcknowledge; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| long long ActiveMQConnectionFactory::getOptimizeAcknowledgeTimeOut() const { |
| return this->settings->optimizeAcknowledgeTimeOut; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setOptimizeAcknowledgeTimeOut(long long optimizeAcknowledgeTimeOut) { |
| this->settings->optimizeAcknowledgeTimeOut = optimizeAcknowledgeTimeOut; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| long long ActiveMQConnectionFactory::getOptimizedAckScheduledAckInterval() const { |
| return this->settings->optimizedAckScheduledAckInterval; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setOptimizedAckScheduledAckInterval(long long optimizedAckScheduledAckInterval) { |
| this->settings->optimizedAckScheduledAckInterval = optimizedAckScheduledAckInterval; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| long long ActiveMQConnectionFactory::getConsumerFailoverRedeliveryWaitPeriod() const { |
| return this->settings->consumerFailoverRedeliveryWaitPeriod; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setConsumerFailoverRedeliveryWaitPeriod(long long value) { |
| this->settings->consumerFailoverRedeliveryWaitPeriod = value; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isUseRetroactiveConsumer() const { |
| return this->settings->useRetroactiveConsumer; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setUseRetroactiveConsumer(bool useRetroactiveConsumer) { |
| this->settings->useRetroactiveConsumer = useRetroactiveConsumer; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isExclusiveConsumer() const { |
| return this->settings->exclusiveConsumer; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setExclusiveConsumer(bool exclusiveConsumer) { |
| this->settings->exclusiveConsumer = exclusiveConsumer; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isAlwaysSessionAsync() const { |
| return this->settings->alwaysSessionAsync; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setAlwaysSessionAsync(bool alwaysSessionAsync) { |
| this->settings->alwaysSessionAsync = alwaysSessionAsync; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| bool ActiveMQConnectionFactory::isConsumerExpiryCheckEnabled() { |
| return this->settings->consumerExpiryCheckEnabled; |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void ActiveMQConnectionFactory::setConsumerExpiryCheckEnabled(bool consumerExpiryCheckEnabled) { |
| this->settings->consumerExpiryCheckEnabled = consumerExpiryCheckEnabled; |
| } |
