[asterisk-scf-commits] asterisk-scf/integration/media_rtp_pjmedia.git branch "basecomponent" created.
Commits to the Asterisk SCF project code repositories
asterisk-scf-commits at lists.digium.com
Fri Jul 15 17:35:16 CDT 2011
branch "basecomponent" has been created
at 61b4accfa3b8fd620ab29559f64d972bd5474766 (commit)
- Log -----------------------------------------------------------------
commit 61b4accfa3b8fd620ab29559f64d972bd5474766
Author: Ken Hunt <ken.hunt at digium.com>
Date: Fri Jul 15 17:35:21 2011 -0500
Using the base Component class.
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 859c2e9..cd5cc35 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -3,9 +3,10 @@ include_directories(${ice-util-cpp_dir}/include)
include_directories(${logger_dir}/include)
asterisk_scf_slice_include_directories(${API_SLICE_DIR})
-
+
asterisk_scf_component_init(media_rtp_pjmedia)
-asterisk_scf_component_add_file(media_rtp_pjmedia MediaRTPpjmedia.cpp)
+asterisk_scf_component_add_file(media_rtp_pjmedia Component.cpp)
+asterisk_scf_component_add_file(media_rtp_pjmedia RtpReplicationContext.h)
asterisk_scf_component_add_file(media_rtp_pjmedia RTPSession.cpp)
asterisk_scf_component_add_file(media_rtp_pjmedia RTPSource.cpp)
asterisk_scf_component_add_file(media_rtp_pjmedia RTPSink.cpp)
@@ -22,6 +23,7 @@ asterisk_scf_component_add_boost_libraries(media_rtp_pjmedia core thread)
asterisk_scf_component_build_icebox(media_rtp_pjmedia)
target_link_libraries(media_rtp_pjmedia logging-client)
target_link_libraries(media_rtp_pjmedia asterisk-scf-api)
+target_link_libraries(media_rtp_pjmedia ice-util-cpp)
pjproject_link(media_rtp_pjmedia pjlib)
pjproject_link(media_rtp_pjmedia pjlib-util)
pjproject_link(media_rtp_pjmedia pjmedia)
diff --git a/src/Component.cpp b/src/Component.cpp
new file mode 100644
index 0000000..82ec722
--- /dev/null
+++ b/src/Component.cpp
@@ -0,0 +1,561 @@
+/*
+ * Asterisk SCF -- An open-source communications framework.
+ *
+ * Copyright (C) 2010, Digium, Inc.
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk SCF project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE.txt file
+ * at the top of the source tree.
+ */
+
+#include <pjlib.h>
+#include <pjmedia.h>
+
+#include <Ice/Ice.h>
+#include <IceBox/IceBox.h>
+#include <IceUtil/UUID.h>
+
+#include <boost/shared_ptr.hpp>
+
+#include <AsteriskSCF/Core/Discovery/ServiceLocatorIf.h>
+#include <AsteriskSCF/Media/MediaIf.h>
+#include <AsteriskSCF/Media/RTP/MediaRTPIf.h>
+#include <AsteriskSCF/System/Component/ConfigurationIf.h>
+#include <AsteriskSCF/Logger/IceLogger.h>
+#include <AsteriskSCF/logger.h>
+#include <AsteriskSCF/Discovery/SmartProxy.h>
+#include <AsteriskSCF/Component/Component.h>
+
+#include "RtpReplicationContext.h"
+#include "RTPSession.h"
+#include "RtpStateReplicator.h"
+#include "RTPConfiguration.h"
+#include "RtpConfigurationIf.h"
+
+using namespace std;
+using namespace AsteriskSCF::Core::Discovery::V1;
+using namespace AsteriskSCF::Media::V1;
+using namespace AsteriskSCF::Media::RTP::V1;
+using namespace AsteriskSCF::System::Configuration::V1;
+using namespace AsteriskSCF::System::Component::V1;
+using namespace AsteriskSCF::System::Logging;
+using namespace AsteriskSCF::Discovery;
+using namespace AsteriskSCF::Replication;
+
+namespace
+{
+Logger lg = getLoggerFactory().getLogger("AsteriskSCF.MediaRTP");
+}
+
+static const string ReplicaServiceId("MediaRtpReplica");
+static const string MediaServiceId("RTPMediaService");
+static const string MediaComparatorServiceId("RTPMediaServiceComparator");
+
+/**
+ * Implementation of the RTPMediaService interface as defined in MediaRTPIf.ice
+ */
+class RTPMediaServiceImpl : public RTPMediaService
+{
+public:
+ RTPMediaServiceImpl(const Ice::ObjectAdapterPtr&,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr&);
+
+ RTPSessionPrx allocate(const RTPServiceLocatorParamsPtr&, const Ice::Current&);
+ pj_pool_factory *getPoolFactory() { return &mCachingPool.factory; };
+
+private:
+
+ Ice::ObjectAdapterPtr mAdapter;
+ pj_caching_pool mCachingPool;
+ pj_pool_t* mMemoryPool;
+ RtpReplicationContextPtr mReplicationContext;
+ ConfigurationServiceImplPtr mConfigurationService;
+};
+
+/**
+ * Typedef which gives us a smart pointer type for RTPMediaServiceImpl class.
+ */
+typedef IceUtil::Handle<RTPMediaServiceImpl> RTPMediaServiceImplPtr;
+
+/**
+ * Implementation of the ServiceLocatorParamsCompare class
+ */
+class RTPMediaServiceCompareServiceImpl : public ServiceLocatorParamsCompare
+{
+public:
+ bool isSupported(const ServiceLocatorParamsPtr& locatorParams, const Ice::Current&)
+ {
+ RTPServiceLocatorParamsPtr params;
+
+ if (!(params = RTPServiceLocatorParamsPtr::dynamicCast(locatorParams)))
+ {
+ return false;
+ }
+
+ bool result = true;
+
+ // This is done on purpose for additional checks in the future
+ if (params->ipv6 == true)
+ {
+#if defined(PJ_HAS_IPV6) && PJ_HAS_IPV6!=0
+ result = true;
+#else
+ result = false;
+#endif
+ }
+
+ return result;
+ };
+};
+
+/**
+ * Implementation of the Component class.
+ */
+class Component : public AsteriskSCF::Component::Component
+{
+public:
+ Component() :
+ AsteriskSCF::Component::Component(lg, AsteriskSCF::Media::RTP::V1::ComponentServiceDiscoveryCategory),
+ mListeningToReplicator(false), mGeneralState(new RtpGeneralStateItem()) { mGeneralState->key = IceUtil::generateUUID(); };
+
+private:
+ // Required base Component overrides
+ virtual void createPrimaryServices();
+ virtual void registerPrimaryServices();
+ virtual void createReplicationStateListeners();
+ virtual void stopListeningToStateReplicators();
+ virtual void listenToStateReplicators();
+ virtual void findRemoteServices();
+
+ // Optional base Component notifcation overrides
+ virtual void onSuspend();
+ virtual void onResume();
+ virtual void onPreInitialize();
+ virtual void onActivated();
+ virtual void onStandby();
+ virtual void onStop();
+ virtual void onStart();
+
+ // Other base Component overrides
+ virtual void registerBackplaneServices();
+ ReplicationContextPtr createReplicationContext(ReplicationStateType state);
+
+ // A proxy to the service locator manager for the component service.
+ ServiceManagementPrx mComponentServiceManagement;
+
+ // State replicator listener.
+ RtpStateReplicatorListenerPtr mReplicatorListener;
+ RtpStateReplicatorListenerPrx mReplicatorListenerProxy;
+ bool mListeningToReplicator;
+
+ // An instance of the general state information class.
+ RtpGeneralStateItemPtr mGeneralState;
+
+ // Media service
+ RTPMediaServiceImplPtr mRtpMediaServicePtr;
+ RTPMediaServicePrx mRtpMediaServicePrx;
+ LocatorRegistrationWrapperPtr mRtpMediaServiceRegistration;
+
+ // Media comparator service
+ ServiceLocatorParamsComparePtr mRtpMediaComparatorService;
+ ServiceLocatorParamsComparePrx mRtpMediaComparatorServicePrx;
+
+ // Configuration state
+ ConfigurationServiceImplPtr mConfigurationService;
+ ConfigurationServicePrx mConfigurationServicePrx;
+ LocatorRegistrationWrapperPtr mConfigurationRegistration;
+};
+
+void Component::onSuspend()
+{
+ mGeneralState->mServiceManagement->suspend();
+}
+
+void Component::onResume()
+{
+ mGeneralState->mServiceManagement->unsuspend();
+}
+
+/**
+ * Wrapper class around pj_thread_desc.
+ */
+class ThreadDescWrapper
+{
+public:
+ /**
+ * pjthread thread description information, must persist for the life of the thread
+ */
+ pj_thread_desc mDesc;
+};
+
+/**
+ * Type definition used to create a smart pointer for the above.
+ */
+typedef boost::shared_ptr<ThreadDescWrapper> ThreadDescWrapperPtr;
+
+/**
+ * Implementation of the Ice::ThreadNotification class.
+ */
+class pjlibHook : public Ice::ThreadNotification
+{
+public:
+ /**
+ * Implementation of the start function which is called when a thread starts.
+ */
+ void start()
+ {
+ ThreadDescWrapperPtr wrapper = ThreadDescWrapperPtr(new ThreadDescWrapper());
+ pj_thread_t *thread;
+ pj_thread_register("ICE Thread", wrapper->mDesc, &thread);
+ boost::lock_guard<boost::mutex> lock(mLock);
+ pjThreads.insert(make_pair(thread, wrapper));
+ }
+
+ /**
+ * Implementation of the stop function which is called when a thread stops.
+ */
+ void stop()
+ {
+ if (pj_thread_is_registered())
+ {
+ boost::lock_guard<boost::mutex> lock(mLock);
+ pjThreads.erase(pj_thread_this());
+ }
+ }
+private:
+ /**
+ * A map containing thread lifetime persistent data.
+ */
+ map<pj_thread_t*, ThreadDescWrapperPtr> pjThreads;
+
+ /**
+ * Mutex to protect the map
+ */
+ boost::mutex mLock;
+};
+
+/**
+ * Constructor for the RTPMediaServiceImpl class.
+ */
+RTPMediaServiceImpl::RTPMediaServiceImpl(const Ice::ObjectAdapterPtr& adapter,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr& configurationService) :
+ mAdapter(adapter), mReplicationContext(replicationContext), mConfigurationService(configurationService)
+{
+ /* Initialize the memory caching pool using default policy as specified by pjlib. */
+ pj_caching_pool_init(&mCachingPool, &pj_pool_factory_default_policy, 0);
+
+ /* Initialize the memory pool that pjmedia will draw from. */
+ mMemoryPool = pj_pool_create(&mCachingPool.factory, "media_rtp_pjmedia", 1000, 1000, NULL);
+}
+
+/**
+ * Implementation of the allocate method as defined in MediaRTPIf.ice
+ */
+RTPSessionPrx RTPMediaServiceImpl::allocate(const RTPServiceLocatorParamsPtr& params, const Ice::Current&)
+{
+ RTPSessionImplPtr session =
+ new RTPSessionImpl(mAdapter, params, &mCachingPool.factory, mReplicationContext, mConfigurationService);
+ return session->getProxy();
+}
+
+void Component::onPreInitialize()
+{
+ /* Initialize pjlib as pjmedia will be using it */
+ pj_status_t status = pj_init();
+ if (status != PJ_SUCCESS)
+ {
+ lg(Error) << "PJ library initialization failed.";
+ return;
+ }
+
+ if ((status = pjlib_util_init()) != PJ_SUCCESS)
+ {
+ lg(Error) << "PJ Utility library initialization failed.";
+ return;
+ }
+
+ lg(Info) << "Initializing pjmedia rtp component" << endl;
+
+ Ice::InitializationData id;
+ id.threadHook = new pjlibHook();
+ id.properties = getCommunicator()->getProperties();
+
+ // To use our thread-hook, we need to set an alternate
+ // communicator in our Component base.
+ setCommunicator(Ice::initialize(id));
+}
+
+/**
+ * Override of factory method to create our custom replication context.
+ */
+ReplicationContextPtr Component::createReplicationContext(ReplicationStateType state)
+{
+ RtpReplicationContextPtr context(new RtpReplicationContext(state));
+ return context;
+}
+
+/**
+ * Create the objects that implement the main services this component provides
+ * the system.
+ */
+void Component::createPrimaryServices()
+{
+ try
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ mConfigurationService = new ConfigurationServiceImpl();
+ ConfigurationServicePrx mConfigurationServiceProxy = ConfigurationServicePrx::uncheckedCast(
+ getServiceAdapter()->addWithUUID(mConfigurationService));
+
+ mRtpMediaServicePtr =
+ new RTPMediaServiceImpl(getServiceAdapter(), rtpReplicationContext, mConfigurationService);
+ mRtpMediaServicePrx = RTPMediaServicePrx::uncheckedCast(getServiceAdapter()->add(mRtpMediaServicePtr,
+ getCommunicator()->stringToIdentity(MediaServiceId)));
+
+ mRtpMediaComparatorService = new RTPMediaServiceCompareServiceImpl();
+ mRtpMediaComparatorServicePrx = ServiceLocatorParamsComparePrx::uncheckedCast(
+ getServiceAdapter()->add(mRtpMediaComparatorService, getCommunicator()->stringToIdentity(MediaComparatorServiceId)));
+
+ if (rtpReplicationContext->isActive() == true)
+ {
+ mGeneralState->mComparatorId = IceUtil::generateUUID();
+ getServiceLocatorManagement()->addCompare(mGeneralState->mComparatorId, mRtpMediaComparatorServicePrx);
+ }
+
+ }
+ catch(const Ice::Exception& e)
+ {
+ lg(Critical) << getName() << " : " << BOOST_CURRENT_FUNCTION << " : " << e.what();
+ }
+}
+
+/**
+ * Register this component's backplane interfaces with the Service Locator.
+ * This enables other Asterisk SCF components to locate our interfaces.
+ */
+void Component::registerBackplaneServices()
+{
+ // Insure the default Component services are registered.
+ Component::registerBackplaneServices();
+
+ try
+ {
+ std::string serviceName = getCommunicator()->getProperties()->getPropertyWithDefault(
+ getName() + ".Service", "default");
+
+ // Register our configuration interface with the Service Locator.
+ mConfigurationRegistration = wrapServiceForRegistration(mConfigurationServicePrx,
+ ConfigurationDiscoveryCategory,
+ serviceName,
+ getName());
+ bool registered = manageBackplaneService(mConfigurationRegistration);
+ }
+ catch(const std::exception& e)
+ {
+ lg(Error) << "Exception in " << getName() << ", " << BOOST_CURRENT_FUNCTION << " : " << e.what();
+ }
+}
+
+void Component::findRemoteServices()
+{
+ if (getReplicationContext()->getState() == ACTIVE_STANDALONE)
+ {
+ return;
+ }
+
+ // Look for the configured state replicator or default one
+ ServiceLocatorParamsPtr replicatorParams = new ServiceLocatorParams();
+ replicatorParams->category = StateReplicatorDiscoveryCategory;
+ replicatorParams->service =
+ getCommunicator()->getProperties()->getPropertyWithDefault("Rtp.StateReplicatorService", "default");
+
+ try
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> pw(getServiceLocator(), replicatorParams, lg);
+ rtpReplicationContext->setReplicator(pw);
+
+ // Since we're not in standalone mode, we'll get our configuration updates routed via the
+ // replicator service.
+ ConfigurationReplicatorPrx configurationReplicator = ConfigurationReplicatorPrx::checkedCast(
+ rtpReplicationContext->getReplicator().initialize(), ReplicatorFacet);
+ configurationReplicator->registerConfigurationService(mConfigurationServicePrx);
+
+ }
+ catch (...)
+ {
+ lg(Error) << "State replicator could not be found, operating without.";
+ }
+}
+
+void Component::createReplicationStateListeners()
+{
+ try
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ // Create and publish our state replicator listener interface.
+ mReplicatorListener = new RtpStateReplicatorListenerI(getServiceAdapter(), mRtpMediaServicePtr->getPoolFactory(),
+ mGeneralState, rtpReplicationContext, mConfigurationService);
+ RtpStateReplicatorListenerPrx replicatorListener = RtpStateReplicatorListenerPrx::uncheckedCast(
+ getBackplaneAdapter()->addWithUUID(mReplicatorListener));
+ mReplicatorListenerProxy = RtpStateReplicatorListenerPrx::uncheckedCast(replicatorListener->ice_oneway());
+
+ lg(Debug) << "Got proxy to RTP state replicator";
+ }
+ catch(const Ice::Exception &e)
+ {
+ lg(Error) << getName() << " in " << BOOST_CURRENT_FUNCTION << " : " << e.what();
+ throw;
+ }
+}
+
+void Component::listenToStateReplicators()
+{
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ if (mListeningToReplicator == true)
+ {
+ return;
+ }
+
+ if (!rtpReplicationContext->getReplicator().isInitialized())
+ {
+ lg(Error) << getName() << " : State replicator could not be found. Unable to listen for state updates!";
+ return;
+ }
+
+ try
+ {
+ // Are we in standby mode?
+ if (rtpReplicationContext->getState() == ACTIVE_IN_REPLICA_GROUP)
+ {
+ rtpReplicationContext->getReplicator()->addListener(mReplicatorListenerProxy);
+ mListeningToReplicator = true;
+ }
+ }
+ catch (const Ice::Exception& e)
+ {
+ lg(Error) << e.what();
+ throw;
+ }
+}
+
+/**
+ * Unregister as a listener to our state replicator.
+ * A component in active mode doesn't neeed to listen to
+ * state replication data.
+ */
+void Component::stopListeningToStateReplicators()
+{
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ if ((!rtpReplicationContext->getReplicator().isInitialized()) || (mListeningToReplicator == false))
+ {
+ return;
+ }
+
+ try
+ {
+ rtpReplicationContext->getReplicator()->removeListener(mReplicatorListenerProxy);
+ mListeningToReplicator = false;
+ }
+ catch (const Ice::Exception& e)
+ {
+ lg(Error) << e.what();
+ throw;
+ }
+}
+
+/**
+ * Register this component's primary public interfaces with the Service Locator.
+ * This enables other Asterisk SCF components to locate the interfaces we are publishing.
+ *
+ */
+void Component::registerPrimaryServices()
+{
+ try
+ {
+ std::string serviceName = getCommunicator()->getProperties()->getPropertyWithDefault(
+ getName() + ".Service", "default");
+
+ mRtpMediaServiceRegistration = wrapServiceForRegistration(mRtpMediaServicePrx,
+ "rtp",
+ serviceName,
+ getName());
+ bool registered = managePrimaryService(mRtpMediaServiceRegistration);
+ }
+ catch(const std::exception& e)
+ {
+ lg(Error) << "Unable to publish component interfaces in " << getName() << BOOST_CURRENT_FUNCTION <<
+ ". Exception: " << e.what();
+ throw; // rethrow
+ }
+}
+
+/**
+ * These overrides will be needed. I don't think this component
+ * handled this well in the past. It simply checked if it was active/standby
+ * during startup.
+ */
+void Component::onActivated()
+{
+}
+
+/**
+ * These overrides will be needed. I don't think this component
+ * handled this well in the past. It simply checked if it was active/standby
+ * during startup.
+ */
+void Component::onStandby()
+{
+}
+
+void Component::onStart()
+{
+ // Note: I don't think this is necessary. If we make the
+ // id computed from a "service" identifier (which could default
+ // to "default", there's nothing replicated here that the standby component
+ // couldn't already determine itself.
+ if (getReplicationContext()->isReplicating() == true)
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ RtpStateItemSeq items;
+ items.push_back(mGeneralState);
+ RtpStateReplicatorPrx oneway = RtpStateReplicatorPrx::uncheckedCast(rtpReplicationContext->getReplicator()->ice_oneway());
+ oneway->setState(items);
+ }
+}
+
+void Component::onStop()
+{
+ if (getReplicationContext()->isActive() == true)
+ {
+ mGeneralState->mServiceManagement->unregister();
+ }
+}
+
+extern "C"
+{
+ASTERISK_SCF_ICEBOX_EXPORT IceBox::Service* create(Ice::CommunicatorPtr)
+{
+ return new Component;
+}
+}
diff --git a/src/MediaRTPpjmedia.cpp b/src/MediaRTPpjmedia.cpp
deleted file mode 100644
index 78671fb..0000000
--- a/src/MediaRTPpjmedia.cpp
+++ /dev/null
@@ -1,625 +0,0 @@
-/*
- * Asterisk SCF -- An open-source communications framework.
- *
- * Copyright (C) 2010, Digium, Inc.
- *
- * See http://www.asterisk.org for more information about
- * the Asterisk SCF project. Please do not directly contact
- * any of the maintainers of this project for assistance;
- * the project provides a web site, mailing lists and IRC
- * channels for your use.
- *
- * This program is free software, distributed under the terms of
- * the GNU General Public License Version 2. See the LICENSE.txt file
- * at the top of the source tree.
- */
-
-#include <pjlib.h>
-#include <pjmedia.h>
-
-#include <Ice/Ice.h>
-#include <IceBox/IceBox.h>
-#include <IceUtil/UUID.h>
-
-#include <boost/shared_ptr.hpp>
-
-#include <AsteriskSCF/Core/Discovery/ServiceLocatorIf.h>
-#include <AsteriskSCF/Media/MediaIf.h>
-#include <AsteriskSCF/Media/RTP/MediaRTPIf.h>
-#include <AsteriskSCF/System/Component/ConfigurationIf.h>
-#include <AsteriskSCF/System/Component/ComponentServiceIf.h>
-#include <AsteriskSCF/System/Component/ReplicaIf.h>
-#include <AsteriskSCF/Logger/IceLogger.h>
-#include <AsteriskSCF/logger.h>
-#include <AsteriskSCF/Discovery/SmartProxy.h>
-
-#include "RtpStateReplicationIf.h"
-
-#include "RTPSession.h"
-#include "RtpStateReplicator.h"
-#include "RTPConfiguration.h"
-#include "RtpConfigurationIf.h"
-
-using namespace std;
-using namespace AsteriskSCF::Core::Discovery::V1;
-using namespace AsteriskSCF::Media::V1;
-using namespace AsteriskSCF::Media::RTP::V1;
-using namespace AsteriskSCF::System::Configuration::V1;
-using namespace AsteriskSCF::System::Component::V1;
-using namespace AsteriskSCF::System::Logging;
-using namespace AsteriskSCF::Discovery;
-
-namespace
-{
-Logger lg = getLoggerFactory().getLogger("AsteriskSCF.MediaRTP");
-}
-
-static const string ReplicaServiceId("MediaRtpReplica");
-static const string MediaServiceId("RTPMediaService");
-static const string MediaComparatorServiceId("RTPMediaServiceComparator");
-
-/**
- * Implementation of the RTPMediaService interface as defined in MediaRTPIf.ice
- */
-class RTPMediaServiceImpl : public RTPMediaService
-{
-public:
- RTPMediaServiceImpl(const Ice::ObjectAdapterPtr&, const ReplicaPtr&,
- const AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx>&,
- const ConfigurationServiceImplPtr&);
- RTPSessionPrx allocate(const RTPServiceLocatorParamsPtr&, const Ice::Current&);
- pj_pool_factory *getPoolFactory() { return &mCachingPool.factory; };
-private:
- /**
- * A pointer to the object adapter that objects should be added to.
- */
- Ice::ObjectAdapterPtr mAdapter;
-
- /**
- * Memory caching pool.
- */
- pj_caching_pool mCachingPool;
-
- /**
- * Memory pool.
- */
- pj_pool_t* mMemoryPool;
-
- /**
- * A pointer to the replica service.
- */
- ReplicaPtr mReplicaService;
-
- /**
- * A pointer to the configuration service.
- */
- ConfigurationServiceImplPtr mConfigurationService;
-
- /**
- * A proxy to the state replicator.
- */
- AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> mStateReplicator;
-};
-
-/**
- * Typedef which gives us a smart pointer type for RTPMediaServiceImpl class.
- */
-typedef IceUtil::Handle<RTPMediaServiceImpl> RTPMediaServiceImplPtr;
-
-/**
- * This class provides implementation for the Replica interface.
- */
-class ReplicaImpl : public Replica
-{
-public:
- ReplicaImpl(const Ice::ObjectAdapterPtr& adapter) : mAdapter(adapter), mPaused(false), mActive(true) { }
-
- bool isActive(const Ice::Current&)
- {
- return mActive;
- }
-
- bool activate(const Ice::Current&)
- {
- mActive = true;
-
- for (vector<AsteriskSCF::System::Component::V1::ReplicaListenerPrx>::const_iterator listener =
- mListeners.begin(); listener != mListeners.end(); ++listener)
- {
- (*listener)->activated(ReplicaPrx::uncheckedCast(
- mAdapter->createDirectProxy(mAdapter->getCommunicator()->stringToIdentity(ReplicaServiceId))));
- }
-
- return true;
- }
-
- void standby(const Ice::Current&)
- {
- mActive = false;
-
- for (vector<AsteriskSCF::System::Component::V1::ReplicaListenerPrx>::const_iterator listener =
- mListeners.begin(); listener != mListeners.end(); ++listener)
- {
- (*listener)->onStandby(ReplicaPrx::uncheckedCast(
- mAdapter->createDirectProxy(mAdapter->getCommunicator()->stringToIdentity(ReplicaServiceId))));
- }
- }
-
- void addListener(const AsteriskSCF::System::Component::V1::ReplicaListenerPrx& listener, const Ice::Current&)
- {
- mListeners.push_back(listener);
- }
-
- void removeListener(const AsteriskSCF::System::Component::V1::ReplicaListenerPrx& listener, const Ice::Current&)
- {
- mListeners.erase(std::remove(mListeners.begin(), mListeners.end(), listener), mListeners.end());
- }
-
-private:
- /**
- * Pointer to the object adapter we exist on.
- */
- Ice::ObjectAdapterPtr mAdapter;
-
- /**
- * Listeners that we need to push state change notifications out to.
- */
- vector<AsteriskSCF::System::Component::V1::ReplicaListenerPrx> mListeners;
-
- bool mPaused;
-
- bool mActive;
-};
-
-/**
- * Implementation of the ServiceLocatorParamsCompare class
- */
-class RTPMediaServiceCompareServiceImpl : public ServiceLocatorParamsCompare
-{
-public:
- bool isSupported(const ServiceLocatorParamsPtr& locatorParams, const Ice::Current&)
- {
- RTPServiceLocatorParamsPtr params;
-
- if (!(params = RTPServiceLocatorParamsPtr::dynamicCast(locatorParams)))
- {
- return false;
- }
-
- bool result = true;
-
- // This is done on purpose for additional checks in the future
- if (params->ipv6 == true)
- {
-#if defined(PJ_HAS_IPV6) && PJ_HAS_IPV6!=0
- result = true;
-#else
- result = false;
-#endif
- }
-
- return result;
- };
-};
-
-/**
- * Implementation of the IceBox::Service class
- */
-class MediaRTPpjmediaApp : public IceBox::Service
-{
-public:
- MediaRTPpjmediaApp() : mGeneralState(new RtpGeneralStateItem()) { mGeneralState->key = IceUtil::generateUUID(); };
- void start(const std::string&, const Ice::CommunicatorPtr&, const Ice::StringSeq&);
- void stop();
-
-private:
- /**
- * Ice Communicator used for this service.
- */
- Ice::CommunicatorPtr mCommunicator;
-
- /**
- * Object adapter that global stuff is associated with.
- */
- Ice::ObjectAdapterPtr mGlobalAdapter;
-
- /**
- * Object adapter that local stuff is associated with.
- */
- Ice::ObjectAdapterPtr mLocalAdapter;
-
- /**
- * The object adapter for the Logger.
- */
- Ice::ObjectAdapterPtr mLoggerAdapter;
-
- /**
- * A proxy to the service locator manager for the component service.
- */
- ServiceManagementPrx mComponentServiceManagement;
-
- /**
- * Instance of our replica implementation.
- */
- ReplicaPtr mReplicaService;
-
- /**
- * Instance of our configuration service implementation.
- */
- ConfigurationServiceImplPtr mConfigurationService;
-
- /**
- * Instance of our state replicator listener.
- */
- RtpStateReplicatorListenerPtr mReplicatorListener;
-
- /**
- * A proxy to our state replicator listener.
- */
- RtpStateReplicatorListenerPrx mReplicatorListenerProxy;
-
- /**
- * A proxy to the state replicator.
- */
- AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> mStateReplicator;
-
- /**
- * An instance of the general state information class.
- */
- RtpGeneralStateItemPtr mGeneralState;
-
- /**
- * A proxy to the service locator management service.
- */
- ServiceLocatorManagementPrx mManagement;
-
- /**
- * A proxy to the service locator manager for the configuration service.
- */
- ServiceManagementPrx mConfigurationManagement;
-
- /**
- * Unique guid for configuration service name comparator.
- */
- std::string mConfigCompareGuid;
-};
-
-/**
- * Implementation of the ComponentService interface as defined in ComponentServiceIf.ice
- */
-class ComponentServicepjmediaImpl : public ComponentService
-{
-public:
- /**
- * A constructor for this implementation which just sets a few variables, nothing extreme.
- */
- ComponentServicepjmediaImpl(MediaRTPpjmediaApp& app, const RtpGeneralStateItemPtr& generalState) :
- mApplication(app), mGeneralState(generalState) { };
-
- /**
- * An implementation of the suspend method which actually suspends ourselves
- * from the service locator.
- */
- virtual void suspend(const ::Ice::Current&)
- {
- mGeneralState->mServiceManagement->suspend();
- }
-
- /**
- * An implementation of the resume method which actually unsuspends ourselves
- * from the service locator.
- */
- virtual void resume(const ::Ice::Current&)
- {
- mGeneralState->mServiceManagement->unsuspend();
- }
-
- /**
- * An implementation of the shutdown method which really does shut us down.
- * Goodbye cruel world.
- */
- virtual void shutdown(const ::Ice::Current&)
- {
- // TODO - Actually support this
- }
-
-private:
- /**
- * Our application instance, used for shutting the component down.
- */
- MediaRTPpjmediaApp& mApplication;
-
- /**
- * Pointer to general state information.
- */
- RtpGeneralStateItemPtr mGeneralState;
-};
-
-/**
- * Wrapper class around pj_thread_desc.
- */
-class ThreadDescWrapper
-{
-public:
- /**
- * pjthread thread description information, must persist for the life of the thread
- */
- pj_thread_desc mDesc;
-};
-
-/**
- * Type definition used to create a smart pointer for the above.
- */
-typedef boost::shared_ptr<ThreadDescWrapper> ThreadDescWrapperPtr;
-
-/**
- * Implementation of the Ice::ThreadNotification class.
- */
-class pjlibHook : public Ice::ThreadNotification
-{
-public:
- /**
- * Implementation of the start function which is called when a thread starts.
- */
- void start()
- {
- ThreadDescWrapperPtr wrapper = ThreadDescWrapperPtr(new ThreadDescWrapper());
- pj_thread_t *thread;
- pj_thread_register("ICE Thread", wrapper->mDesc, &thread);
- boost::lock_guard<boost::mutex> lock(mLock);
- pjThreads.insert(make_pair(thread, wrapper));
- }
-
- /**
- * Implementation of the stop function which is called when a thread stops.
- */
- void stop()
- {
- if (pj_thread_is_registered())
- {
- boost::lock_guard<boost::mutex> lock(mLock);
- pjThreads.erase(pj_thread_this());
- }
- }
-private:
- /**
- * A map containing thread lifetime persistent data.
- */
- map<pj_thread_t*, ThreadDescWrapperPtr> pjThreads;
-
- /**
- * Mutex to protect the map
- */
- boost::mutex mLock;
-};
-
-/**
- * Constructor for the RTPMediaServiceImpl class.
- */
-RTPMediaServiceImpl::RTPMediaServiceImpl(const Ice::ObjectAdapterPtr& adapter, const ReplicaPtr& replicaService,
- const AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx>& stateReplicator,
- const ConfigurationServiceImplPtr& configurationService) :
- mAdapter(adapter), mReplicaService(replicaService), mConfigurationService(configurationService),
- mStateReplicator(stateReplicator)
-{
- /* Initialize the memory caching pool using default policy as specified by pjlib. */
- pj_caching_pool_init(&mCachingPool, &pj_pool_factory_default_policy, 0);
-
- /* Initialize the memory pool that pjmedia will draw from. */
- mMemoryPool = pj_pool_create(&mCachingPool.factory, "media_rtp_pjmedia", 1000, 1000, NULL);
-}
-
-/**
- * Implementation of the allocate method as defined in MediaRTPIf.ice
- */
-RTPSessionPrx RTPMediaServiceImpl::allocate(const RTPServiceLocatorParamsPtr& params, const Ice::Current&)
-{
- RTPSessionImplPtr session =
- new RTPSessionImpl(mAdapter, params, &mCachingPool.factory, mReplicaService, mStateReplicator,
- mConfigurationService);
- return session->getProxy();
-}
-
-/**
- * Implementation of the IceBox::Service::start method.
- */
-void MediaRTPpjmediaApp::start(const std::string&, const Ice::CommunicatorPtr& communicator,
- const Ice::StringSeq&)
-{
- // we need a logger before we're ready to build the real communicator.
- // use the one we're provided to create the IceLogger.
- mLoggerAdapter = communicator->createObjectAdapter("MediaRTPpjmediaAdapterLogger");
- ConfiguredIceLoggerPtr iceLogger = createIceLogger(mLoggerAdapter);
- getLoggerFactory().setLogOutput(iceLogger->getLogger());
- mLoggerAdapter->activate();
-
- /* Initialize pjlib as pjmedia will be using it */
- pj_status_t status = pj_init();
- if (status != PJ_SUCCESS)
- {
- lg(Error) << "PJ library initialization failed.";
- return;
- }
-
- if ((status = pjlib_util_init()) != PJ_SUCCESS)
- {
- lg(Error) << "PJ Utility library initialization failed.";
- return;
- }
-
- lg(Info) << "Initializing pjmedia rtp component" << endl;
-
- Ice::InitializationData id;
- id.threadHook = new pjlibHook();
- id.properties = communicator->getProperties();
-
- mCommunicator = Ice::initialize(id);
-
- mLocalAdapter = mCommunicator->createObjectAdapter("MediaRTPpjmediaAdapterLocal");
-
- mReplicaService = new ReplicaImpl(mLocalAdapter);
- mLocalAdapter->add(mReplicaService, mCommunicator->stringToIdentity(ReplicaServiceId));
-
- mConfigurationService = new ConfigurationServiceImpl();
- ConfigurationServicePrx mConfigurationServiceProxy = ConfigurationServicePrx::uncheckedCast(
- mLocalAdapter->addWithUUID(mConfigurationService));
-
- mLocalAdapter->activate();
-
- mGlobalAdapter = mCommunicator->createObjectAdapter("MediaRTPpjmediaAdapter");
-
- mGlobalAdapter->activate();
-
- lg(Info) << "Activated pjmedia rtp component media service." << endl;
-
- mManagement = ServiceLocatorManagementPrx::checkedCast(mCommunicator->propertyToProxy("ServiceLocatorManagementProxy"));
-
- // The service locator is required for state replicator operation, so go ahead and find it
- ServiceLocatorPrx locator = ServiceLocatorPrx::checkedCast(mCommunicator->propertyToProxy("LocatorService.Proxy"));
-
- // Look for the configured state replicator or default one
- ServiceLocatorParamsPtr replicatorParams = new ServiceLocatorParams();
- replicatorParams->category = StateReplicatorDiscoveryCategory;
- replicatorParams->service = mCommunicator->getProperties()->getPropertyWithDefault("Rtp.StateReplicatorService", "default");
- replicatorParams->id =
- mCommunicator->getProperties()->getProperty("Rtp.StateReplicatorName");
-
- try
- {
- AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> pw(locator, replicatorParams, lg);
- mStateReplicator = pw;
- }
- catch (...)
- {
- lg(Error) << "State replicator could not be found, operating without.";
- }
-
- std::string serviceName = mCommunicator->getProperties()->getPropertyWithDefault(
- "Rtp.ServiceName", "default");
-
- RTPMediaServiceImplPtr rtpmediaservice =
- new RTPMediaServiceImpl(mGlobalAdapter, mReplicaService, mStateReplicator, mConfigurationService);
-
- if (mCommunicator->getProperties()->getPropertyWithDefault("Rtp.Standalone", "false") == "true")
- {
- // Publish the configuration service IceStorm topic so everybody gets configuration
- mConfigurationManagement = ServiceManagementPrx::uncheckedCast(
- mManagement->addService(mConfigurationServiceProxy, ""));
-
- // Populate the configuration parameters with details so we can be found
- ServiceLocatorParamsPtr configurationParams = new ServiceLocatorParams();
- configurationParams->category = ConfigurationDiscoveryCategory;
- configurationParams->service = serviceName;
- configurationParams->id = mCommunicator->getProperties()->getPropertyWithDefault("RtpConfiguration.Name", "");
-
- mConfigurationManagement->addLocatorParams(configurationParams, "");
- }
- else if (mStateReplicator)
- {
- ConfigurationReplicatorPrx configurationReplicator = ConfigurationReplicatorPrx::checkedCast(
- mStateReplicator.initialize(), ReplicatorFacet);
- configurationReplicator->registerConfigurationService(mConfigurationServiceProxy);
- }
-
- if (mStateReplicator)
- {
- mReplicatorListener =
- new RtpStateReplicatorListenerI(mGlobalAdapter, rtpmediaservice->getPoolFactory(), mGeneralState,
- mConfigurationService);
- mReplicatorListenerProxy =
- RtpStateReplicatorListenerPrx::uncheckedCast(mLocalAdapter->addWithUUID(mReplicatorListener));
-
- if (mCommunicator->getProperties()->getPropertyWithDefault("Rtp.StateReplicatorListener", "no") == "yes")
- {
- mStateReplicator->addListener(mReplicatorListenerProxy);
- mReplicaService->standby();
- lg(Info) << "Operating as a standby replica." << endl;
- }
- else
- {
- lg(Info) << "Operating in an active state." << endl;
- }
- }
-
- ServiceLocatorParamsComparePtr rtpmediacomparatorservice = new RTPMediaServiceCompareServiceImpl();
- ServiceLocatorParamsComparePrx RTPMediaComparatorServiceProxy = ServiceLocatorParamsComparePrx::uncheckedCast(
- mGlobalAdapter->add(rtpmediacomparatorservice, mCommunicator->stringToIdentity(MediaComparatorServiceId)));
-
- if (mReplicaService->isActive() == true)
- {
- mGeneralState->mComparatorId = IceUtil::generateUUID();
- mManagement->addCompare(mGeneralState->mComparatorId, RTPMediaComparatorServiceProxy);
- }
-
-
- RTPMediaServicePrx RTPMediaServiceProxy = RTPMediaServicePrx::uncheckedCast(mGlobalAdapter->add(rtpmediaservice,
- mCommunicator->stringToIdentity(MediaServiceId)));
-
- RTPServiceLocatorParamsPtr rtpparams = new RTPServiceLocatorParams();
-
- if (mReplicaService->isActive() == true)
- {
- mGeneralState->mServiceManagement = ServiceManagementPrx::uncheckedCast(
- mManagement->addService(RTPMediaServiceProxy, "media_rtp_pjmedia"));
-
- /* Now we can add some parameters to help find us. */
- rtpparams->category = "rtp";
- rtpparams->service = serviceName;
- mGeneralState->mServiceManagement->addLocatorParams(rtpparams, mGeneralState->mComparatorId);
- }
-
- ServiceLocatorParamsPtr genericparams = new ServiceLocatorParams();
-
- /* One must provide a component service to manage us, if someone wants to */
- ComponentServicePtr ComponentService = new ComponentServicepjmediaImpl(*this, mGeneralState);
- ComponentServicePrx ComponentServiceProxy =
- ComponentServicePrx::uncheckedCast(mLocalAdapter->addWithUUID(ComponentService));
-
- /* Let's add the component service to the service locator first */
- mComponentServiceManagement =
- ServiceManagementPrx::uncheckedCast(mManagement->addService(ComponentServiceProxy, "media_rtp_pjmedia"));
- genericparams->category = "Component/media_rtp_pjmedia";
- genericparams->service = serviceName;
- mComponentServiceManagement->addLocatorParams(genericparams, "");
-
- // Replicate general state information so the backup is ready
- if (mStateReplicator && mReplicaService->isActive() == true)
- {
- RtpStateItemSeq items;
- items.push_back(mGeneralState);
- RtpStateReplicatorPrx oneway = RtpStateReplicatorPrx::uncheckedCast(mStateReplicator->ice_oneway());
- oneway->setState(items);
- }
-}
-
-/**
- * Implementation of the IceBox::Service::stop method.
- */
-void MediaRTPpjmediaApp::stop()
-{
- mComponentServiceManagement->unregister();
- if (mReplicaService->isActive() == true)
- {
- mGeneralState->mServiceManagement->unregister();
- }
- if (mConfigurationManagement)
- {
- mConfigurationManagement->unregister();
- }
- if (!mConfigCompareGuid.empty())
- {
- mManagement->removeCompare(mConfigCompareGuid);
- ServiceLocatorManagementPrx management =
- ServiceLocatorManagementPrx::checkedCast(mCommunicator->propertyToProxy("ServiceLocatorManagementProxy"));
- management->removeCompare(mGeneralState->mComparatorId);
- }
- mCommunicator->destroy();
-}
-
-extern "C"
-{
-ASTERISK_SCF_ICEBOX_EXPORT IceBox::Service* create(Ice::CommunicatorPtr)
-{
- return new MediaRTPpjmediaApp;
-}
-}
diff --git a/src/RTPConfiguration.h b/src/RTPConfiguration.h
index 96d2180..790d475 100644
--- a/src/RTPConfiguration.h
+++ b/src/RTPConfiguration.h
@@ -54,3 +54,4 @@ private:
* A typedef which creates a smart pointer type for ConfigurationServiceImpl.
*/
typedef IceUtil::Handle<ConfigurationServiceImpl> ConfigurationServiceImplPtr;
+
diff --git a/src/RTPSession.cpp b/src/RTPSession.cpp
index 7bbf164..c016922 100644
--- a/src/RTPSession.cpp
+++ b/src/RTPSession.cpp
@@ -56,11 +56,10 @@ class RTPSessionImplPriv
{
public:
RTPSessionImplPriv(const Ice::ObjectAdapterPtr& adapter, const FormatSeq& formats,
- const ReplicaPtr& replicaService,
- const AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx>& stateReplicator) :
+ const RtpReplicationContextPtr& replicationContext) :
mAdapter(adapter), mFormats(formats),
mSessionStateItem(new RtpSessionStateItem()),
- mReplicaService(replicaService), mStateReplicator(stateReplicator) { };
+ mReplicationContext(replicationContext) { };
~RTPSessionImplPriv();
/**
@@ -118,25 +117,16 @@ public:
*/
RtpSessionStateItemPtr mSessionStateItem;
- /**
- * A pointer to the replica instance.
- */
- ReplicaPtr mReplicaService;
-
- /**
- * A proxy to the state replicator where we are sending updates to.
- */
- AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> mStateReplicator;
+ RtpReplicationContextPtr mReplicationContext;
};
/**
* Constructor for the RTPSessionImpl class (used by Ice).
*/
RTPSessionImpl::RTPSessionImpl(const Ice::ObjectAdapterPtr& adapter, const RTPServiceLocatorParamsPtr& params,
- pj_pool_factory* factory, const ReplicaPtr& replicaService,
- const AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx>& stateReplicator,
+ pj_pool_factory* factory, const RtpReplicationContextPtr& replicationContext,
const ConfigurationServiceImplPtr& configurationService) :
- mImpl(new RTPSessionImplPriv(adapter, params->formats, replicaService, stateReplicator))
+ mImpl(new RTPSessionImplPriv(adapter, params->formats, replicationContext))
{
/* Add ourselves to the ICE ASM so we can be used. */
mImpl->mProxy = RTPSessionPrx::uncheckedCast(adapter->addWithUUID(this));
@@ -182,12 +172,12 @@ RTPSessionImpl::RTPSessionImpl(const Ice::ObjectAdapterPtr& adapter, const RTPSe
}
/**
- * Constructor for the RTPSessionImpl class (used by state replicator).
+ * Constructor for the RTPSessionImpl class (used by state replicator listener).
*/
RTPSessionImpl::RTPSessionImpl(const Ice::ObjectAdapterPtr& adapter, pj_pool_factory* factory,
const Ice::Identity& sessionIdentity, const Ice::Identity& sinkIdentity, const Ice::Identity& sourceIdentity,
- Ice::Int port, const FormatSeq& formats, bool ipv6, const ConfigurationServiceImplPtr& configurationService) :
- mImpl(new RTPSessionImplPriv(adapter, formats, 0, *(new AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx>)))
+ Ice::Int port, const FormatSeq& formats, bool ipv6, const RtpReplicationContextPtr& replicationContext, const ConfigurationServiceImplPtr& configurationService) :
+ mImpl(new RTPSessionImplPriv(adapter, formats, replicationContext))
{
mImpl->mProxy = RTPSessionPrx::uncheckedCast(adapter->add(this, sessionIdentity));
@@ -407,42 +397,34 @@ void RTPSessionImpl::replicateState(const RtpSessionStateItemPtr& session, const
const RtpStreamSourceStateItemPtr& source)
{
// If state replication has been disabled do nothing
- if (!mImpl->mStateReplicator || mImpl->mReplicaService->isActive() == false)
+ if (mImpl->mReplicationContext->isReplicating() == false)
{
- return;
+ return;
}
RtpStateItemSeq items;
if (session)
{
- items.push_back(session);
+ items.push_back(session);
}
if (sink)
{
- items.push_back(sink);
+ items.push_back(sink);
}
if (source)
{
- items.push_back(source);
+ items.push_back(source);
}
if (items.size() == 0)
{
- return;
+ return;
}
- try
- {
- RtpStateReplicatorPrx oneway = RtpStateReplicatorPrx::uncheckedCast(mImpl->mStateReplicator->ice_oneway());
- oneway->setState(items);
- }
- catch (...)
- {
- mImpl->mStateReplicator->setState(items);
- }
+ mImpl->mReplicationContext->getReplicator().tryOneWay()->setState(items);
}
/**
@@ -452,7 +434,7 @@ void RTPSessionImpl::removeState(const RtpSessionStateItemPtr& session, const Rt
const RtpStreamSourceStateItemPtr& source)
{
// If state replication has been disabled do nothing
- if (!mImpl->mStateReplicator || mImpl->mReplicaService->isActive() == false)
+ if (mImpl->mReplicationContext->isReplicating() == false)
{
return;
}
@@ -461,31 +443,23 @@ void RTPSessionImpl::removeState(const RtpSessionStateItemPtr& session, const Rt
if (session)
{
- items.push_back(session->key);
+ items.push_back(session->key);
}
if (sink)
{
- items.push_back(sink->key);
+ items.push_back(sink->key);
}
if (source)
{
- items.push_back(source->key);
+ items.push_back(source->key);
}
if (items.size() == 0)
{
- return;
+ return;
}
- try
- {
- RtpStateReplicatorPrx oneway = RtpStateReplicatorPrx::uncheckedCast(mImpl->mStateReplicator->ice_oneway());
- oneway->removeState(items);
- }
- catch (...)
- {
- mImpl->mStateReplicator->removeState(items);
- }
+ mImpl->mReplicationContext->getReplicator().tryOneWay()->removeState(items);
}
diff --git a/src/RTPSession.h b/src/RTPSession.h
index 4ba6e86..b5a659a 100644
--- a/src/RTPSession.h
+++ b/src/RTPSession.h
@@ -11,6 +11,7 @@
#include <boost/shared_ptr.hpp>
#include <AsteriskSCF/Discovery/SmartProxy.h>
+#include "RtpReplicationContext.h"
#include "RTPConfiguration.h"
/**
@@ -44,13 +45,23 @@ typedef IceUtil::Handle<StreamSourceRTPImpl> StreamSourceRTPImplPtr;
class RTPSessionImpl : public AsteriskSCF::Media::RTP::V1::RTPSession
{
public:
- RTPSessionImpl(const Ice::ObjectAdapterPtr&, const AsteriskSCF::Media::RTP::V1::RTPServiceLocatorParamsPtr&,
- pj_pool_factory*, const AsteriskSCF::System::Component::V1::ReplicaPtr&,
- const AsteriskSCF::Discovery::SmartProxy<AsteriskSCF::Media::RTP::V1::RtpStateReplicatorPrx>&,
- const ConfigurationServiceImplPtr&);
- RTPSessionImpl(const Ice::ObjectAdapterPtr&, pj_pool_factory*, const Ice::Identity&, const Ice::Identity&,
- const Ice::Identity&, Ice::Int, const AsteriskSCF::Media::V1::FormatSeq&, bool,
- const ConfigurationServiceImplPtr&);
+ RTPSessionImpl(const Ice::ObjectAdapterPtr& adapter,
+ const AsteriskSCF::Media::RTP::V1::RTPServiceLocatorParamsPtr& locatorParams,
+ pj_pool_factory* factory,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr& configurationService);
+
+ RTPSessionImpl(const Ice::ObjectAdapterPtr& adapter,
+ pj_pool_factory* factory,
+ const Ice::Identity& sessionIdentity,
+ const Ice::Identity& sinkIdentity,
+ const Ice::Identity& sourceIdentity,
+ Ice::Int port,
+ const AsteriskSCF::Media::V1::FormatSeq& formats,
+ bool ipv6,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr& configurationService);
+
AsteriskSCF::Media::V1::StreamSourceSeq getSources(const Ice::Current&);
AsteriskSCF::Media::V1::StreamSinkSeq getSinks(const Ice::Current&);
std::string getId(const Ice::Current&);
diff --git a/src/RTPSource.cpp b/src/RTPSource.cpp
index 3bc22c1..e629d5c 100644
--- a/src/RTPSource.cpp
+++ b/src/RTPSource.cpp
@@ -278,4 +278,4 @@ void StreamSourceRTPImpl::setRemoteDetails(const string& address, Ice::Int port)
RtpStreamSourceStateItemPtr StreamSourceRTPImpl::getStateItem()
{
return mImpl->mSourceStateItem;
-}
+}
\ No newline at end of file
diff --git a/src/RtpReplicationContext.h b/src/RtpReplicationContext.h
new file mode 100644
index 0000000..4fa6e9d
--- /dev/null
+++ b/src/RtpReplicationContext.h
@@ -0,0 +1,78 @@
+/*
+ * Asterisk SCF -- An open-source communications framework.
+ *
+ * Copyright (C) 2010-2011, Digium, Inc.
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk SCF project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE.txt file
+ * at the top of the source tree.
+ */
+#pragma once
+
+#include <boost/shared_ptr.hpp>
+
+#include <AsteriskSCF/Discovery/SmartProxy.h>
+#include <AsteriskSCF/Replication/ReplicationContext.h>
+#include <AsteriskSCF/Component/TestContext.h>
+
+#include "RtpStateReplicationIf.h"
+
+typedef AsteriskSCF::Discovery::SmartProxy<AsteriskSCF::Media::RTP::V1::RtpStateReplicatorPrx> ReplicatorSmartPrx;
+
+/**
+ * This class provides the component's classes with the context needed to perform replication.
+ */
+class RtpReplicationContext : public AsteriskSCF::Replication::ReplicationContext
+{
+public:
+ RtpReplicationContext(AsteriskSCF::Replication::ReplicationStateType state) :
+ AsteriskSCF::Replication::ReplicationContext(state)
+ {
+ }
+
+ // Override
+ virtual bool isReplicating()
+ {
+ // If the base context says we aren't replicating, we aren't.
+ if (!ReplicationContext::isReplicating())
+ {
+ return false;
+ }
+
+ // Do we have a replicator proxy?
+ if (mReplicator.isInitialized())
+ {
+ return true;
+ }
+
+ return false;
+ }
+
+ /**
+ * Get a reference to the state replicator service.
+ */
+ ReplicatorSmartPrx getReplicator()
+ {
+ return mReplicator;
+ }
+
+ /**
+ * Sets the reference to the state replicator service.
+ */
+ void setReplicator(const ReplicatorSmartPrx& replicator)
+ {
+ mReplicator = replicator;
+ }
+
+private:
+ ReplicatorSmartPrx mReplicator;
+};
+typedef boost::shared_ptr<RtpReplicationContext> RtpReplicationContextPtr;
+
+
diff --git a/src/RtpStateReplicator.h b/src/RtpStateReplicator.h
index b244dda..0a461fe 100644
--- a/src/RtpStateReplicator.h
+++ b/src/RtpStateReplicator.h
@@ -24,9 +24,9 @@
#include <boost/shared_ptr.hpp>
+#include "RtpReplicationContext.h"
#include "RTPConfiguration.h"
-
typedef AsteriskSCF::Replication::StateReplicator<
AsteriskSCF::Media::RTP::V1::RtpStateReplicator,
AsteriskSCF::Media::RTP::V1::RtpStateItemPtr,
@@ -42,9 +42,11 @@ struct RtpStateReplicatorListenerImpl;
class RtpStateReplicatorListenerI : public AsteriskSCF::Media::RTP::V1::RtpStateReplicatorListener
{
public:
- RtpStateReplicatorListenerI(const Ice::ObjectAdapterPtr&, pj_pool_factory*,
- const AsteriskSCF::Media::RTP::V1::RtpGeneralStateItemPtr&,
- const ConfigurationServiceImplPtr&);
+ RtpStateReplicatorListenerI(const Ice::ObjectAdapterPtr&,
+ pj_pool_factory*,
+ const AsteriskSCF::Media::RTP::V1::RtpGeneralStateItemPtr&,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr&);
~RtpStateReplicatorListenerI();
void stateRemoved(const Ice::StringSeq&, const Ice::Current&);
void stateSet(const AsteriskSCF::Media::RTP::V1::RtpStateItemSeq&, const Ice::Current&);
diff --git a/src/RtpStateReplicatorApp.cpp b/src/RtpStateReplicatorApp.cpp
index ee6e851..3d93864 100644
--- a/src/RtpStateReplicatorApp.cpp
+++ b/src/RtpStateReplicatorApp.cpp
@@ -152,7 +152,7 @@ void RtpStateReplicatorService::registerWithServiceLocator(const Ice::Communicat
if (mServiceLocatorManagement == 0)
{
- lg(Error) << "Unable to obtain proxy to ServiceLocatorManagement interface. Check config file. "
+ lg(Error) << "Unable to obtain proxy to LocatorServiceManagement interface. Check config file. "
"This component can't be found until this is corrected.";
return;
}
diff --git a/src/RtpStateReplicatorListener.cpp b/src/RtpStateReplicatorListener.cpp
index 4f3cef1..bab64f1 100644
--- a/src/RtpStateReplicatorListener.cpp
+++ b/src/RtpStateReplicatorListener.cpp
@@ -54,10 +54,13 @@ private:
struct RtpStateReplicatorListenerImpl
{
public:
- RtpStateReplicatorListenerImpl(const Ice::ObjectAdapterPtr& adapter, pj_pool_factory *poolFactory,
- const RtpGeneralStateItemPtr& generalState,
- const ConfigurationServiceImplPtr& configurationService)
+ RtpStateReplicatorListenerImpl(const Ice::ObjectAdapterPtr& adapter,
+ pj_pool_factory *poolFactory,
+ const RtpGeneralStateItemPtr& generalState,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr& configurationService)
: mId(IceUtil::generateUUID()), mAdapter(adapter), mPoolFactory(poolFactory), mGeneralState(generalState),
+ mReplicationContext(replicationContext),
mConfigurationService(configurationService) {}
void removeStateNoticeImpl(const Ice::StringSeq& itemKeys)
@@ -98,9 +101,10 @@ public:
mImpl->mStateItems.insert(make_pair(item->mSessionId, newitem));
RTPSessionImplPtr localSession =
- new RTPSessionImpl(mImpl->mAdapter, mImpl->mPoolFactory, item->mSessionIdentity,
- item->mSinkIdentity, item->mSourceIdentity, item->mPort, item->mFormats, item->mIPv6,
- mImpl->mConfigurationService);
+ new RTPSessionImpl(mImpl->mAdapter, mImpl->mPoolFactory, item->mSessionIdentity,
+ item->mSinkIdentity, item->mSourceIdentity, item->mPort, item->mFormats, item->mIPv6,
+ mImpl->mReplicationContext,
+ mImpl->mConfigurationService);
localitem->setSession(localSession);
}
else
@@ -147,13 +151,15 @@ public:
Ice::ObjectAdapterPtr mAdapter;
pj_pool_factory *mPoolFactory;
RtpGeneralStateItemPtr mGeneralState;
+ RtpReplicationContextPtr mReplicationContext;
ConfigurationServiceImplPtr mConfigurationService;
};
RtpStateReplicatorListenerI::RtpStateReplicatorListenerI(const Ice::ObjectAdapterPtr& adapter,
pj_pool_factory *poolFactory, const RtpGeneralStateItemPtr& generalState,
+ const RtpReplicationContextPtr& replicationContext,
const ConfigurationServiceImplPtr& configurationService)
- : mImpl(new RtpStateReplicatorListenerImpl(adapter, poolFactory, generalState, configurationService)) {}
+ : mImpl(new RtpStateReplicatorListenerImpl(adapter, poolFactory, generalState, replicationContext, configurationService)) {}
RtpStateReplicatorListenerI::~RtpStateReplicatorListenerI()
{
-----------------------------------------------------------------------
--
asterisk-scf/integration/media_rtp_pjmedia.git
More information about the asterisk-scf-commits
mailing list