Remove incomplete discovery transport classes for this release. Project: http://git-wip-us.apache.org/repos/asf/activemq-cpp/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq-cpp/commit/e237191d Tree: http://git-wip-us.apache.org/repos/asf/activemq-cpp/tree/e237191d Diff: http://git-wip-us.apache.org/repos/asf/activemq-cpp/diff/e237191d
Branch: refs/heads/3.9.x Commit: e237191dc706dbc7b24df7096b0c68e9a14ac492 Parents: 2ad5237 Author: Timothy Bish <[email protected]> Authored: Thu Aug 6 17:26:46 2015 -0400 Committer: Timothy Bish <[email protected]> Committed: Thu Aug 6 17:26:46 2015 -0400 ---------------------------------------------------------------------- activemq-cpp/src/main/Makefile.am | 22 - .../src/main/activemq/library/ActiveMQCPP.cpp | 13 - .../discovery/AbstractDiscoveryAgent.cpp | 547 ------------------- .../discovery/AbstractDiscoveryAgent.h | 272 --------- .../discovery/AbstractDiscoveryAgentFactory.cpp | 82 --- .../discovery/AbstractDiscoveryAgentFactory.h | 89 --- .../discovery/DiscoveredBrokerData.cpp | 47 -- .../transport/discovery/DiscoveredBrokerData.h | 126 ----- .../transport/discovery/DiscoveryAgent.cpp | 26 - .../transport/discovery/DiscoveryAgent.h | 95 ---- .../discovery/DiscoveryAgentFactory.cpp | 26 - .../transport/discovery/DiscoveryAgentFactory.h | 61 --- .../discovery/DiscoveryAgentRegistry.cpp | 114 ---- .../discovery/DiscoveryAgentRegistry.h | 130 ----- .../transport/discovery/DiscoveryListener.cpp | 26 - .../transport/discovery/DiscoveryListener.h | 54 -- .../transport/discovery/DiscoveryTransport.cpp | 232 -------- .../transport/discovery/DiscoveryTransport.h | 107 ---- .../discovery/DiscoveryTransportFactory.cpp | 104 ---- .../discovery/DiscoveryTransportFactory.h | 53 -- .../discovery/http/HttpDiscoveryAgent.cpp | 197 ------- .../discovery/http/HttpDiscoveryAgent.h | 117 ---- .../http/HttpDiscoveryAgentFactory.cpp | 51 -- .../discovery/http/HttpDiscoveryAgentFactory.h | 47 -- activemq-cpp/src/test/Makefile.am | 8 - .../AbstractDiscoveryAgentFactoryTest.cpp | 112 ---- .../AbstractDiscoveryAgentFactoryTest.h | 45 -- .../discovery/AbstractDiscoveryAgentTest.cpp | 145 ----- .../discovery/AbstractDiscoveryAgentTest.h | 45 -- .../discovery/DiscoveryAgentRegistryTest.cpp | 90 --- .../discovery/DiscoveryAgentRegistryTest.h | 45 -- .../discovery/DiscoveryTransportFactoryTest.cpp | 116 ---- .../discovery/DiscoveryTransportFactoryTest.h | 47 -- activemq-cpp/src/test/testRegistry.cpp | 9 - 34 files changed, 3300 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/Makefile.am ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/Makefile.am b/activemq-cpp/src/main/Makefile.am index 98ec152..66d9f8b 100644 --- a/activemq-cpp/src/main/Makefile.am +++ b/activemq-cpp/src/main/Makefile.am @@ -155,17 +155,6 @@ cc_sources = \ activemq/transport/TransportFilter.cpp \ activemq/transport/TransportRegistry.cpp \ activemq/transport/correlator/ResponseCorrelator.cpp \ - activemq/transport/discovery/AbstractDiscoveryAgent.cpp \ - activemq/transport/discovery/AbstractDiscoveryAgentFactory.cpp \ - activemq/transport/discovery/DiscoveredBrokerData.cpp \ - activemq/transport/discovery/DiscoveryAgent.cpp \ - activemq/transport/discovery/DiscoveryAgentFactory.cpp \ - activemq/transport/discovery/DiscoveryAgentRegistry.cpp \ - activemq/transport/discovery/DiscoveryListener.cpp \ - activemq/transport/discovery/DiscoveryTransport.cpp \ - activemq/transport/discovery/DiscoveryTransportFactory.cpp \ - activemq/transport/discovery/http/HttpDiscoveryAgent.cpp \ - activemq/transport/discovery/http/HttpDiscoveryAgentFactory.cpp \ activemq/transport/failover/BackupTransport.cpp \ activemq/transport/failover/BackupTransportPool.cpp \ activemq/transport/failover/CloseTransportsTask.cpp \ @@ -836,17 +825,6 @@ h_sources = \ activemq/transport/TransportListener.h \ activemq/transport/TransportRegistry.h \ activemq/transport/correlator/ResponseCorrelator.h \ - activemq/transport/discovery/AbstractDiscoveryAgent.h \ - activemq/transport/discovery/AbstractDiscoveryAgentFactory.h \ - activemq/transport/discovery/DiscoveredBrokerData.h \ - activemq/transport/discovery/DiscoveryAgent.h \ - activemq/transport/discovery/DiscoveryAgentFactory.h \ - activemq/transport/discovery/DiscoveryAgentRegistry.h \ - activemq/transport/discovery/DiscoveryListener.h \ - activemq/transport/discovery/DiscoveryTransport.h \ - activemq/transport/discovery/DiscoveryTransportFactory.h \ - activemq/transport/discovery/http/HttpDiscoveryAgent.h \ - activemq/transport/discovery/http/HttpDiscoveryAgentFactory.h \ activemq/transport/failover/BackupTransport.h \ activemq/transport/failover/BackupTransportPool.h \ activemq/transport/failover/CloseTransportsTask.h \ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/library/ActiveMQCPP.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/library/ActiveMQCPP.cpp b/activemq-cpp/src/main/activemq/library/ActiveMQCPP.cpp index 376e6f4..daa63fe 100644 --- a/activemq-cpp/src/main/activemq/library/ActiveMQCPP.cpp +++ b/activemq-cpp/src/main/activemq/library/ActiveMQCPP.cpp @@ -20,7 +20,6 @@ #include <decaf/lang/Runtime.h> #include <activemq/wireformat/WireFormatRegistry.h> #include <activemq/transport/TransportRegistry.h> -#include <activemq/transport/discovery/DiscoveryAgentRegistry.h> #include <activemq/util/IdGenerator.h> @@ -31,9 +30,6 @@ #include <activemq/transport/tcp/TcpTransportFactory.h> #include <activemq/transport/tcp/SslTransportFactory.h> #include <activemq/transport/failover/FailoverTransportFactory.h> -#include <activemq/transport/discovery/DiscoveryTransportFactory.h> - -#include <activemq/transport/discovery/http/HttpDiscoveryAgentFactory.h> using namespace activemq; using namespace activemq::library; @@ -42,8 +38,6 @@ using namespace activemq::transport; using namespace activemq::transport::tcp; using namespace activemq::transport::mock; using namespace activemq::transport::failover; -using namespace activemq::transport::discovery; -using namespace activemq::transport::discovery::http; using namespace activemq::wireformat; //////////////////////////////////////////////////////////////////////////////// @@ -85,7 +79,6 @@ void ActiveMQCPP::shutdownLibrary() { WireFormatRegistry::shutdown(); TransportRegistry::shutdown(); - DiscoveryAgentRegistry::shutdown(); // Now it should be safe to shutdown Decaf. decaf::lang::Runtime::shutdownRuntime(); @@ -114,10 +107,4 @@ void ActiveMQCPP::registerTransports() { TransportRegistry::getInstance().registerFactory("nio+ssl", new SslTransportFactory()); TransportRegistry::getInstance().registerFactory("mock", new MockTransportFactory()); TransportRegistry::getInstance().registerFactory("failover", new FailoverTransportFactory()); - TransportRegistry::getInstance().registerFactory("discovery", new DiscoveryTransportFactory()); - - // Each discovery agent implemented in this library must be registered here. - DiscoveryAgentRegistry::initialize(); - - DiscoveryAgentRegistry::getInstance().registerFactory("http", new HttpDiscoveryAgentFactory); } http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.cpp b/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.cpp deleted file mode 100644 index f933ff8..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.cpp +++ /dev/null @@ -1,547 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <activemq/transport/discovery/AbstractDiscoveryAgent.h> - -#include <activemq/transport/discovery/DiscoveredBrokerData.h> -#include <activemq/transport/discovery/DiscoveryListener.h> - -#include <decaf/net/URI.h> -#include <decaf/lang/Thread.h> -#include <decaf/util/HashMap.h> -#include <decaf/util/concurrent/Mutex.h> -#include <decaf/util/concurrent/atomic/AtomicBoolean.h> -#include <decaf/util/concurrent/ThreadPoolExecutor.h> -#include <decaf/util/concurrent/TimeUnit.h> -#include <decaf/util/concurrent/LinkedBlockingQueue.h> - -using namespace activemq; -using namespace activemq::commands; -using namespace activemq::transport; -using namespace activemq::transport::discovery; -using namespace decaf; -using namespace decaf::lang; -using namespace decaf::lang::exceptions; -using namespace decaf::net; -using namespace decaf::util; -using namespace decaf::util::concurrent; -using namespace decaf::util::concurrent::atomic; - -//////////////////////////////////////////////////////////////////////////////// -const int AbstractDiscoveryAgent::DEFAULT_INITIAL_RECONNECT_DELAY = 5000; -const int AbstractDiscoveryAgent::DEFAULT_BACKOFF_MULTIPLIER = 2; -const int AbstractDiscoveryAgent::DEFAULT_MAX_RECONNECT_DELAY = 30000; -const int AbstractDiscoveryAgent::WORKER_KILL_TIME_SECONDS = 1000; -const int AbstractDiscoveryAgent::HEARTBEAT_MISS_BEFORE_DEATH = 10; -const int AbstractDiscoveryAgent::DEFAULT_KEEPALIVE_INTERVAL = 500; - -//////////////////////////////////////////////////////////////////////////////// -namespace activemq { -namespace transport { -namespace discovery { - - class AbstractDiscoveryAgentImpl { - private: - - AbstractDiscoveryAgentImpl(const AbstractDiscoveryAgentImpl&); - AbstractDiscoveryAgentImpl& operator=(const AbstractDiscoveryAgentImpl&); - - public: - - long long initialReconnectDelay; - long long maxReconnectDelay; - long long backOffMultiplier; - bool useExponentialBackOff; - int maxReconnectAttempts; - long long keepAliveInterval; - - AtomicBoolean started; - Pointer<Thread> worker; - Pointer<ThreadPoolExecutor> executor; - - HashMap<std::string, Pointer<DiscoveredBrokerData> > discoveredServices; - Mutex discoveredServicesLock; - - URI discoveryUri; - std::string selfService; - std::string group; - DiscoveryListener* listener; - long long lastAdvertizeTime; - bool reportAdvertizeFailed; - - public: - - AbstractDiscoveryAgentImpl() : initialReconnectDelay(AbstractDiscoveryAgent::DEFAULT_INITIAL_RECONNECT_DELAY), - maxReconnectDelay(AbstractDiscoveryAgent::DEFAULT_MAX_RECONNECT_DELAY), - backOffMultiplier(AbstractDiscoveryAgent::DEFAULT_BACKOFF_MULTIPLIER), - useExponentialBackOff(false), - maxReconnectAttempts(0), - keepAliveInterval(AbstractDiscoveryAgent::DEFAULT_KEEPALIVE_INTERVAL), - started(), - worker(), - executor(), - discoveredServices(), - discoveredServicesLock(), - discoveryUri(), - selfService(), - group("default"), - listener(), - lastAdvertizeTime(0), - reportAdvertizeFailed(true) - {} - - ~AbstractDiscoveryAgentImpl() { - if (started.compareAndSet(true, false)) { - if (worker == NULL) { - worker->join(5000); - - if (!worker->isAlive()) { - worker->interrupt(); - worker->join(1000); - } - - worker.reset(NULL); - } - - executor->shutdown(); - executor->awaitTermination(1, TimeUnit::MINUTES); - } - - } - - Executor& getExecutor() { - if (executor == NULL) { - synchronized(&discoveredServicesLock) { - if (executor == NULL) { - executor.reset( - new ThreadPoolExecutor(1, 1, 45, TimeUnit::SECONDS, - new LinkedBlockingQueue<Runnable*>())); - } - } - } - return *executor; - } - - /** - * Returns true if this Broker has been marked as failed and it is now time to - * start a recovery attempt. - */ - bool isTimeForRecovery(Pointer<DiscoveredBrokerData> service) { - synchronized(&discoveredServicesLock) { - - if (!service->isFailed()) { - return false; - } - - int maxReconnectAttempts = this->maxReconnectAttempts; - - // Are we done trying to recover this guy? - if (maxReconnectAttempts > 0 && service->getFailureCount() > maxReconnectAttempts) { - return false; - } - - // Is it not yet time? - if (System::currentTimeMillis() < service->getNextRecoveryTime()) { - return false; - } - - service->setFailed(false); - return true; - } - - return false; - } - - void updateHeartBeat(Pointer<DiscoveredBrokerData> service) { - synchronized(&discoveredServicesLock) { - - service->setLastHeartBeatTime(System::currentTimeMillis()); - - // Consider that the broker recovery has succeeded if it has not failed in 60 seconds. - if (!service->isFailed() && service->getFailureCount() > 0 && - (service->getLastHeartBeatTime() - service->getNextRecoveryTime()) > TimeUnit::MINUTES.toSeconds(60)) { - - service->setFailureCount(0); - service->setNextRecoveryTime(System::currentTimeMillis()); - } - } - } - - bool markFailed(Pointer<DiscoveredBrokerData> service) { - synchronized(&discoveredServicesLock) { - - if (!service->isFailed()) { - service->setFailed(true); - service->setFailureCount(service->getFailureCount() + 1); - - long long reconnectDelay = 0; - if (!useExponentialBackOff) { - reconnectDelay = initialReconnectDelay; - } else { - reconnectDelay = (long) Math::pow((double)backOffMultiplier, (double)service->getFailureCount()); - reconnectDelay = Math::min(reconnectDelay, maxReconnectDelay); - } - - service->setNextRecoveryTime(System::currentTimeMillis() + reconnectDelay); - return true; - } - } - return false; - } - }; - - class ServiceAddedRunnable : public Runnable { - private: - - AbstractDiscoveryAgent* agent; - Pointer<DiscoveredBrokerData> event; - - public: - - ServiceAddedRunnable(AbstractDiscoveryAgent* agent, Pointer<DiscoveredBrokerData> event) : - Runnable(), agent(agent), event(event) { - } - virtual ~ServiceAddedRunnable() {} - - virtual void run() { - DiscoveryListener* listener = agent->getDiscoveryListener(); - if (listener != NULL) { - listener->onServiceAdd(event.get()); - } - } - }; - - class ServiceRemovedRunnable : public Runnable { - private: - - AbstractDiscoveryAgent* agent; - Pointer<DiscoveredBrokerData> event; - - public: - - ServiceRemovedRunnable(AbstractDiscoveryAgent* agent, Pointer<DiscoveredBrokerData> event) : - Runnable(), agent(agent), event(event) {} - virtual ~ServiceRemovedRunnable() {} - - virtual void run() { - DiscoveryListener* listener = agent->getDiscoveryListener(); - if (listener != NULL) { - listener->onServiceRemove(event.get()); - } - } - }; - -}}} - -//////////////////////////////////////////////////////////////////////////////// -AbstractDiscoveryAgent::AbstractDiscoveryAgent() : DiscoveryAgent(), impl(new AbstractDiscoveryAgentImpl) { -} - -//////////////////////////////////////////////////////////////////////////////// -AbstractDiscoveryAgent::~AbstractDiscoveryAgent() { - try { - delete this->impl; - } - DECAF_CATCHALL_NOTHROW() -} - -//////////////////////////////////////////////////////////////////////////////// -bool AbstractDiscoveryAgent::isStarted() const { - return impl->started.get(); -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::start() { - if (impl->started.compareAndSet(false, true)) { - doStart(); - - if (impl->worker == NULL) { - impl->worker.reset(new Thread(this)); - impl->worker->start(); - } - - doAdvertizeSelf(); - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::stop() { - // Changing the isStarted flag will signal the thread that it needs to shut down. - if (impl->started.compareAndSet(true, false)) { - doStop(); - - if (impl->worker == NULL) { - impl->worker->join(WORKER_KILL_TIME_SECONDS); - - if (!impl->worker->isAlive()) { - impl->worker->interrupt(); - impl->worker->join(WORKER_KILL_TIME_SECONDS); - } - - impl->worker.reset(NULL); - } - - impl->executor->shutdown(); - impl->executor->awaitTermination(1, TimeUnit::MINUTES); - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::run() { - - Thread::currentThread()->setName("Discovery Agent Thread."); - - while (impl->started.get()) { - doTimeKeepingServices(); - try { - doDiscovery(); - } catch (InterruptedException& ex) { - return; - } catch (Exception& ignore) { - } - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::registerService(const std::string& name) { - impl->selfService = name; - if (impl->started.get()) { - try { - doAdvertizeSelf(); - } catch (Exception& e) { - // If a the advertise fails, chances are all subsequent sends will fail - // too.. No need to keep reporting the same error over and over. - if (impl->reportAdvertizeFailed) { - impl->reportAdvertizeFailed = false; - } - } - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::serviceFailed(const activemq::commands::DiscoveryEvent& event) { - - Pointer<DiscoveredBrokerData> service; - synchronized(&impl->discoveredServicesLock) { - try { - service = impl->discoveredServices.get(event.getServiceName()); - } catch (NoSuchElementException& ex) {} - } - - if (service != NULL && impl->markFailed(service)) { - fireServiceRemovedEvent(service); - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setDiscoveryListener(DiscoveryListener* listener) { - this->impl->listener = listener; -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryListener* AbstractDiscoveryAgent::getDiscoveryListener() const { - return this->impl->listener; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setDiscoveryURI(const URI& discoveryURI) { - impl->discoveryUri = discoveryURI; -} - -//////////////////////////////////////////////////////////////////////////////// -URI AbstractDiscoveryAgent::getDiscoveryURI() const { - return impl->discoveryUri; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setServiceName(const std::string& name) { - impl->selfService = name; -} - -//////////////////////////////////////////////////////////////////////////////// -std::string AbstractDiscoveryAgent::getServiceName() const { - return impl->selfService; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setKeepAliveInterval(long long interval) { - impl->keepAliveInterval = interval; -} - -//////////////////////////////////////////////////////////////////////////////// -long long AbstractDiscoveryAgent::getKeepAliveInterval() const { - return impl->keepAliveInterval; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setInitialReconnectDelay(long long initialReconnectDelay) { - impl->initialReconnectDelay = initialReconnectDelay; -} - -//////////////////////////////////////////////////////////////////////////////// -long long AbstractDiscoveryAgent::getInitialReconnectDelay() const { - return impl->initialReconnectDelay; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setMaxReconnectAttempts(int maxReconnectAttempts) { - impl->maxReconnectAttempts = maxReconnectAttempts; -} - -//////////////////////////////////////////////////////////////////////////////// -int AbstractDiscoveryAgent::getMaxReconnectAttempts() const { - return impl->maxReconnectAttempts; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setMaxReconnectDelay(long long maxReconnectDelay) { - impl->maxReconnectDelay = maxReconnectDelay; -} - -//////////////////////////////////////////////////////////////////////////////// -long long AbstractDiscoveryAgent::getMaxReconnectDelay() const { - return impl->maxReconnectDelay; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setUseExponentialBackOff(bool useExponentialBackOff) { - impl->useExponentialBackOff = useExponentialBackOff; -} - -//////////////////////////////////////////////////////////////////////////////// -bool AbstractDiscoveryAgent::isUseExponentialBackOff() const { - return impl->useExponentialBackOff; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setBackOffMultiplier(long long multiplier) { - impl->backOffMultiplier = multiplier; -} - -//////////////////////////////////////////////////////////////////////////////// -long long AbstractDiscoveryAgent::getBackOffMultiplier() const { - return impl->backOffMultiplier; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::setGroup(const std::string& group) { - impl->group = group; -} - -//////////////////////////////////////////////////////////////////////////////// -std::string AbstractDiscoveryAgent::getGroup() const { - return impl->group; -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::fireServiceRemovedEvent(Pointer<DiscoveredBrokerData> event) { - if (impl->listener != NULL && impl->started.get()) { - // Have the listener process the event async so that - // he does not block this thread since we are doing time sensitive - // processing of events. - impl->getExecutor().execute(new ServiceRemovedRunnable(this, event)); - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::fireServiceAddedEvent(Pointer<DiscoveredBrokerData> event) { - if (impl->listener != NULL && impl->started.get()) { - // Have the listener process the event async so that - // he does not block this thread since we are doing time sensitive - // processing of events. - impl->getExecutor().execute(new ServiceAddedRunnable(this, event)); - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::processLiveService(const std::string& brokerName, const std::string& service) { - - if (getServiceName().empty() || service != getServiceName()) { - Pointer<DiscoveredBrokerData> remoteBroker; - synchronized(&impl->discoveredServicesLock) { - try { - remoteBroker = impl->discoveredServices.get(service); - } catch (NoSuchElementException& ignored) { - } - } - - if (remoteBroker == NULL) { - remoteBroker.reset(new DiscoveredBrokerData(brokerName, service)); - impl->discoveredServices.put(service, remoteBroker); - fireServiceAddedEvent(remoteBroker); - doAdvertizeSelf(); - } else { - impl->updateHeartBeat(remoteBroker); - if (impl->isTimeForRecovery(remoteBroker)) { - fireServiceAddedEvent(remoteBroker); - } - } - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::processDeadService(const std::string& service) { - - if (service != getServiceName()) { - - Pointer<DiscoveredBrokerData> remoteBroker; - synchronized(&impl->discoveredServicesLock) { - try { - remoteBroker = impl->discoveredServices.get(service); - } catch (NoSuchElementException& ignored) { - } - } - - if (remoteBroker != NULL && !remoteBroker->isFailed()) { - fireServiceRemovedEvent(remoteBroker); - } - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::doTimeKeepingServices() { - - if (impl->started.get()) { - long long currentTime = System::currentTimeMillis(); - if (currentTime < impl->lastAdvertizeTime || - ((currentTime - impl->keepAliveInterval) > impl->lastAdvertizeTime)) { - - doAdvertizeSelf(); - impl->lastAdvertizeTime = currentTime; - } - doExpireOldServices(); - } -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgent::doExpireOldServices() { - long long expireTime = System::currentTimeMillis() - - (impl->keepAliveInterval * HEARTBEAT_MISS_BEFORE_DEATH); - - std::vector< Pointer<DiscoveredBrokerData> > services; - synchronized(&impl->discoveredServicesLock) { - services = impl->discoveredServices.values().toArray(); - } - - std::vector< Pointer<DiscoveredBrokerData> >::iterator iter = services.begin(); - for (; iter != services.end(); ++iter) { - Pointer<DiscoveredBrokerData> service = *iter; - if (service->getLastHeartBeatTime() < expireTime) { - processDeadService(service->getServiceName()); - } - } -} http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.h ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.h b/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.h deleted file mode 100644 index 098f0aa..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgent.h +++ /dev/null @@ -1,272 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef _ACTIVEMQ_TRANSPORT_DISCOVERY_ABSTRACTDISCOVERYAGENT_H_ -#define _ACTIVEMQ_TRANSPORT_DISCOVERY_ABSTRACTDISCOVERYAGENT_H_ - -#include <activemq/util/Config.h> - -#include <activemq/transport/discovery/DiscoveryAgent.h> -#include <activemq/transport/discovery/DiscoveredBrokerData.h> - -#include <decaf/lang/Runnable.h> -#include <decaf/lang/Pointer.h> - -namespace activemq { -namespace transport { -namespace discovery { - - class AbstractDiscoveryAgentImpl; - - /** - * Abstract base class that provides all the basic implementation needed to create - * a DiscoveryAgent instance without needing to implement a lot of boilerplate code. - * - * @since 3.9.0 - */ - class AMQCPP_API AbstractDiscoveryAgent : public DiscoveryAgent, public decaf::lang::Runnable { - private: - - AbstractDiscoveryAgentImpl* impl; - - public: - - static const int DEFAULT_INITIAL_RECONNECT_DELAY; - static const int DEFAULT_BACKOFF_MULTIPLIER; - static const int DEFAULT_MAX_RECONNECT_DELAY; - static const int WORKER_KILL_TIME_SECONDS; - static const int HEARTBEAT_MISS_BEFORE_DEATH; - static const int DEFAULT_KEEPALIVE_INTERVAL; - - private: - - AbstractDiscoveryAgent(const AbstractDiscoveryAgent&); - AbstractDiscoveryAgent& operator= (const AbstractDiscoveryAgent&); - - public: - - AbstractDiscoveryAgent(); - virtual ~AbstractDiscoveryAgent(); - - virtual void start(); - virtual void stop(); - - virtual void registerService(const std::string& name); - virtual void serviceFailed(const activemq::commands::DiscoveryEvent& event); - - virtual void setDiscoveryListener(DiscoveryListener* listener); - virtual DiscoveryListener* getDiscoveryListener() const; - - virtual void setDiscoveryURI(const decaf::net::URI& discoveryURI); - virtual decaf::net::URI getDiscoveryURI() const; - - /** - * @return true if this agent is currently started. - */ - bool isStarted() const; - - /** - * Sets the service that is publish by this agent if it supports publishing. - * - * @param name - * The service name to publish, typically the URI. - */ - void setServiceName(const std::string& name); - - /** - * Gets the configured service to publish, not all agents can publish so this value - * may not mean that an actual service advertisement is ever done. - * - * @return the configured service to publish. - */ - std::string getServiceName() const; - - /** - * Sets the keep alive interval used to control how long an service that has not been - * seen is kept in the list of discovered services before being idle to long. Also this - * value controls how often this service will advertise itself if it supports that. - * - * @param interval - * Time in milliseconds for the keep alive interval. - */ - void setKeepAliveInterval(long long interval); - - /** - * Gets the keep alive interval used to control how long an service that has not been - * seen is kept in the list of discovered services before being idle to long. Also this - * value controls how often this service will advertise itself if it supports that. - * - * @return Time in milliseconds for the keep alive interval. - */ - long long getKeepAliveInterval() const; - - /** - * Sets the agents reconnect backoff multiplier. - * - * @param multiplier - * The back multiplier to use when calculating the next recovery time. - */ - void setBackOffMultiplier(long long multiplier); - - /** - * Gets the configured backoff multiplier for calculating the next recovery time. - * - * @return the configured backoff multiplier for calculating the next recovery time. - */ - long long getBackOffMultiplier() const; - - /** - * Sets the initial amount of time the agent should wait before attempt a reconnect on - * a discovered service. - * - * @param initialReconnectDelay - * The time to wait before the initial reconnect attempt in milliseconds. - */ - void setInitialReconnectDelay(long long initialReconnectDelay); - - /** - * Gets the initial amount of time the agent should wait before attempt a reconnect on - * a discovered service. - * - * @return The time to wait before the initial reconnect attempt in milliseconds. - */ - long long getInitialReconnectDelay() const; - - /** - * Sets the maximum number of reconnect attempts that occur before a service is considered to - * be failed and removed. - * - * @param maxReconnectAttempts - * The maximum number of reconnect attempts allowed or zero for no limit. - */ - void setMaxReconnectAttempts(int maxReconnectAttempts); - - /** - * Returns the maximum number of reconnect attempts that will be attempted before a - * service is considered to be failed. - * - * @return the maximum number of allowed reconnect attempts. - */ - int getMaxReconnectAttempts() const; - - /** - * The maximum time that a reconnect delay is allowed to grow to, in Milliseconds. - * - * @param maxReconnectDelay - * The maximum time in milliseconds to limit the reconnect delays. - */ - void setMaxReconnectDelay(long long maxReconnectDelay); - - /** - * Returns the maximum time allowed between reconnect attempts. This limits the amount - * of time that reconnect will wait when exponential backoff is enabled. - * - * @return the maximum reconnect delay in milliseconds. - */ - long long getMaxReconnectDelay() const; - - /** - * Sets whether reconnect attempts are delayed exponentially until a new connection is made - * or the maximum number of attempts is exceeded. Using an exponential back-off can - * reduce CPU overhead but can also lead to a longer time to reconnect to a service that - * is known to have short drop outs. - * - * @param useExponentialBackOff - * When true it indicates that the delay between reconnect attempts grows exponentially. - */ - void setUseExponentialBackOff(bool useExponentialBackOff); - - /** - * Returns true if successive attempts to reconnect to a discovered service are delayed - * by an exponentially growing time factor. - * - * @return whether exponential back-off is enabled. - */ - bool isUseExponentialBackOff() const; - - /** - * Sets the discovery group that this agent is assigned. The default value for this - * group is "default". - * - * @param group - * The assigned group for the discovery agent. - */ - void setGroup(const std::string& group); - - /** - * Returns the assigned group for this discovery agent. - * - * @return the assigned group for this discovery agent. - */ - std::string getGroup() const; - - protected: - - /** - * Default implementation of the DiscoveryAgent's background worker thread processing. - * - * In general there is no need for an agent to override this method, however some agents - * may require special handling of their discovery mechanisms so it is allowable for this - * method to be overridden. Implementors of this method must ensure that the run method - * will return quickly once the started value becomes false or the Thread is interrupted. - */ - virtual void run(); - - virtual void processLiveService(const std::string& brokerName, const std::string& service); - virtual void processDeadService(const std::string& service); - - virtual void fireServiceAddedEvent(decaf::lang::Pointer<DiscoveredBrokerData> event); - virtual void fireServiceRemovedEvent(decaf::lang::Pointer<DiscoveredBrokerData> event); - - protected: - - /** - * The real agent class must implement this method to perform any necessary resource allocation - * prior to the completion of the start call. - */ - virtual void doStart() = 0; - - /** - * The real agent class must implement this method to perform any necessary resource cleanup - * prior to the completion of the stop call. - */ - virtual void doStop() = 0; - - /** - * Optional method that allows the agent to perform an advertisement of this clients - * service. Not all agents can do this so this method should be a no-op if this is - * not supported. - */ - virtual void doAdvertizeSelf() = 0; - - /** - * Performs the actual discovery operation for this agent. This method is called in the - * worker thread context of the AbstractDiscoveryAgent and should not block forever. The - * method will be called repeatedly to allow the agent to continually probe for new services. - */ - virtual void doDiscovery() = 0; - - private: - - void doExpireOldServices(); - void doTimeKeepingServices(); - - }; - -}}} - -#endif /* _ACTIVEMQ_TRANSPORT_DISCOVERY_ABSTRACTDISCOVERYAGENT_H_ */ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.cpp b/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.cpp deleted file mode 100644 index 6d9e33b..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.cpp +++ /dev/null @@ -1,82 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <activemq/transport/discovery/AbstractDiscoveryAgentFactory.h> - -#include <activemq/util/URISupport.h> -#include <activemq/exceptions/ActiveMQException.h> - -#include <decaf/lang/Boolean.h> -#include <decaf/lang/Integer.h> -#include <decaf/lang/Long.h> - -using namespace decaf; -using namespace decaf::net; -using namespace decaf::util; -using namespace decaf::lang; -using namespace activemq; -using namespace activemq::util; -using namespace activemq::exceptions; -using namespace activemq::transport; -using namespace activemq::transport::discovery; - -//////////////////////////////////////////////////////////////////////////////// -AbstractDiscoveryAgentFactory::~AbstractDiscoveryAgentFactory() { -} - -//////////////////////////////////////////////////////////////////////////////// -Pointer<DiscoveryAgent> AbstractDiscoveryAgentFactory::createAgent(const URI& agentURI) { - - try { - - Pointer<AbstractDiscoveryAgent> agent = this->doCreateAgent(); - - agent->setDiscoveryURI(agentURI); - Properties options = URISupport::parseParameters(agentURI); - doConfigureAgent(agent, options); - - return agent; - } - AMQ_CATCH_RETHROW(ActiveMQException) - AMQ_CATCH_EXCEPTION_CONVERT(Exception, ActiveMQException) - AMQ_CATCHALL_THROW(ActiveMQException) -} - -//////////////////////////////////////////////////////////////////////////////// -void AbstractDiscoveryAgentFactory::doConfigureAgent(Pointer<AbstractDiscoveryAgent> agent, const Properties& options) { - - try { - - agent->setKeepAliveInterval( - Long::parseLong(options.getProperty("keepAliveInterval", "500"))); - agent->setMaxReconnectDelay( - Long::parseLong(options.getProperty("maxReconnectDelay", "30000"))); - agent->setUseExponentialBackOff( - Boolean::parseBoolean(options.getProperty("useExponentialBackOff", "true"))); - agent->setBackOffMultiplier( - Long::parseLong(options.getProperty("backOffMultiplier", "2"))); - agent->setMaxReconnectAttempts( - Integer::parseInt(options.getProperty("maxReconnectAttempts", "-1"))); - agent->setInitialReconnectDelay( - Long::parseLong(options.getProperty("initialReconnectDelay", "10"))); - agent->setGroup(options.getProperty("group", "default")); - agent->setGroup(options.getProperty("service", "")); - } - AMQ_CATCH_RETHROW(ActiveMQException) - AMQ_CATCH_EXCEPTION_CONVERT(Exception, ActiveMQException) - AMQ_CATCHALL_THROW(ActiveMQException) -} http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.h ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.h b/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.h deleted file mode 100644 index 76dfa4a..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/AbstractDiscoveryAgentFactory.h +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef _ACTIVEMQ_TRANSPORT_DISCOVERY_ABSTRACTDISCOVERYAGENTFACTORY_H_ -#define _ACTIVEMQ_TRANSPORT_DISCOVERY_ABSTRACTDISCOVERYAGENTFACTORY_H_ - -#include <activemq/util/Config.h> - -#include <decaf/lang/Pointer.h> -#include <decaf/util/Properties.h> -#include <decaf/net/URI.h> - -#include <activemq/transport/discovery/DiscoveryAgentFactory.h> -#include <activemq/transport/discovery/AbstractDiscoveryAgent.h> - -namespace activemq { -namespace transport { -namespace discovery { - - class AMQCPP_API AbstractDiscoveryAgentFactory : public DiscoveryAgentFactory { - public: - - virtual ~AbstractDiscoveryAgentFactory(); - - /** - * Provides the implementation of the createAgent method, subclasses should implement the - * doCreateAgent method to do the actual agent creation and override the doConfigureAgent if - * they have additional URI options that need to be applied before returning the newly - * created agent instance. - * - * @param agentURI - * The URI that describes the agent and any requested configuration changes. - * - * @return a new DiscoveryAgent instance with all URI options applied. - */ - virtual decaf::lang::Pointer<DiscoveryAgent> createAgent(const decaf::net::URI& agentURI); - - protected: - - /** - * Subclasses are required to implement this method and return the correct Agent instance. - * - * After this method is called by createAgent the configuration method will be invoked to - * allow the URI options to be applied. - * - * @return a Pointer to the newly created discovery agent instance. - * - * @throws ActiveMQException if an error occurs while creating the agent. - */ - virtual decaf::lang::Pointer<AbstractDiscoveryAgent> doCreateAgent() = 0; - - protected: - - /** - * Called from createAgent after the new agent instance has been created. The default - * implementation of this method will configure all the known AbstractDiscoveryAgent - * options using the URI options given. Subclasses can override this method to apply - * addition URI options specific to the agent type being created but should also call - * this method to ensure that all options are applied. - * - * @param agent - * The AbstractDiscoveryAgent instance that is to be configured. - * @param options - * The Properties object that contians all agent options parsed from the URI. - * - * @throws ActiveMQException if an error occurs while applying the options. - */ - virtual void doConfigureAgent(decaf::lang::Pointer<AbstractDiscoveryAgent> agent, - const decaf::util::Properties& options); - - }; - -}}} - -#endif /* _ACTIVEMQ_TRANSPORT_DISCOVERY_ABSTRACTDISCOVERYAGENTFACTORY_H_ */ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.cpp b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.cpp deleted file mode 100644 index 8657760..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.cpp +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "DiscoveredBrokerData.h" - -using namespace activemq; -using namespace activemq::commands; -using namespace activemq::transport; -using namespace activemq::transport::discovery; - -//////////////////////////////////////////////////////////////////////////////// -DiscoveredBrokerData::DiscoveredBrokerData() : DiscoveryEvent(), - lastHeartBeatTime(0), - nextRecoveryTime(0), - failureCount(0), - failed(false) { -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveredBrokerData::DiscoveredBrokerData(const std::string& brokerName, const std::string& service) : - DiscoveryEvent(), - lastHeartBeatTime(0), - nextRecoveryTime(0), - failureCount(0), - failed(false) { - - setBrokerName(brokerName); - setServiceName(service); -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveredBrokerData::~DiscoveredBrokerData() { -} http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.h ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.h b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.h deleted file mode 100644 index 6aaf6e7..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveredBrokerData.h +++ /dev/null @@ -1,126 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVEREDBROKERDATA_H_ -#define _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVEREDBROKERDATA_H_ - -#include <activemq/util/Config.h> -#include <activemq/commands/DiscoveryEvent.h> - -namespace activemq { -namespace transport { -namespace discovery { - - /** - * Enhanced DiscoveryEvent object used to store additional data about discovered - * broker services. - * - * @since 3.9.0 - */ - class AMQCPP_API DiscoveredBrokerData : public activemq::commands::DiscoveryEvent { - private: - - long long lastHeartBeatTime; - long long nextRecoveryTime; - int failureCount; - bool failed; - - public: - - DiscoveredBrokerData(); - DiscoveredBrokerData(const std::string& brokerName, const std::string& service); - - virtual ~DiscoveredBrokerData(); - - /** - * Gets the time of the last heart best from this Broker - * - * @return the time of the last received heart beat event from this Broker - */ - long long getLastHeartBeatTime() const { - return this->lastHeartBeatTime; - } - - /** - * Sets the time of the last received heart beat event from this Broker. - * - * @param lastHeartBeatTime - * Time since last heart beat was received. - */ - void setLastHeartBeatTime(long long lastHeartBeatTime) { - this->lastHeartBeatTime = lastHeartBeatTime; - } - - /** - * @return true is this service has been marked as failed. - */ - bool isFailed() const { - return failed; - } - - /** - * Marks this service as failed. - * - * @param failed - * Set to true to mark this broker as failed. - */ - void setFailed(bool failed) { - this->failed = failed; - } - - /** - * Gets the number of times that there was a failure contacting this broker. - * - * @return count of the number of failures of this service. - */ - int getFailureCount() const { - return failureCount; - } - - /** - * Sets the number of failures that are recorded for this service. - * - * @param failureCount - * The new value of the failure count for this service. - */ - void setFailureCount(int failureCount) { - this->failureCount = failureCount; - } - - /** - * Gets the set time for the next recovery attempt on this service. - * - * @return the next set time that this service can have a recovery attempt. - */ - long long getNextRecoveryTime() const { - return nextRecoveryTime; - } - - /** - * Sets the next recovery time value for this service. - * - * @param recoveryTime - * The next time a recovery of this service should be attempted. - */ - void setNextRecoveryTime(long long nextRecoveryTime) { - this->nextRecoveryTime = nextRecoveryTime; - } - }; - -}}} - -#endif /* _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVEREDBROKERDATA_H_ */ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.cpp b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.cpp deleted file mode 100644 index d356690..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.cpp +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "DiscoveryAgent.h" - -using namespace activemq; -using namespace activemq::transport; -using namespace activemq::transport::discovery; - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryAgent::~DiscoveryAgent() { -} http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.h ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.h b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.h deleted file mode 100644 index db1165d..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgent.h +++ /dev/null @@ -1,95 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENT_H_ -#define _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENT_H_ - -#include <activemq/util/Config.h> -#include <activemq/util/Service.h> -#include <decaf/net/URI.h> -#include <decaf/util/Properties.h> - -#include <activemq/commands/DiscoveryEvent.h> - -namespace activemq { -namespace transport { -namespace discovery { - - class DiscoveryListener; - - class AMQCPP_API DiscoveryAgent : public activemq::util::Service { - public: - - virtual ~DiscoveryAgent(); - - /** - * Sets the URI that was used to create this discovery agent. This URI can - * be used to configure the discovery agent. - */ - virtual void setDiscoveryURI(const decaf::net::URI& discoveryURI) = 0; - - /** - * Sets the URI that was used to create this discovery agent. This URI can - * be used to configure the discovery agent. - */ - virtual decaf::net::URI getDiscoveryURI() const = 0; - - /** - * Sets the discovery listener which will be notified on the add or remove of - * a discovered service. - * - * @param listener - * Pointer to a listener instance that will be notified, or null to reset. - */ - virtual void setDiscoveryListener(DiscoveryListener* listener) = 0; - - /** - * Register a service with this DiscoveryAgent. If the agent supports advertisement - * then the new service will be broadcast. - * - * @param service - * The service to register and advertise. - * - * @throws IOException if an error occurs. - */ - virtual void registerService(const std::string& name) = 0; - - /** - * A process actively using a service may see it go down before the DiscoveryAgent - * notices the service's failure. That process can use this method to notify the - * DiscoveryAgent of the failure so that other listeners of this DiscoveryAgent can - * also be made aware of the failure. - * - * @paran event - * A DiscoveryEvent that contains information on the failed service. - * - * @throws IOException if an error occurs processing the failure event. - */ - virtual void serviceFailed(const activemq::commands::DiscoveryEvent& event) = 0; - - /** - * Returns a descriptive string that represents this discovery agent. - * - * @return a string that descibes this discovery agent. - */ - virtual std::string toString() const = 0; - - }; - -}}} - -#endif /* _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENT_H_ */ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.cpp b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.cpp deleted file mode 100644 index ccef7d5..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.cpp +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "DiscoveryAgentFactory.h" - -using namespace activemq; -using namespace activemq::transport; -using namespace activemq::transport::discovery; - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryAgentFactory::~DiscoveryAgentFactory() { -} http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.h ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.h b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.h deleted file mode 100644 index 302e476..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentFactory.h +++ /dev/null @@ -1,61 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENTFACTORY_H_ -#define _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENTFACTORY_H_ - -#include <activemq/util/Config.h> - -#include <decaf/lang/Pointer.h> -#include <decaf/net/URI.h> - -#include <activemq/transport/discovery/DiscoveryAgent.h> - -namespace activemq { -namespace transport { -namespace discovery { - - /** - * Factory class for creating discovery agents. All agents are required to - * provide a factory class that can be registered in the DiscoveryAgentRegistry. - * - * @since 3.9.0 - */ - class AMQCPP_API DiscoveryAgentFactory { - public: - - virtual ~DiscoveryAgentFactory(); - - /** - * Creates and returns a new DiscoveryAgentFactory instance that can be used to - * create the agent referred to in the given URI. The factory should apply all - * configuration options to the agent prior to returning it. - * - * @param agentURI - * The URI that defines the agent to create along with it configuration options. - * - * @return a new DiscoveryAgent instance for the given URI. - * - * @throws IOException if an error occurs creating the given agent. - */ - virtual decaf::lang::Pointer<DiscoveryAgent> createAgent(const decaf::net::URI& agentURI) = 0; - - }; - -}}} - -#endif /* _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENTFACTORY_H_ */ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.cpp b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.cpp deleted file mode 100644 index 17bb845..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.cpp +++ /dev/null @@ -1,114 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "DiscoveryAgentRegistry.h" - -#include <activemq/exceptions/ActiveMQException.h> -#include <activemq/transport/discovery/DiscoveryAgentFactory.h> - -using namespace std; -using namespace activemq; -using namespace activemq::exceptions; -using namespace activemq::transport; -using namespace activemq::transport::discovery; -using namespace decaf; -using namespace decaf::util; -using namespace decaf::lang; -using namespace decaf::lang::exceptions; - -//////////////////////////////////////////////////////////////////////////////// -namespace { - DiscoveryAgentRegistry* theOnlyInstance; -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryAgentRegistry::DiscoveryAgentRegistry() : registry() { -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryAgentRegistry::~DiscoveryAgentRegistry() { - try { - this->unregisterAllFactories(); - } - AMQ_CATCHALL_NOTHROW() -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryAgentFactory* DiscoveryAgentRegistry::findFactory(const std::string& name) const { - - if (!this->registry.containsKey(name)) { - throw NoSuchElementException(__FILE__, __LINE__, - "No Matching Factory Registered for format := %s", name.c_str()); - } - - return this->registry.get(name); -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryAgentRegistry::registerFactory(const std::string& name, DiscoveryAgentFactory* factory) { - - if (name == "") { - throw IllegalArgumentException(__FILE__, __LINE__, "DiscoveryAgentFactory name cannot be the empty string"); - } - - if (factory == NULL) { - throw NullPointerException(__FILE__, __LINE__, "Supplied DiscoveryAgentFactory pointer was NULL"); - } - - this->registry.put(name, factory); -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryAgentRegistry::unregisterFactory(const std::string& name) { - if (this->registry.containsKey(name)) { - delete this->registry.get(name); - this->registry.remove(name); - } -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryAgentRegistry::unregisterAllFactories() { - - Pointer<Iterator<DiscoveryAgentFactory*> > iterator(this->registry.values().iterator()); - while (iterator->hasNext()) { - delete iterator->next(); - } - - this->registry.clear(); -} - -//////////////////////////////////////////////////////////////////////////////// -std::vector<std::string> DiscoveryAgentRegistry::getAgentNames() const { - return this->registry.keySet().toArray(); -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryAgentRegistry& DiscoveryAgentRegistry::getInstance() { - return *theOnlyInstance; -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryAgentRegistry::initialize() { - theOnlyInstance = new DiscoveryAgentRegistry(); -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryAgentRegistry::shutdown() { - theOnlyInstance->unregisterAllFactories(); - delete theOnlyInstance; - theOnlyInstance = NULL; -} http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.h ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.h b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.h deleted file mode 100644 index bf680b7..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryAgentRegistry.h +++ /dev/null @@ -1,130 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENTREGISTRY_H_ -#define _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENTREGISTRY_H_ - -#include <activemq/util/Config.h> - -#include <string> -#include <vector> - -#include <decaf/util/StlMap.h> - -namespace activemq { -namespace library { - class ActiveMQCPP; -} -namespace transport { -namespace discovery { - - class DiscoveryAgentFactory; - - /** - * Registry of all Discovery Agent Factories that are available to the client - * at runtime. New Agents must have a factory registered here before an attempt - * to create a DiscoveryTansport which uses that agent. - * - * @since 3.9.0 - */ - class AMQCPP_API DiscoveryAgentRegistry { - private: - - decaf::util::StlMap<std::string, DiscoveryAgentFactory*> registry; - - private: - - DiscoveryAgentRegistry(); - DiscoveryAgentRegistry(const DiscoveryAgentRegistry& registry); - DiscoveryAgentRegistry& operator=(const DiscoveryAgentRegistry& registry); - - public: - - virtual ~DiscoveryAgentRegistry(); - - /** - * Gets a Registered DiscoveryAgentFactory from the Registry and returns it - * if there is not a registered format factory with the given name an exception - * is thrown. - * - * @param name - * The name of the Factory to find in the Registry. - * - * @return the Factory registered under the given name. - * - * @throws NoSuchElementException if no factory is registered with that name. - */ - DiscoveryAgentFactory* findFactory(const std::string& name) const; - - /** - * Registers a new DiscoveryAgentFactory with this Registry. If a Factory with the - * given name is already registered it is overwritten with the new one. Once a - * factory is added to the Registry its lifetime is controlled by the Registry, it - * will be deleted once the Registry has been deleted. - * - * @param name - * The name of the new Factory to register. - * @param factory - * The new Factory to add to the Registry. - * - * @throws IllegalArgumentException is name is the empty string. - * @throws NullPointerException if the Factory is Null. - */ - void registerFactory(const std::string& name, DiscoveryAgentFactory* factory); - - /** - * Unregisters the Factory with the given name and deletes that instance of the - * Factory. - * - * @param name - * Name of the Factory to unregister and destroy - */ - void unregisterFactory(const std::string& name); - - /** - * Removes all Factories and deletes the instances of the Factory objects. - */ - void unregisterAllFactories(); - - /** - * Retrieves a list of the names of all the Registered Agents in this - * Registry. - * - * @return stl vector of strings with all the Agent names registered. - */ - std::vector<std::string> getAgentNames() const; - - public: - - /** - * Gets the single instance of the TransportRegistry - * @return reference to the single instance of this Registry - */ - static DiscoveryAgentRegistry& getInstance(); - - private: - - static void initialize(); - static void shutdown(); - - friend class activemq::library::ActiveMQCPP; - - }; - -}}} - -#endif /* _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYAGENTREGISTRY_H_ */ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.cpp b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.cpp deleted file mode 100644 index a260fe7..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.cpp +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "DiscoveryListener.h" - -using namespace activemq; -using namespace activemq::transport; -using namespace activemq::transport::discovery; - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryListener::~DiscoveryListener() { -} http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.h ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.h b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.h deleted file mode 100644 index 461d8a3..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryListener.h +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYLISTENER_H_ -#define _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYLISTENER_H_ - -#include <activemq/util/Config.h> - -#include <activemq/commands/DiscoveryEvent.h> - -namespace activemq { -namespace transport { -namespace discovery { - - class AMQCPP_API DiscoveryListener { - public: - - virtual ~DiscoveryListener(); - - /** - * Called when an discovery agent becomes aware of a new service. - * - * @param event - * A DiscoveryEvent that contains information on the newly discovered service. - */ - virtual void onServiceAdd(const activemq::commands::DiscoveryEvent* event) = 0; - - /** - * Called when an discovery agent determines that a service is no longer available. - * - * @param event - * A DiscoveryEvent that contains information on the removed service. - */ - virtual void onServiceRemove(const activemq::commands::DiscoveryEvent* event) = 0; - - }; - -}}} - -#endif /* _ACTIVEMQ_TRANSPORT_DISCOVERY_DISCOVERYLISTENER_H_ */ http://git-wip-us.apache.org/repos/asf/activemq-cpp/blob/e237191d/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryTransport.cpp ---------------------------------------------------------------------- diff --git a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryTransport.cpp b/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryTransport.cpp deleted file mode 100644 index 731957e..0000000 --- a/activemq-cpp/src/main/activemq/transport/discovery/DiscoveryTransport.cpp +++ /dev/null @@ -1,232 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "DiscoveryTransport.h" - -#include <activemq/util/URISupport.h> -#include <activemq/util/Suspendable.h> -#include <activemq/exceptions/ActiveMQException.h> - -#include <decaf/util/HashMap.h> -#include <decaf/util/StlMap.h> -#include <decaf/util/Properties.h> -#include <decaf/util/concurrent/Mutex.h> -#include <decaf/net/URISyntaxException.h> -#include <decaf/lang/exceptions/NullPointerException.h> - -using namespace decaf; -using namespace decaf::util; -using namespace decaf::util::concurrent; -using namespace decaf::io; -using namespace decaf::net; -using namespace decaf::lang; -using namespace decaf::lang::exceptions; -using namespace activemq; -using namespace activemq::commands; -using namespace activemq::exceptions; -using namespace activemq::util; -using namespace activemq::transport; -using namespace activemq::transport::discovery; - -//////////////////////////////////////////////////////////////////////////////// -const std::string DiscoveryTransport::DISCOVERED_OPTION_PREFIX = "discovered."; - -//////////////////////////////////////////////////////////////////////////////// -namespace activemq { -namespace transport { -namespace discovery { - - class DiscoveryTransportData { - public: - - Pointer<CompositeTransport> next; - Pointer<DiscoveryAgent> agent; - StlMap<std::string, URI> serviceURIs; - Properties parameters; - Mutex lock; - - private: - - DiscoveryTransportData(const DiscoveryTransportData&); - DiscoveryTransportData& operator= (const DiscoveryTransportData&); - - public: - - DiscoveryTransportData() : next(), agent(), serviceURIs(), parameters(), lock() {} - }; - -}}} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryTransport::DiscoveryTransport(Pointer<CompositeTransport> next) : - TransportFilter(next), impl(new DiscoveryTransportData) { -} - -//////////////////////////////////////////////////////////////////////////////// -DiscoveryTransport::~DiscoveryTransport() { - try { - this->close(); - } - AMQ_CATCHALL_NOTHROW() - try { - delete this->impl; - } - AMQ_CATCHALL_NOTHROW() -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::start() { - if (this->impl->agent == NULL) { - throw IllegalStateException(__FILE__, __LINE__, "discoveryAgent not configured"); - } - - // lets pass into the agent the broker name and connection details - this->impl->agent->setDiscoveryListener(this); - this->impl->agent->start(); - - TransportFilter::start(); -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::stop() { - - try { - IOException error; - bool hasException = false; - - try { - this->impl->agent->stop(); - } catch (IOException& ex) { - error = ex; - error.setMark(__FILE__, __LINE__); - hasException = true; - } - - try { - TransportFilter::stop(); - } catch (IOException& ex) { - if (!hasException) { - error = ex; - error.setMark(__FILE__, __LINE__); - hasException = true; - } - } - - if (hasException) { - throw error; - } - } - AMQ_CATCH_RETHROW(IOException) - AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException) - AMQ_CATCHALL_THROW(IOException) -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::doClose() { - try { - this->impl->next.reset(NULL); - } - AMQ_CATCH_RETHROW(IOException) - AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException) - AMQ_CATCHALL_THROW(IOException) -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::setDiscoveryAgent(decaf::lang::Pointer<DiscoveryAgent> agent) { - if (agent == NULL) { - throw NullPointerException(__FILE__, __LINE__, "DiscoveryAgent required to be non-null"); - } - - this->impl->agent = agent; -} - -//////////////////////////////////////////////////////////////////////////////// -Pointer<DiscoveryAgent> DiscoveryTransport::getDiscoveryAgent() const { - return this->impl->agent; -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::setParameters(const Properties& properties) { - this->impl->parameters = properties; -} - -//////////////////////////////////////////////////////////////////////////////// -Properties DiscoveryTransport::getParameters() const { - return this->impl->parameters; -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::onServiceAdd(const DiscoveryEvent* event) { - std::string url = event->getServiceName(); - if (!url.empty()) { - try { - URI uri(url); - uri = URISupport::applyParameters(uri, this->impl->parameters, DISCOVERED_OPTION_PREFIX); - synchronized(&this->impl->lock) { - this->impl->serviceURIs.put(event->getServiceName(), uri); - } - LinkedList<URI> uris; - uris.add(uri); - this->impl->next->addURI(false, uris); - } catch (URISyntaxException& e) { - } - } -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::onServiceRemove(const DiscoveryEvent* event) { - try { - URI uri; - synchronized(&this->impl->lock) { - uri = this->impl->serviceURIs.get(event->getServiceName()); - } - LinkedList<URI> uris; - uris.add(uri); - this->impl->next->removeURI(false, uris); - } catch (NoSuchElementException& e) {} -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::transportInterrupted() { - - Pointer<Suspendable> suspendable; - try { - suspendable = this->impl->next.dynamicCast<Suspendable>(); - suspendable->resume(); - } catch (ClassCastException& e) { - // Not a Suspendable instance. - } catch (Exception& e) { - // Failed to Resume - } - - TransportFilter::transportInterrupted(); -} - -//////////////////////////////////////////////////////////////////////////////// -void DiscoveryTransport::transportResumed() { - Pointer<Suspendable> suspendable; - try { - suspendable = this->impl->next.dynamicCast<Suspendable>(); - suspendable->suspend(); - } catch (ClassCastException& e) { - // Not a Suspendable instance. - } catch (Exception& e) { - // Failed to Suspend - } - - TransportFilter::transportResumed(); -}
