[asterisk-scf-commits] asterisk-scf/release/routing.git branch "master" updated.
Commits to the Asterisk SCF project code repositories
asterisk-scf-commits at lists.digium.com
Fri May 6 10:23:51 CDT 2011
branch "master" has been updated
via 41d4c5414c87350f82cead4811cde33583ee68ba (commit)
from c41eaab05360240a5a46ce507f00fb0891ff5e23 (commit)
Summary of changes:
config/routing-state-replicator.conf | 18 +
config/routingtest-integ.config | 68 +-
local-slice/BasicRoutingStateReplicationIf.ice | 188 +++
src/BasicRoutingServiceApp.cpp | 578 ++++++++--
src/BasicRoutingStateReplicatorApp.cpp | 357 ++++++
src/CMakeLists.txt | 72 ++-
src/ConnectBridgedSessionsOperation.cpp | 183 +++
src/ConnectBridgedSessionsOperation.h | 97 ++
...nectBridgedSessionsWithDestinationOperation.cpp | 523 +++++++++
...onnectBridgedSessionsWithDestinationOperation.h | 179 +++
src/EndpointRegistry.cpp | 301 +++--
src/EndpointRegistry.h | 12 +-
src/OperationReplicaCache.cpp | 69 ++
src/OperationReplicaCache.h | 160 +++
src/ReplicationContext.cpp | 99 ++
src/ReplicationContext.h | 75 ++
src/RouteSessionOperation.cpp | 536 +++++++++
src/RouteSessionOperation.h | 181 +++
src/RoutingServiceEventPublisher.cpp | 44 +-
src/RoutingServiceEventPublisher.h | 12 +-
src/RoutingStateReplicatorListener.cpp | 307 +++++
src/RoutingStateReplicatorListener.cpp~HEAD | 215 ++++
src/RoutingStateReplicatorListener.h | 54 +
src/RoutingStateReplicatorListener.h~HEAD | 54 +
src/ScriptProcessor.h | 3 +
src/SessionListener.cpp | 263 +++++
src/SessionListener.h | 99 ++
src/SessionRouter.cpp | 1226 ++------------------
src/SessionRouter.h | 75 +-
src/SessionRouterOperation.cpp | 272 +++++
src/SessionRouterOperation.h | 400 +++++++
src/SimpleWorkQueue.cpp | 273 -----
src/SimpleWorkQueue.h | 58 -
src/TestContext.cpp | 82 ++
src/TestContext.h | 59 +
src/WorkQueue.h | 85 --
test/CMakeLists.txt | 4 +
test/MockEndpointLocator.cpp | 6 +
test/MockEndpointLocator.h | 4 +
test/SharedTestData.h | 11 +-
test/TestRouting.cpp | 496 ++++++--
41 files changed, 5850 insertions(+), 1948 deletions(-)
create mode 100644 config/routing-state-replicator.conf
create mode 100644 local-slice/BasicRoutingStateReplicationIf.ice
create mode 100644 src/BasicRoutingStateReplicatorApp.cpp
create mode 100644 src/ConnectBridgedSessionsOperation.cpp
create mode 100644 src/ConnectBridgedSessionsOperation.h
create mode 100644 src/ConnectBridgedSessionsWithDestinationOperation.cpp
create mode 100644 src/ConnectBridgedSessionsWithDestinationOperation.h
create mode 100644 src/OperationReplicaCache.cpp
create mode 100644 src/OperationReplicaCache.h
create mode 100644 src/ReplicationContext.cpp
create mode 100644 src/ReplicationContext.h
create mode 100644 src/RouteSessionOperation.cpp
create mode 100644 src/RouteSessionOperation.h
create mode 100644 src/RoutingStateReplicatorListener.cpp
create mode 100644 src/RoutingStateReplicatorListener.cpp~HEAD
create mode 100644 src/RoutingStateReplicatorListener.h
create mode 100644 src/RoutingStateReplicatorListener.h~HEAD
create mode 100644 src/SessionListener.cpp
create mode 100644 src/SessionListener.h
create mode 100644 src/SessionRouterOperation.cpp
create mode 100644 src/SessionRouterOperation.h
delete mode 100644 src/SimpleWorkQueue.cpp
delete mode 100644 src/SimpleWorkQueue.h
create mode 100644 src/TestContext.cpp
create mode 100644 src/TestContext.h
delete mode 100644 src/WorkQueue.h
- Log -----------------------------------------------------------------
commit 41d4c5414c87350f82cead4811cde33583ee68ba
Author: Ken Hunt <ken.hunt at digium.com>
Date: Wed Dec 22 13:43:11 2010 -0600
- Routing async support.
- Replication support.
diff --git a/config/routing-state-replicator.conf b/config/routing-state-replicator.conf
new file mode 100644
index 0000000..c1c67ed
--- /dev/null
+++ b/config/routing-state-replicator.conf
@@ -0,0 +1,18 @@
+# Adapter parameters for this component
+RoutingStateReplicator.Endpoints=default
+RoutingStateReplicator.ThreadPool.Size=4
+
+# A proxy to the IceStorm topic manager
+TopicManager.Proxy=AsteriskSCFIceStorm/TopicManager:default -p 10000
+
+# A proxy to the service locator management service
+LocatorServiceManagement.Proxy=LocatorServiceManagement:tcp -p 4422
+
+# A proxy to the service locator service
+LocatorService.Proxy=LocatorService:tcp -p 4411
+
+# The name of the state replicator
+RoutingStateReplicator.Name=default
+
+IceBox.InheritProperties = 1
+IceBox.Service.RoutingStateReplicator=RoutingStateReplicator:create
diff --git a/config/routingtest-integ.config b/config/routingtest-integ.config
index 74836c8..57dbcf1 100644
--- a/config/routingtest-integ.config
+++ b/config/routingtest-integ.config
@@ -1,15 +1,28 @@
# This is a configuration file a single process test of the Routing Service.
+Ice.Warn.UnknownProperties=0
+
+Ice.ThreadPool.Client.Size=4
+
+#Ice.Admin.Endpoints=tcp -p 10006
+#Ice.Admin.InstanceName=IceBox
+IceBox.InstanceName=IceBox
+IceBox.ServiceManager.Endpoints=tcp -p 10007
+
IceBox.InheritProperties=1
IceBox.Service.ServiceDiscovery=service_locator:create
-IceBox.Service.BasicRoutingService=BasicRoutingService:create
+IceBox.Service.RoutingService=BasicRoutingService:create
+IceBox.Service.RoutingService2=BasicRoutingService:create
+
# Boost Test arguments can be passed here.
-# IceBox.Service.RoutingTest=RoutingTest:create --log_level=all
-IceBox.Service.RoutingTest=RoutingTest:create
+IceBox.Service.RoutingTest = RoutingTest:create --log_level=all
+#IceBox.Service.RoutingTest = RoutingTest:create
+
+IceBox.Service.Replicator=BasicRoutingStateReplicator:create
-IceBox.LoadOrder=ServiceDiscovery,BasicRoutingService,RoutingTest
+IceBox.LoadOrder=ServiceDiscovery,Replicator,RoutingService,RoutingService2,RoutingTest
# Where to find the Service Locator manager. We need the Service Locator in order to be able to plug in to the Asterisk SCF system Discovery mechanisms.
LocatorServiceManagement.Proxy=LocatorServiceManagement:tcp -p 4422
@@ -26,12 +39,30 @@ BridgeManager.ServiceLocatorId=BridgeService
###########################################
# Routing Service properties
-BasicRoutingServiceAdapter.Endpoints=tcp -p 10050
-
-BasicRoutingServiceAdapter.ThreadPool.Size=4
-BasicRoutingServiceAdapter.ThreadPool.SizeMax=10
-BasicRoutingServiceAdapter.ThreadPool.SizeWarn=9
-
+RoutingService.Endpoints=tcp -p 10050
+RoutingService.ComponentService.Endpoints=tcp -p 10051
+RoutingService.ThreadPool.Size=8
+RoutingService.ThreadPool.SizeMax=14
+RoutingService.ThreadPool.SizeWarn=9
+RoutingService.Standby=no
+RoutingService.StateReplicatorName=Replicator
+
+RoutingService2.Endpoints=tcp -p 10052
+RoutingService2.ComponentService.Endpoints=tcp -p 10053
+RoutingService2.ThreadPool.Size=8
+RoutingService2.ThreadPool.SizeMax=14
+RoutingService2.ThreadPool.SizeWarn=9
+RoutingService2.Standby=yes
+RoutingService2.StateReplicatorName=Replicator
+
+Replicator.InstanceName=Replicator
+Replicator.Endpoints=default -p 10054
+Replicator.ComponentService.Endpoints=default -p 10055
+Replicator.ThreadPool.Size=8
+Replicator.ThreadPool.SizeMax=14
+Replicator.ThreadPool.SizeWarn=9
+
+# AsteriskSCF.RoutingService.logger=Debug
###########################################
# Test properties
@@ -40,17 +71,25 @@ BasicRoutingServiceAdapter.ThreadPool.SizeWarn=9
TestRoutingAdapterOut.Endpoints=tcp -p 10070
TestRoutingAdapterIn.Endpoints=tcp -p 10071
+# NOTE: For testing, we grab direct proxy to the same servant on
+# backup service.
+
# Where to look for the Routing Service LocatorRegistry interface
LocatorRegistry.Proxy=RoutingServiceLocatorRegistry:tcp -p 10050
+BackupLocatorRegistry.Proxy=RoutingServiceLocatorRegistry:tcp -p 10052
SessionRouter.Proxy=SessionRouter:tcp -p 10050
+BackupSessionRouter.Proxy=SessionRouter:tcp -p 10052
+
+Replica.Proxy=BasicRoutingServiceReplica:tcp -p 10050
+BackupReplica.Proxy=BasicRoutingServiceReplica:tcp -p 10052
-TestRoutingAdapterIn.ThreadPool.Size=4
-TestRoutingAdapterIn.ThreadPool.SizeMax=10
+TestRoutingAdapterIn.ThreadPool.Size=8
+TestRoutingAdapterIn.ThreadPool.SizeMax=14
TestRoutingAdapterIn.ThreadPool.SizeWarn=9
-TestRoutingAdapterOut.ThreadPool.Size=4
-TestRoutingAdapterOut.ThreadPool.SizeMax=10
+TestRoutingAdapterOut.ThreadPool.Size=8
+TestRoutingAdapterOut.ThreadPool.SizeMax=14
TestRoutingAdapterOut.ThreadPool.SizeWarn=9
##########################################
@@ -94,3 +133,4 @@ ServiceLocatorLocalAdapter.Endpoints=tcp -p 4412
# Logger configuration
LoggerAdapter.Endpoints=default
+AsteriskSCF.Logging.logger.AsteriskSCF=Error
diff --git a/local-slice/BasicRoutingStateReplicationIf.ice b/local-slice/BasicRoutingStateReplicationIf.ice
new file mode 100644
index 0000000..4ea55c1
--- /dev/null
+++ b/local-slice/BasicRoutingStateReplicationIf.ice
@@ -0,0 +1,188 @@
+/*
+ * 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 <Ice/BuiltinSequences.ice>
+#include <Ice/Identity.ice>
+#include <AsteriskSCF/Core/Discovery/ServiceLocatorIf.ice>
+#include <AsteriskSCF/SessionCommunications/SessionCommunicationsIf.ice>
+
+module AsteriskSCF
+{
+module BasicRoutingService
+{
+["suppress"]
+module V1
+{
+ const string StateReplicatorComponentCategory = "RoutingStateReplicatorComponent";
+ const string StateReplicatorDiscoveryCategory = "RoutingStateReplicator";
+
+ unsliceable class RoutingStateReplicatorParams extends AsteriskSCF::Core::Discovery::V1::ServiceLocatorParams
+ {
+ string name;
+ };
+
+ ///////////////////////////////////////////////////////////////////////
+ // These classes and interfaces implement the replication
+ // pattern of Asterisk SCF.
+
+ ["visitor"] local class RoutingStateItemVisitor
+ {
+ };
+
+ /**
+ * Base class for an item that will be replicated.
+ * The key will be unique among all state items stored in the
+ * state replicator.
+ */
+ ["visitor:RoutingStateItemVisitor"] class RoutingStateItem
+ {
+ string key;
+ };
+
+ sequence<RoutingStateItem> RoutingStateItemSeq;
+
+ /**
+ * Listener interface. Typically implemented by
+ * a routing service in standby mode.
+ */
+ interface RoutingStateReplicatorListener
+ {
+ void stateRemoved(Ice::StringSeq itemKeys);
+ void stateRemovedForItems(RoutingStateItemSeq items);
+ void stateSet(RoutingStateItemSeq items);
+ };
+
+ /**
+ * The state replicator interface.
+ */
+ interface RoutingStateReplicator
+ {
+ void addListener(RoutingStateReplicatorListener *listener);
+ void removeListener(RoutingStateReplicatorListener *listener);
+ void setState (RoutingStateItemSeq items);
+ void removeState(Ice::StringSeq items);
+ void removeStateForItems(RoutingStateItemSeq items);
+ idempotent RoutingStateItemSeq getState(Ice::StringSeq itemKeys);
+ idempotent RoutingStateItemSeq getAllState();
+ };
+
+ /**
+ * All transient operations will derive from this.
+ */
+ class OperationStateItem extends RoutingStateItem
+ {
+ string operationId;
+ };
+
+ ///////////////////////////////////////////////////////////////////////
+ // These state items represent the state transistions
+ // of the RouteSession operation.
+
+ /**
+ * Indicates the RouteSessionOperation started.
+ * The key (in the base state item) is the operationId of this
+ * operation + RouteSessionOpStartKeyMod
+ */
+ class RouteSessionOpStart extends OperationStateItem
+ {
+ AsteriskSCF::SessionCommunications::V1::Session *source;
+ string destination;
+ };
+ const string RouteSessionOpStartKeyMod = ".START";
+
+ /**
+ * Indicates the RouteSessionOperation has completed an AMI endpoint lookup().
+ * The key (in the base state item) is the operationId of this
+ * operation + RouteSessionOpWaitLookupKeyMod
+ */
+ class RouteSessionOpWaitLookupState extends OperationStateItem
+ {
+ AsteriskSCF::Core::Endpoint::V1::EndpointSeq endpoints;
+ };
+ const string RouteSessionOpWaitLookupKeyMod = ".WAITLOOKUP";
+
+ /**
+ * Indicates the RouteSessionOperation has created the bridge.
+ * The key (in the base state item) is the operationId of this
+ * operation + RouteSessionOpBridgingKeyMod
+ */
+ class RouteSessionOpBridgingState extends OperationStateItem
+ {
+ AsteriskSCF::SessionCommunications::V1::Bridge* bridge;
+ };
+ const string RouteSessionOpBridgingKeyMod = ".BRIDGING";
+
+ ///////////////////////////////////////////////////////////////////////
+ // These state items represent the state transistions of the
+ // ConnectBridgedSessionsWithDestination operation.
+
+ /**
+ * Indicates the ConnectBridgedSessionsWithDestinationOperation started.
+ * The key (in the base state item) is the operationId of this
+ * operation + ConnectBridgedSessionsWithDestStartKeyMod
+ */
+ class ConnectBridgedSessionsWithDestinationOpStart extends OperationStateItem
+ {
+ AsteriskSCF::SessionCommunications::V1::Session *sessionToReplace;
+ string destination;
+ };
+ const string ConnectBridgedSessionsWithDestStartKeyMod = ".START";
+
+ /**
+ * Indicates the ConnectBridgedSessionsWithDestinationOperation completed
+ * an AMI endpoint lookup().
+ * The key (in the base state item) is the operationId of this
+ * operation + ConnectBridgedSessionsWithDestWaitLookupKeyMod
+ */
+ class ConnectBridgedSessionsWithDestinationOpWaitLookupState extends OperationStateItem
+ {
+ AsteriskSCF::Core::Endpoint::V1::EndpointSeq endpoints;
+ AsteriskSCF::SessionCommunications::V1::SessionSeq remainingSessions;
+ };
+ const string ConnectBridgedSessionsWithDestWaitLookupKeyMod = ".WAITLOOKUP";
+
+ /**
+ * Indicates the ConnectBridgedSessionsWithDestinationOperation has modified the bridge.
+ * The key (in the base state item) is the operationId of this
+ * operation + ConnectBridgedSessionsWithDestBridgingKeyMod
+ */
+ class ConnectBridgedSessionsWithDestinationOpBridgingState extends OperationStateItem
+ {
+ AsteriskSCF::SessionCommunications::V1::Bridge* bridge;
+ };
+ const string ConnectBridgedSessionsWithDestBridgingKeyMod = ".BRIDGING";
+
+ ///////////////////////////////////////////////////////////////////////
+ // NOTE: There is no value in replicating the ConnectBridgedSessions
+ // operation. No intermediate results are obtained to cache.
+
+ ///////////////////////////////////////////////////////////////////////
+ // Endpoint locator state items
+
+ /**
+ * Represents an added endpoint locator.
+ * The key (in the base state item) is the locator id.
+ */
+ class EndpointLocatorState extends RoutingStateItem
+ {
+ AsteriskSCF::Core::Routing::V1::RegExSeq regExList;
+ AsteriskSCF::Core::Routing::V1::EndpointLocator *locator;
+ };
+
+}; //module V1
+}; //module BasicRouting
+}; //module Asterisk SCF
diff --git a/src/BasicRoutingServiceApp.cpp b/src/BasicRoutingServiceApp.cpp
index 867ad78..08e2572 100644
--- a/src/BasicRoutingServiceApp.cpp
+++ b/src/BasicRoutingServiceApp.cpp
@@ -1,7 +1,7 @@
/*
* Asterisk SCF -- An open-source communications framework.
*
- * Copyright (C) 2010, Digium, Inc.
+ * Copyright (C) 2010-2011, Digium, Inc.
*
* See http://www.asterisk.org for more information about
* the Asterisk SCF project. Please do not directly contact
@@ -19,19 +19,26 @@
#include <boost/thread.hpp>
#include <boost/shared_ptr.hpp>
-#include <AsteriskSCF/SmartProxy.h>
#include <AsteriskSCF/Core/Routing/RoutingIf.h>
+#include <AsteriskSCF/System/Component/ReplicaIf.h>
#include <AsteriskSCF/Core/Discovery/ServiceLocatorIf.h>
-#include <AsteriskSCF/SessionCommunications/SessionCommunicationsIf.h>
#include <AsteriskSCF/System/Component/ComponentServiceIf.h>
+#include <AsteriskSCF/SessionCommunications/SessionCommunicationsIf.h>
+
+#include <AsteriskSCF/SmartProxy.h>
#include <AsteriskSCF/Logger/IceLogger.h>
#include <AsteriskSCF/logger.h>
+#include "AsteriskSCF/Threading/SimpleWorkQueue.h"
+#include "RoutingStateReplicatorListener.h"
#include "LuaScriptProcessor.h"
#include "RoutingServiceEventPublisher.h"
#include "EndpointRegistry.h"
#include "RoutingAdmin.h"
#include "SessionRouter.h"
+#include "OperationReplicaCache.h"
+#include "ReplicationContext.h"
+#include "TestContext.h"
using namespace std;
using namespace AsteriskSCF::SessionCommunications::V1;
@@ -41,6 +48,7 @@ using namespace AsteriskSCF::Core::Discovery::V1;
using namespace AsteriskSCF::System::Component::V1;
using namespace AsteriskSCF::System::Logging;
using namespace AsteriskSCF::SmartProxy;
+using namespace AsteriskSCF::BasicRoutingService::V1;
namespace
{
@@ -53,45 +61,61 @@ namespace AsteriskSCF
namespace BasicRoutingService
{
+class ComponentServiceImpl;
+typedef ::IceUtil::Handle<ComponentServiceImpl> ComponentServiceImplPtr;
+
+class ReplicaManagement;
+typedef ::IceUtil::Handle<ReplicaManagement> ReplicaManagementPtr;
+
class BasicRoutingServiceApp : public IceBox::Service
{
public:
BasicRoutingServiceApp()
- : mDone(false),
+ : mImplementationId("BasicRoutingService"),
+ mDone(false),
mInitialized(false),
mRunning(false),
- mWorkQueue( new SimpleWorkQueue("SessionRouterWorkQueue", lg))
+ mPublishTestInterface(false),
+ mSessionContext(new SessionContext()),
+ mWorkQueue( new AsteriskSCF::Threading::SimpleWorkQueue("SessionRouterWorkQueue")),
+ mListeningToReplicator(false)
{
}
- ~BasicRoutingServiceApp()
- {
- // Smart pointers do your thing.
- mSessionRouter = 0;
- mAdminInteface = 0;
- mComponentService = 0;
- mEndpointRegistry = 0;
- mEventPublisher = 0;
- // XXX unnecessary.
- }
-
void resume();
void suspend();
-public: // Overrides of IceBox::Service
- virtual void start(const string& name, const Ice::CommunicatorPtr& ic, const Ice::StringSeq& args);
+ void activated();
+ void onStandby();
+ bool isActive();
+
+ ////// Overrides of IceBox::Service
+
+ virtual void start(const string& name, const ::Ice::CommunicatorPtr& ic, const ::Ice::StringSeq& args);
virtual void stop();
private:
+ void suspendService(bool shuttingDown);
void initialize();
- void registerWithServiceLocator();
- void deregisterFromServiceLocator();
- void setCategory(const Discovery::V1::ServiceManagementPrx& serviceManagement, const string& category);
-
+ void locateBridgeManager();
+ void locateStateReplicator();
+ void registerWithServiceLocator(bool includeComponentService);
+ void deregisterFromServiceLocator(bool includeComponentService);
+
+ void listenToStateReplicator();
+ void stopListeningToStateReplicator();
+
+ /**
+ * Get an impementation-specific ID for createing service discovery guids.
+ * This is not the same as the app name, which is configurable.
+ */
+ const std::string mImplementationId;
bool mDone;
bool mInitialized;
bool mRunning;
- boost::shared_ptr<SimpleWorkQueue> mWorkQueue;
+ bool mPublishTestInterface;
+ SessionContextPtr mSessionContext;
+ boost::shared_ptr<AsteriskSCF::Threading::SimpleWorkQueue> mWorkQueue;
std::string mAppName;
ServiceLocatorManagementPrx mServiceLocatorManagement;
@@ -105,20 +129,36 @@ private:
// Our published interfaces.
BasicSessionRouterPtr mSessionRouter;
RoutingServiceAdminPtr mAdminInteface;
- ComponentServicePtr mComponentService;
+ ComponentServiceImplPtr mComponentService;
+ ComponentTestPtr mComponentTest;
AsteriskSCF::SmartProxy::SmartProxy<BridgeManagerPrx> mBridgeManager;
- RoutingEventsPtr mEventPublisher;
+ RoutingServiceEventPublisherPtr mEventPublisher;
+ boost::shared_ptr<OperationReplicaCache> mOperationReplicaCache;
EndpointRegistryPtr mEndpointRegistry;
+ // Replication support
+ ReplicationContextPtr mReplicationContext;
+ ReplicaManagementPtr mReplicaManagement;
+ AsteriskSCF::SmartProxy::SmartProxy<RoutingStateReplicatorPrx> mStateReplicator;
+ RoutingStateReplicatorListenerPtr mReplicatorListener;
+
+ RoutingStateReplicatorListenerPrx mReplicatorListenerProxy;
+ bool mListeningToReplicator;
+
// Implementation
- Ice::ObjectAdapterPtr mAdapter;
- Ice::CommunicatorPtr mCommunicator;
+ ::Ice::ObjectAdapterPtr mAdapter;
+ ::Ice::ObjectAdapterPtr mComponentServiceAdapter;
+ ::Ice::CommunicatorPtr mCommunicator;
+
+ boost::shared_mutex mReplicatorLock;
};
+typedef ::IceUtil::Handle<BasicRoutingServiceApp> BasicRoutingServiceAppPtr;
static const string RegistryLocatorObjectId("RoutingServiceLocatorRegistry");
static const string RoutingAdminObjectId("RoutingAdmin");
static const string ComponentServiceId("BasicRoutingComponent");
static const string SessionRouterObjectId("SessionRouter");
+static const string ReplicaServiceId("BasicRoutingServiceReplica");
/**
* This class provides implementation for the ComponentService interface.
@@ -127,36 +167,227 @@ static const string SessionRouterObjectId("SessionRouter");
class ComponentServiceImpl : public ComponentService
{
public:
- ComponentServiceImpl(BasicRoutingServiceApp &app) :
+ ComponentServiceImpl(BasicRoutingServiceApp* app) :
mApp(app)
{
}
public: // Overrides of the ComponentService interface.
- void suspend(const Ice::Current&)
+ void suspend(const ::Ice::Current&)
+ {
+ mApp->suspend();
+ }
+
+ void resume(const ::Ice::Current&)
+ {
+ mApp->resume();
+ }
+
+ void shutdown(const ::Ice::Current&)
+ {
+ mApp->stop();
+ }
+
+ void shutdownNotice()
+ {
+ mApp = 0;
+ }
+
+private:
+ BasicRoutingServiceAppPtr mApp;
+};
+
+/**
+ * This interface is published as a facet of the Component Service interface.
+ * It exists to allow more elaborate interaction with a component during testing.
+ */
+class ComponentTestImpl : public ComponentTest
+{
+public:
+ ComponentTestImpl(const TestContextPtr& testContext) :
+ mTestContext(testContext)
+ {
+ }
+
+ void setTestMode(const ::std::string& mode, const ::Ice::Current&)
+ {
+ mTestContext->setTestMode(mode);
+ }
+
+ void setTestModeWithArgs(const ::std::string& mode,
+ const ComponentTestParamSeq&,
+ const ::Ice::Current&)
{
- mApp.suspend();
+ mTestContext->setTestMode(mode);
}
- void resume(const Ice::Current&)
+ void clearTestMode(const string&, const ::Ice::Current&)
{
- mApp.resume();
+ mTestContext->clearTestMode();
}
- void shutdown(const Ice::Current&)
+ void clearTestModes(const ::Ice::Current&)
{
- mApp.stop();
+ mTestContext->clearTestMode();
}
private:
- BasicRoutingServiceApp& mApp;
+ TestContextPtr mTestContext;
};
/**
+ * This class provides implementation for this component's Replica management interface.
+ */
+class ReplicaManagement : public Replica
+{
+public:
+ /**
+ * Constructor.
+ * @param app
+ * @param adapter The adapter is assumed to have been activated.
+ */
+ ReplicaManagement(BasicRoutingServiceApp* app, ::Ice::ObjectAdapterPtr adapter) : mApp(app), mAdapter(adapter)
+ {
+ }
+
+ bool isActive(const ::Ice::Current&)
+ {
+ return mApp->isActive();
+ }
+
+ bool activate(const ::Ice::Current&)
+ {
+ mApp->activated();
+
+ 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&)
+ {
+ mApp->onStandby();
+
+ 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:
+ BasicRoutingServiceAppPtr mApp;
+
+ /**
+ * 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 BasicRoutingServiceApp::isActive()
+{
+ // The Replication Context is tracking our current status.
+ return mReplicationContext->isComponentActive();
+}
+
+void BasicRoutingServiceApp::activated()
+{
+ mReplicationContext->setComponentActive();
+ stopListeningToStateReplicator();
+}
+
+void BasicRoutingServiceApp::onStandby()
+{
+ mReplicationContext->setComponentStandby();
+ listenToStateReplicator();
+}
+
+/**
+ * Register as a listener to our state replicator.
+ * A component in standby mode will do this to monitor state changes
+ * being sent from an active component.
+ */
+void BasicRoutingServiceApp::listenToStateReplicator()
+{
+ boost::unique_lock<boost::shared_mutex> lock(mReplicatorLock);
+
+ if ((mReplicationContext->getReplicatorService() == 0) || (mListeningToReplicator == true))
+ {
+ return;
+ }
+
+ try
+ {
+ // Are we in standby mode?
+ if (mReplicationContext->isComponentActive() == false)
+ {
+ AsteriskSCF::BasicRoutingService::V1::RoutingStateReplicatorPrx oneWayStateReplicator =
+ AsteriskSCF::BasicRoutingService::V1::RoutingStateReplicatorPrx::uncheckedCast(mReplicationContext->getReplicatorService()->ice_oneway());
+
+ oneWayStateReplicator->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 BasicRoutingServiceApp::stopListeningToStateReplicator()
+{
+ boost::unique_lock<boost::shared_mutex> lock(mReplicatorLock);
+
+ if ((mReplicationContext->getReplicatorService() == 0) || (mListeningToReplicator == false))
+ {
+ return;
+ }
+
+ try
+ {
+ AsteriskSCF::BasicRoutingService::V1::RoutingStateReplicatorPrx oneWayStateReplicator =
+ AsteriskSCF::BasicRoutingService::V1::RoutingStateReplicatorPrx::uncheckedCast(mReplicationContext->getReplicatorService()->ice_oneway());
+
+ oneWayStateReplicator->removeListener(mReplicatorListenerProxy);
+
+ mListeningToReplicator = false;
+ }
+ catch (const Ice::Exception& e)
+ {
+ lg(Error) << e.what();
+ throw;
+ }
+}
+
+/**
* Helper function to add some parameters to one of our registered interfaces in the ServiceLocator, so that
* other components can look up our interfaces.
*/
-void BasicRoutingServiceApp::setCategory(const Discovery::V1::ServiceManagementPrx& serviceManagement, const string& category)
+void setCategory(const Discovery::V1::ServiceManagementPrx& serviceManagement, const string& category)
{
// Add category as a parameter to enable other components look this component up.
ServiceLocatorParamsPtr genericparams = new ServiceLocatorParams;
@@ -167,56 +398,64 @@ void BasicRoutingServiceApp::setCategory(const Discovery::V1::ServiceManagementP
/**
* Register this component's primary public interfaces with the Service Locator.
* This enables other Asterisk SCF components to locate the interfaces we are publishing.
+ *
+ * @param includeComponentService If true, registers our ComponentService interface
+ * in addition to all our other interfaces. While it is not uncommon to need to
+ * re-register most of our interfaces due to a pause()/resume(), we only deregister
+ * the ComponentService during shutdown. So this variable would typically be true only
+ * during initial startup.
*/
-void BasicRoutingServiceApp::registerWithServiceLocator()
+void BasicRoutingServiceApp::registerWithServiceLocator(bool includeComponentService)
{
try
{
- // Get a proxy to the management interface for the Service Locator, so we can add ourselves into the system discovery mechanisms.
- mServiceLocatorManagement = ServiceLocatorManagementPrx::checkedCast(mCommunicator->propertyToProxy("LocatorServiceManagement.Proxy"));
-
- if (mServiceLocatorManagement == 0)
- {
- lg(Error) << "Unable to obtain proxy to ServiceLocatorManagement interface. Check config file. This component can't be found until this is corrected.";
- return;
- }
-
// Register our RoutingAdmin interface with the Service Locator.
- Ice::ObjectPrx adminObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(RoutingAdminObjectId));
+ ::Ice::ObjectPrx adminObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(RoutingAdminObjectId));
RoutingServiceAdminPrx adminPrx = RoutingServiceAdminPrx::checkedCast(adminObjectPrx);
- string adminServiceGuid("BasicRoutingServiceAdmin"); // Should be unique for reporting.
+ string adminServiceGuid(mImplementationId + "." + Routing::V1::RoutingServiceAdminDiscoveryCategory); // Should be unique for reporting.
mAdminManagement = ServiceManagementPrx::uncheckedCast(mServiceLocatorManagement->addService(adminPrx, adminServiceGuid));
setCategory(mAdminManagement, Routing::V1::RoutingServiceAdminDiscoveryCategory);
// Register our RegistryLocator interface with the Service Locator.
- Ice::ObjectPrx locatorObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(RegistryLocatorObjectId));
+ ::Ice::ObjectPrx locatorObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(RegistryLocatorObjectId));
LocatorRegistryPrx locatorRegistryPrx = LocatorRegistryPrx::checkedCast(locatorObjectPrx);
- string locatorServiceGuid("BasicRoutingServiceRegistryLocator"); // Should be unique for reporting.
+ string locatorServiceGuid(mImplementationId + "." + Routing::V1::RoutingServiceLocatorRegistryDiscoveryCategory); // Should be unique for reporting.
mRegistryLocatorManagement = ServiceManagementPrx::uncheckedCast(mServiceLocatorManagement->addService(locatorRegistryPrx, locatorServiceGuid));
setCategory(mRegistryLocatorManagement, Routing::V1::RoutingServiceLocatorRegistryDiscoveryCategory);
- // Register the ComponentService interface with the Service Locator.
- Ice::ObjectPrx componentServiceObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(ComponentServiceId));
- ComponentServicePrx componentServicePrx = ComponentServicePrx::checkedCast(componentServiceObjectPrx);
- string componentServiceGuid("BasicRoutingService"); // Should be unique for reporting.
- mComponentServiceManagement = ServiceManagementPrx::uncheckedCast(mServiceLocatorManagement->addService(componentServicePrx, componentServiceGuid));
-
- setCategory(mComponentServiceManagement, Routing::V1::ComponentServiceDiscoveryCategory);
-
// Register the SessionRouter interface with the Service Locator.
- Ice::ObjectPrx sessionRouterObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(SessionRouterObjectId));
- AsteriskSCF::SessionCommunications::V1::SessionRouterPrx sessionRouterPrx = AsteriskSCF::SessionCommunications::V1::SessionRouterPrx::checkedCast(sessionRouterObjectPrx);
- string sessionRouterGuid("SessionRouter"); // Should be unique
+ ::Ice::ObjectPrx sessionRouterObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(SessionRouterObjectId));
+ AsteriskSCF::SessionCommunications::V1::SessionRouterPrx sessionRouterPrx =
+ AsteriskSCF::SessionCommunications::V1::SessionRouterPrx::checkedCast(sessionRouterObjectPrx);
+ string sessionRouterGuid(mImplementationId + "." + Routing::V1::SessionRouterDiscoveryCategory); // Should be unique
mSessionRouterManagement = ServiceManagementPrx::uncheckedCast(mServiceLocatorManagement->addService(sessionRouterPrx, sessionRouterGuid));
setCategory(mSessionRouterManagement, Routing::V1::SessionRouterDiscoveryCategory);
+ if (includeComponentService)
+ {
+ // Register the ComponentService interface with the Service Locator.
+ // Note that this interface has it's own adapter.
+ ::Ice::ObjectPrx componentServiceObjectPrx = mComponentServiceAdapter->createDirectProxy(mCommunicator->stringToIdentity(ComponentServiceId));
+ ComponentServicePrx componentServicePrx = ComponentServicePrx::checkedCast(componentServiceObjectPrx);
+ string componentServiceGuid(mImplementationId + "." + Routing::V1::ComponentServiceDiscoveryCategory); // Should be unique for reporting.
+ mComponentServiceManagement = ServiceManagementPrx::uncheckedCast(mServiceLocatorManagement->addService(componentServicePrx, componentServiceGuid));
+
+ setCategory(mComponentServiceManagement, Routing::V1::ComponentServiceDiscoveryCategory);
+
+ if (mPublishTestInterface)
+ {
+ // Register our test servant as a facet of the ComponentService interface.
+ mComponentServiceAdapter->addFacet(mComponentTest, componentServicePrx->ice_getIdentity(), AsteriskSCF::System::Component::V1::ComponentTestFacet);
+ }
+ }
}
- catch(...)
+ catch(const std::exception& e)
{
- lg(Error) << "Major problems in " << mAppName << " registerWithServiceLocator()";
+ lg(Error) << "Unable to publish component interfaces in " << mAppName << BOOST_CURRENT_FUNCTION << ". Exception: " << e.what();
+ throw; // rethrow
}
}
@@ -224,15 +463,23 @@ void BasicRoutingServiceApp::registerWithServiceLocator()
* Deregister this component's primary public interfaces from the Service Locator.
* This is done at shutdown, and whenever we want to keep other services from locating
* our interfaces.
+ *
+ * @param includeComponentService If true, deregisters our ComponentService interface
+ * in addition to all our other interfaces, making this component unreachable from the
+ * rest of Asterisk SCF. Should only be done during shutdown.
*/
-void BasicRoutingServiceApp::deregisterFromServiceLocator()
+void BasicRoutingServiceApp::deregisterFromServiceLocator(bool includeComponentService)
{
try
{
mRegistryLocatorManagement->unregister();
mAdminManagement->unregister();
- mComponentServiceManagement->unregister();
mSessionRouterManagement->unregister();
+
+ if (includeComponentService)
+ {
+ mComponentServiceManagement->unregister();
+ }
}
catch(...)
{
@@ -241,6 +488,47 @@ void BasicRoutingServiceApp::deregisterFromServiceLocator()
}
/**
+ * Locate the BridgeManager using the Service Locator.
+ */
+void BasicRoutingServiceApp::locateBridgeManager()
+{
+ try
+ {
+ mBridgeManager = AsteriskSCF::SmartProxy::SmartProxy<BridgeManagerPrx>(
+ mServiceLocator,
+ new ServiceLocatorParams(BridgeServiceDiscoveryCategory),
+ lg);
+ }
+ catch(const AsteriskSCF::Core::Discovery::V1::ServiceNotFound&)
+ {
+ lg(Debug) << "BridgeManager not found. (Safe to ignore ServiceNotFound for briding during startup.)";
+ }
+
+ mSessionContext->bridgeManager = mBridgeManager;
+}
+
+/**
+ * Locate our State Replicator using the Service Locator.
+ */
+void BasicRoutingServiceApp::locateStateReplicator()
+{
+ BasicRoutingService::V1::RoutingStateReplicatorParamsPtr replicatorParams = new BasicRoutingService::V1::RoutingStateReplicatorParams();
+ replicatorParams->category = BasicRoutingService::V1::StateReplicatorDiscoveryCategory;
+ replicatorParams->name = mCommunicator->getProperties()->getPropertyWithDefault(mAppName + ".StateReplicatorName", "default");
+
+ try
+ {
+ AsteriskSCF::SmartProxy::SmartProxy<RoutingStateReplicatorPrx> replicator(mServiceLocator, replicatorParams, lg);
+ mReplicationContext->setReplicatorService(replicator);
+ }
+ catch(const AsteriskSCF::Core::Discovery::V1::ServiceNotFound&)
+ {
+ lg(Error) << "StateReplicator not found. Check configuration.";
+ throw;
+ }
+}
+
+/**
* Create the primary functional objects of this component.
* @param appName Name of the application or component.
*/
@@ -248,8 +536,18 @@ void BasicRoutingServiceApp::initialize()
{
try
{
- // Create the adapter.
- mAdapter = mCommunicator->createObjectAdapter("BasicRoutingServiceAdapter");
+ // Create the primary adapter.
+ mAdapter = mCommunicator->createObjectAdapter(mAppName);
+
+ // Create a separate adapter just for the component service. This is
+ // to make it easy to deactivate all of our interfaces other than the
+ // component service for suspend()/resume().
+ mComponentServiceAdapter = mCommunicator->createObjectAdapter(mAppName + ".ComponentService");
+
+ bool active = !(mCommunicator->getProperties()->getPropertyWithDefault(mAppName + ".Standby", "no") == "yes");
+
+ // Create the replication context.
+ mReplicationContext = ReplicationContextPtr(new ReplicationContext(active));
mEventPublisher = new RoutingServiceEventPublisher(mAdapter);
@@ -258,57 +556,90 @@ void BasicRoutingServiceApp::initialize()
getLoggerFactory().setLogOutput(mIceLogger->getLogger());
// Create and configure the EndpointRegistry.
- ScriptProcessor* scriptProcesor(new LuaScriptProcessor());
- mEndpointRegistry = new EndpointRegistry(scriptProcesor, mEventPublisher);
+ mEndpointRegistry = new EndpointRegistry(ScriptProcessorPtr(new LuaScriptProcessor()), mEventPublisher, mReplicationContext);
// Publish the LocatorRegistry interface.
mAdapter->add(mEndpointRegistry, mCommunicator->stringToIdentity(RegistryLocatorObjectId));
+ // Check to see if we're configured to publish a test interface.
+ mPublishTestInterface = (mCommunicator->getProperties()->getPropertyWithDefault(mAppName + ".ComponentTest", "no") == "yes");
+
+ if (mPublishTestInterface)
+ {
+ // Create and publish the ComponentTest servant. This will be exposed
+ // as a facet of ComponentService.
+ mComponentTest = new ComponentTestImpl(mReplicationContext->getTestContext());
+ }
+
+ // Create the session context needed to construct operations.
+ mSessionContext = SessionContextPtr(new SessionContext(mAdapter,
+ mEndpointRegistry,
+ mEventPublisher,
+ mWorkQueue,
+ mReplicationContext));
+
+ // Create the replica cache.
+ mOperationReplicaCache = OperationReplicaCachePtr(new OperationReplicaCache(mSessionContext));
+
// Create publish the SessionRouter interface.
- SessionRouter *rawSessionRouter(new SessionRouter(mAdapter, mEndpointRegistry, mEventPublisher, mWorkQueue));
- BasicSessionRouterPtr basicSessionPtr(rawSessionRouter);
- mSessionRouter = basicSessionPtr;
- mAdapter->add(rawSessionRouter, mCommunicator->stringToIdentity(SessionRouterObjectId));
+ mSessionRouter = new SessionRouter(mSessionContext, mOperationReplicaCache);
+ mAdapter->add(mSessionRouter, mCommunicator->stringToIdentity(SessionRouterObjectId));
// Create and publish the Admin interface support.
mAdminInteface = new RoutingAdmin(mEndpointRegistry);
mAdapter->add(mAdminInteface, mCommunicator->stringToIdentity(RoutingAdminObjectId));
- // Create and publish the ComponentService interface.
- mComponentService = new ComponentServiceImpl(*this);
- mAdapter->add(mComponentService, mCommunicator->stringToIdentity(ComponentServiceId));
+ // Create and publish the ComponentService interface on it's own dedicated adapter.
+ mComponentService = new ComponentServiceImpl(this);
+ mComponentServiceAdapter->add(mComponentService, mCommunicator->stringToIdentity(ComponentServiceId));
+
+ // Create and publish our Replica interface support. This interface allows this component
+ // to be activated or placed in standby mode.
+ mReplicaManagement = new ReplicaManagement(this, mAdapter);
+ mAdapter->add(mReplicaManagement, mCommunicator->stringToIdentity(ReplicaServiceId));
+
+ // Create and publish our state replicator listener interface.
+ mReplicatorListener = new RoutingStateReplicatorListenerImpl(mEndpointRegistry, mOperationReplicaCache);
+ RoutingStateReplicatorListenerPrx replicatorListener = RoutingStateReplicatorListenerPrx::uncheckedCast(mAdapter->addWithUUID(mReplicatorListener));
+ mReplicatorListenerProxy = RoutingStateReplicatorListenerPrx::uncheckedCast(replicatorListener->ice_oneway());
+
+ if (active)
+ {
+ activated();
+ }
+ else
+ {
+ onStandby();
+ }
mAdapter->activate();
+ mComponentServiceAdapter->activate();
+
+ // Get a proxy to the management interface for the Service Locator manager.
+ // This isso we can add ourselves into the system discovery mechanisms.
+ mServiceLocatorManagement = ServiceLocatorManagementPrx::checkedCast(mCommunicator->propertyToProxy("LocatorServiceManagement.Proxy"));
// Get a proxy to the interface for the Service Locator.
+ // This is so we can find the other components we depend on.
mServiceLocator = ServiceLocatorPrx::checkedCast(mCommunicator->propertyToProxy("LocatorService.Proxy"));
mInitialized = true;
}
- catch(const Ice::Exception &exception)
+ catch(const ::Ice::Exception &e)
{
- lg(Error) << "Problems in " << mAppName << " initialization(): " << exception.what();
+ lg(Error) << "Problems in " << mAppName << BOOST_CURRENT_FUNCTION << e.what();
+ throw e;
}
- try
- {
- mBridgeManager = AsteriskSCF::SmartProxy::SmartProxy<BridgeManagerPrx>(
- mServiceLocator,
- new ServiceLocatorParams(BridgeServiceDiscoveryCategory),
- lg);
- }
- catch(const AsteriskSCF::Core::Discovery::V1::ServiceNotFound&)
- {
- lg(Debug) << "BridgeManager not found... safe to ignore ServiceNotFound during startup.";
- }
+ locateBridgeManager();
- mSessionRouter->setBridgeManager(mBridgeManager);
+ locateStateReplicator();
}
/**
* Implementation of the required IceBox::Service start method.
*/
-void BasicRoutingServiceApp::start(const string& name, const Ice::CommunicatorPtr& communicator, const Ice::StringSeq&)
+void BasicRoutingServiceApp::start(const string& name, const ::Ice::CommunicatorPtr& communicator, const ::Ice::StringSeq&)
{
lg(Info) << "Starting...";
@@ -322,11 +653,15 @@ void BasicRoutingServiceApp::start(const string& name, const Ice::CommunicatorPt
else
{
mAdapter->activate();
+ mComponentServiceAdapter->activate();
}
- // Plug back into the Asterisk SCF discovery system so that the interfaces we provide
+ // Plug into the Asterisk SCF discovery system so that the interfaces we provide
// can be located.
- registerWithServiceLocator();
+ registerWithServiceLocator(true);
+
+ // Register with the state replicator in case we are in standby mode.
+ listenToStateReplicator();
mRunning = true;
lg(Info) << "Started";
@@ -339,19 +674,51 @@ void BasicRoutingServiceApp::resume()
{
if (!mRunning)
{
- mAdapter->activate();
+ // If we're in standby by mode, listen for state replication.
+ listenToStateReplicator();
// Plug back into the Asterisk SCF discovery system so that the interfaces we provide
// can be located.
- registerWithServiceLocator();
+ registerWithServiceLocator(false);
+
+ // Reactivate the primary adapter.
+ mAdapter->activate();
}
mRunning = true;
}
+/**
+ * Utility function to suspend the service for a suspend() or stop().
+ */
+void BasicRoutingServiceApp::suspendService(bool shuttingDown)
+{
+ if (mRunning)
+ {
+ // Deregister our servants.
+ deregisterFromServiceLocator(shuttingDown);
+
+ // Remove our interfaces from the state replicator.
+ stopListeningToStateReplicator();
+
+ // Deactive the primary adapter.
+ // The adapter that services the ComponentService stays active.
+ mAdapter->deactivate();
+ }
+
+ mRunning = false;
+}
+
+/**
+ * Handle a notice from the ComponentService.
+ */
void BasicRoutingServiceApp::suspend()
{
- stop();
+ lg(Info) << "Suspending...";
+
+ suspendService(false);
+
+ lg(Info) << "Suspended.";
}
/**
@@ -361,28 +728,25 @@ void BasicRoutingServiceApp::stop()
{
lg(Info) << "Stopping...";
- if (mRunning)
- {
- deregisterFromServiceLocator();
- mAdapter->deactivate();
- }
+ suspendService(true);
- //
- // TODO: Needs to destroy communicator if it is not shared.
- //
+ // Just in case we were in suspend() mode when told to stop.
+ mComponentServiceManagement->unregister();
+
+ // Turn off our ComponentService interface.
+ // Only a start() directly from IceBox can restart us now.
+ mComponentServiceAdapter->deactivate();
- mRunning = false;
lg(Info) << "Stopped.";
}
-
} // end BasicRoutingService
} // end AsteriskSCF
extern "C"
{
-ASTERISK_SCF_ICEBOX_EXPORT IceBox::Service* create(Ice::CommunicatorPtr)
+ASTERISK_SCF_ICEBOX_EXPORT IceBox::Service* create(::Ice::CommunicatorPtr)
{
return new AsteriskSCF::BasicRoutingService::BasicRoutingServiceApp;
}
diff --git a/src/BasicRoutingStateReplicatorApp.cpp b/src/BasicRoutingStateReplicatorApp.cpp
new file mode 100644
index 0000000..31bdcc9
--- /dev/null
+++ b/src/BasicRoutingStateReplicatorApp.cpp
@@ -0,0 +1,357 @@
+/*
+ * 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 <Ice/Ice.h>
+#include <IceUtil/UUID.h>
+#include <IceStorm/IceStorm.h>
+#include <IceBox/IceBox.h>
+
+#include <AsteriskSCF/Core/Discovery/ServiceLocatorIf.h>
+#include <AsteriskSCF/System/Component/ComponentServiceIf.h>
+#include <AsteriskSCF/logger.h>
+#include <AsteriskSCF/Logger/IceLogger.h>
+#include <AsteriskSCF/StateReplicator.h>
+#include "BasicRoutingStateReplicationIf.h"
+
+using namespace std;
+using namespace AsteriskSCF::Core;
+using namespace AsteriskSCF::Core::Discovery::V1;
+using namespace AsteriskSCF::System::Component::V1;
+using namespace AsteriskSCF::System::Logging;
+using namespace AsteriskSCF::BasicRoutingService;
+using namespace AsteriskSCF::BasicRoutingService::V1;
+using namespace AsteriskSCF::Core::Routing::V1;
+
+namespace
+{
+Logger lg = getLoggerFactory().getLogger("AsteriskSCF.BasicRoutingServiceStateReplicator");
+}
+
+namespace AsteriskSCF
+{
+namespace BasicRoutingService
+{
+
+typedef AsteriskSCF::StateReplication::StateReplicator< AsteriskSCF::BasicRoutingService::V1::RoutingStateReplicator,
+ AsteriskSCF::BasicRoutingService::V1::RoutingStateItemPtr,
+ std::string,
+ AsteriskSCF::BasicRoutingService::V1::RoutingStateReplicatorListenerPrx> RoutingStateReplicatorI;
+typedef IceUtil::Handle<RoutingStateReplicatorI> RoutingStateReplicatorIPtr;
+
+};
+};
+
+/**
+ * This class implements a very simplistic replication capability using the StateReplicator template.
+ */
+class BasicRoutingStateReplicatorService : public IceBox::Service
+{
+public:
+ BasicRoutingStateReplicatorService() : mRunning(false), mInitialized(false) {};
+ ~BasicRoutingStateReplicatorService() {};
+
+ void suspend();
+ void resume();
+
+ ////// Overrides of IceBox::Service methods.
+ virtual void start(const string &name, const Ice::CommunicatorPtr& ic, const Ice::StringSeq& args);
+ virtual void stop();
+
+private:
+ void suspendService(bool shuttingDown);
+ void initialize(const std::string& appName);
+ void registerWithServiceLocator();
+ void deregisterFromServiceLocator(bool includeComponentService);
+ std::string mAppName;
+
+ bool mRunning;
+ bool mInitialized;
+ ::Ice::ObjectAdapterPtr mAdapter;
+ ::Ice::ObjectAdapterPtr mComponentServiceAdapter;
+ ::Ice::CommunicatorPtr mCommunicator;
+ ServiceLocatorManagementPrx mServiceLocatorManagement;
+ Discovery::V1::ServiceManagementPrx mComponentServiceManagement;
+ Discovery::V1::ServiceManagementPrx mStateReplicationManagement;
+ ConfiguredIceLoggerPtr mIceLogger;
+ AsteriskSCF::BasicRoutingService::RoutingStateReplicatorIPtr mStateReplicator;
+};
+typedef ::IceUtil::Handle<BasicRoutingStateReplicatorService> BasicRoutingStateReplicatorServicePtr;
+
+static const string ComponentServiceId("BasicRoutingStateReplicatorComponent");
+static const string ServiceDiscoveryId("BasicRoutingStateReplicatorService");
+
+/**
+ * This class provides implementation for the ComponentService interface, which
+ * every Asterisk SCF component is expected to publish.
+ */
+class ComponentServiceImpl : public ComponentService
+{
+public:
+ ComponentServiceImpl(BasicRoutingStateReplicatorService* service) : mService(service) {}
+
+public: // Overrides of the ComponentService interface.
+ virtual void suspend(const ::Ice::Current&)
+ {
+ mService->suspend();
+ }
+
+ virtual void resume(const ::Ice::Current&)
+ {
+ mService->resume();
+ }
+
+ virtual void shutdown(const ::Ice::Current&)
+ {
+ mService->stop();
+ }
+
+private:
+ BasicRoutingStateReplicatorServicePtr mService;
+};
+
+class BasicRoutingStateReplicatorCompare : public ServiceLocatorParamsCompare
+{
+public:
+ BasicRoutingStateReplicatorCompare(const std::string& name) : mName(name) {}
+ bool isSupported(const ServiceLocatorParamsPtr ¶ms, const Ice::Current &)
+ {
+ RoutingStateReplicatorParamsPtr routingParams = RoutingStateReplicatorParamsPtr::dynamicCast(params);
+ if (routingParams->name == mName)
+ {
+ return true;
+ }
+ return false;
+ }
+private:
+ string mName;
+};
+
+typedef IceUtil::Handle<BasicRoutingStateReplicatorCompare> BasicRoutingStateReplicatorComparePtr;
+
+/**
+ * Helper function to add some parameters to one of our registered interfaces in the ServiceLocator, so that
+ * other components can look up our interfaces.
+ */
+void setCategory(const Discovery::V1::ServiceManagementPrx& serviceManagement, const string& category)
+{
+ // Add category as a parameter to enable other components look this component up.
+ ServiceLocatorParamsPtr genericparams = new ServiceLocatorParams;
+ genericparams->category = category;
+ serviceManagement->addLocatorParams(genericparams, "");
+}
+
+/**
+ * Register this component's primary public interfaces with the Service Locator.
+ * This enables other Asterisk SCF components to locate our interfaces.
+ */
+void BasicRoutingStateReplicatorService::registerWithServiceLocator()
+{
+ try
+ {
+ // Get a proxy to our ComponentService interface and add it to the Service Locator.
+ // Note that this interface has its own adapter.
+ Ice::ObjectPrx componentServiceObjectPrx = mComponentServiceAdapter->createDirectProxy(mCommunicator->stringToIdentity(ComponentServiceId));
+ ComponentServicePrx componentServicePrx = ComponentServicePrx::checkedCast(componentServiceObjectPrx);
+
+ string componentServiceGuid(AsteriskSCF::BasicRoutingService::V1::StateReplicatorComponentCategory);
+ mComponentServiceManagement = ServiceManagementPrx::uncheckedCast(mServiceLocatorManagement->addService(componentServicePrx, componentServiceGuid));
+ setCategory(mComponentServiceManagement, AsteriskSCF::BasicRoutingService::V1::StateReplicatorComponentCategory);
+
+ // Get a proxy to our Replicator interface and add it to the Service Locator.
+ Ice::ObjectPrx stateReplicatorObjectPrx = mAdapter->createDirectProxy(mCommunicator->stringToIdentity(ServiceDiscoveryId));
+ RoutingStateReplicatorPrx stateReplicatorPrx = RoutingStateReplicatorPrx::checkedCast(stateReplicatorObjectPrx);
+
+ string stateReplicationGuid(AsteriskSCF::BasicRoutingService::V1::StateReplicatorDiscoveryCategory);
+ mStateReplicationManagement = ServiceManagementPrx::uncheckedCast(mServiceLocatorManagement->addService(stateReplicatorPrx, stateReplicationGuid));
+
+ ServiceLocatorParamsPtr discoveryParams = new ServiceLocatorParams();
+ discoveryParams->category = AsteriskSCF::BasicRoutingService::V1::StateReplicatorDiscoveryCategory;
+
+ string replicatorName = mCommunicator->getProperties()->getPropertyWithDefault(mAppName + ".InstanceName", "default");
+ ServiceLocatorParamsComparePrx compareProxy = ServiceLocatorParamsComparePrx::uncheckedCast(
+ mAdapter->addWithUUID(new BasicRoutingStateReplicatorCompare(replicatorName)));
+
+ string compareGuid = IceUtil::generateUUID();
+ mServiceLocatorManagement->addCompare(compareGuid, compareProxy);
+ mStateReplicationManagement->addLocatorParams(discoveryParams, compareGuid);
+ }
+ catch(...)
+ {
+ lg(Error) << "Exception in " << mAppName << BOOST_CURRENT_FUNCTION;
+ throw;
+ }
+}
+
+/**
+ * Deregister this component's primary public interfaces from the Service Locator.
+ * This is done at shutdown, and whenever we want to keep other services from locating
+ * our interfaces.
+ *
+ * @param includeComponentService If true, deregisters our ComponentService interface
+ * in addition to all our other interfaces, making this component unreachable from the
+ * rest of Asterisk SCF. Should only be done during shutdown.
+ */
+void BasicRoutingStateReplicatorService::deregisterFromServiceLocator(bool includeComponentService)
+{
+ try
+ {
+ mStateReplicationManagement->unregister();
+
+ if (includeComponentService)
+ {
+ mComponentServiceManagement->unregister();
+ }
+ }
+ catch(...)
+ {
+ lg(Error) << "Exception in " << BOOST_CURRENT_FUNCTION;
+ }
+}
+
+void BasicRoutingStateReplicatorService::initialize(const std::string& appName)
+{
+ mAppName = appName;
+
+ // This is the primary adapter for this component.
+ mAdapter = mCommunicator->createObjectAdapter(mAppName);
+
+ // Create a separate adapter just for the component service. This is
+ // to make it easy to deactivate all of our interfaces other than the
+ // component service to support suspend()/resume().
+ mComponentServiceAdapter = mCommunicator->createObjectAdapter(mAppName + ".ComponentService");
+
+ // setup logging client
+ mIceLogger = createIceLogger(mAdapter);
+ getLoggerFactory().setLogOutput(mIceLogger->getLogger());
+
+ // Create and publish our ComponentService interface support on its own adapter.
+ mComponentServiceAdapter->add(new ComponentServiceImpl(this), mCommunicator->stringToIdentity(ComponentServiceId));
+
+ // Create our instance of the StateReplicator template.
+ mStateReplicator = new RoutingStateReplicatorI();
+ mAdapter->add(mStateReplicator, mCommunicator->stringToIdentity(ServiceDiscoveryId));
+
+ // Activate our adapters.
+ mAdapter->activate();
+ mComponentServiceAdapter->activate();
+
+ // Get a proxy to the management interface for the Service Locator, so we can add
+ // ourselves into the system discovery mechanisms.
+ mServiceLocatorManagement = ServiceLocatorManagementPrx::checkedCast(mCommunicator->propertyToProxy("LocatorServiceManagement.Proxy"));
+
+ if (mServiceLocatorManagement == 0)
+ {
+ lg(Error) << "Unable to obtain proxy to ServiceLocatorManagement interface. Check config file. " << BOOST_CURRENT_FUNCTION;
+ }
+
+ mInitialized = true;
+}
+
+void BasicRoutingStateReplicatorService::start(const string &name, const Ice::CommunicatorPtr& ic, const Ice::StringSeq& )
+{
+ mCommunicator = ic;
+
+ if (!mInitialized)
+ {
+ initialize(name);
+ }
+ else
+ {
+ mAdapter->activate();
+ mComponentServiceAdapter->activate();
+ }
+
+ // Plug into the Asterisk SCF discovery system so that the interfaces we provide
+ // can be located.
+ registerWithServiceLocator();
+
+ mRunning = true;
+}
+
+void BasicRoutingStateReplicatorService::resume()
+{
+ if (!mRunning)
+ {
+ mAdapter->add(mStateReplicator, mCommunicator->stringToIdentity(ServiceDiscoveryId));
+
+ // Plug back into the Asterisk SCF discovery system so that the interfaces we provide
+ // can be located.
+ registerWithServiceLocator();
+
+ // Reactivate the primary adapter.
+ mAdapter->activate();
+ }
+
+ mRunning = true;
+}
+
+/**
+ * Utility function to suspend the service for a suspend() or stop().
+ */
+void BasicRoutingStateReplicatorService::suspendService(bool shuttingDown)
+{
+ if (mRunning)
+ {
+ // Deregister our servants.
+ deregisterFromServiceLocator(shuttingDown);
+
+ // Deactive the primary adapter.
+ // The adapter that services the ComponentService stays active.
+ mAdapter->deactivate();
+ }
+
+ mRunning = false;
+}
+
+/**
+ * Handle a notice from the ComponentService.
+ */
+void BasicRoutingStateReplicatorService::suspend()
+{
+ lg(Info) << "Suspending...";
+
+ suspendService(false);
+
+ lg(Info) << "Suspended.";
+}
+
+/**
+ * Implementation of the required IceBox::Service stop method.
+ */
+void BasicRoutingStateReplicatorService::stop()
+{
+ lg(Info) << "Stopping...";
+
+ suspendService(true);
+
+ // Just in case we were in suspend() mode when told to stop.
+ mComponentServiceManagement->unregister();
+
+ // Turn off our ComponentService interface.
+ // Only a start() directly from IceBox can restart us now.
+ mComponentServiceAdapter->deactivate();
+
+ lg(Info) << "Stopped.";
+}
+
+extern "C"
+{
+ASTERISK_SCF_ICEBOX_EXPORT IceBox::Service* create(Ice::CommunicatorPtr )
+{
+ return new BasicRoutingStateReplicatorService;
+}
+}
+
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 54a2557..79868d9 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -1,13 +1,22 @@
if(NOT logger_dir)
message(FATAL_ERROR "The logger directory could not be found ${logger_dir}")
endif()
+
include_directories(${logger_dir}/include)
include_directories(${API_INCLUDE_DIR})
include_directories(${utils_dir}/SmartProxy/include)
find_package(Lua51 REQUIRED)
include_directories(${LUA_INCLUDE_DIR})
+include_directories(${utils_dir}/AmiCollector/include)
+include_directories(${util_cpp_dir}/Threading/include)
+include_directories(${util_cpp_dir}/StateMachine/include)
asterisk_scf_component_init(BasicRoutingService)
+
+asterisk_scf_slice_include_directories(${API_SLICE_DIR})
+
+asterisk_scf_component_add_slice(BasicRoutingService ../local-slice/BasicRoutingStateReplicationIf.ice)
+
asterisk_scf_component_add_file(BasicRoutingService BasicRoutingServiceApp.cpp)
asterisk_scf_component_add_file(BasicRoutingService SessionRouter.cpp)
asterisk_scf_component_add_file(BasicRoutingService SessionRouter.h)
@@ -20,13 +29,68 @@ asterisk_scf_component_add_file(BasicRoutingService LuaScriptProcessor.cpp)
asterisk_scf_component_add_file(BasicRoutingService LuaScriptProcessor.h)
asterisk_scf_component_add_file(BasicRoutingService RoutingServiceEventPublisher.cpp)
asterisk_scf_component_add_file(BasicRoutingService RoutingServiceEventPublisher.h)
-asterisk_scf_component_add_file(BasicRoutingService WorkQueue.h)
-asterisk_scf_component_add_file(BasicRoutingService SimpleWorkQueue.h)
-asterisk_scf_component_add_file(BasicRoutingService SimpleWorkQueue.cpp)
+asterisk_scf_component_add_file(BasicRoutingService SessionRouterOperation.h)
+asterisk_scf_component_add_file(BasicRoutingService SessionRouterOperation.cpp)
+asterisk_scf_component_add_file(BasicRoutingService RouteSessionOperation.h)
+asterisk_scf_component_add_file(BasicRoutingService RouteSessionOperation.cpp)
+asterisk_scf_component_add_file(BasicRoutingService ConnectBridgedSessionsWithDestinationOperation.h)
+asterisk_scf_component_add_file(BasicRoutingService ConnectBridgedSessionsWithDestinationOperation.cpp)
+asterisk_scf_component_add_file(BasicRoutingService ConnectBridgedSessionsOperation.h)
+asterisk_scf_component_add_file(BasicRoutingService ConnectBridgedSessionsOperation.cpp)
+asterisk_scf_component_add_file(BasicRoutingService SessionListener.h)
+asterisk_scf_component_add_file(BasicRoutingService SessionListener.cpp)
+asterisk_scf_component_add_file(BasicRoutingService OperationReplicaCache.h)
+asterisk_scf_component_add_file(BasicRoutingService OperationReplicaCache.cpp)
+asterisk_scf_component_add_file(BasicRoutingService ReplicationContext.h)
+asterisk_scf_component_add_file(BasicRoutingService ReplicationContext.cpp)
+asterisk_scf_component_add_file(BasicRoutingService TestContext.h)
+asterisk_scf_component_add_file(BasicRoutingService TestContext.cpp)
+asterisk_scf_component_add_file(BasicRoutingService RoutingStateReplicatorListener.h)
+asterisk_scf_component_add_file(BasicRoutingService RoutingStateReplicatorListener.cpp)
+
asterisk_scf_component_add_ice_libraries(BasicRoutingService IceStorm)
asterisk_scf_component_add_boost_libraries(BasicRoutingService thread date_time core regex)
+
asterisk_scf_component_build_icebox(BasicRoutingService)
+
target_link_libraries(BasicRoutingService ${LUA_LIBRARIES})
target_link_libraries(BasicRoutingService logging-client)
+
+target_link_libraries(BasicRoutingService Threading)
target_link_libraries(BasicRoutingService asterisk-scf-api)
-asterisk_scf_component_install(BasicRoutingService)
+
+#asterisk_scf_component_install(BasicRoutingService RUNTIME bin "Basic Routing Service" Core)
+
+########################################
+# Basic Routing State replicator
+
+asterisk_scf_component_init(BasicRoutingStateReplicator CXX)
+
+asterisk_scf_slice_include_directories(${API_SLICE_DIR})
+
+asterisk_scf_component_add_slice(BasicRoutingStateReplicator ServiceLocatorIf)
+asterisk_scf_component_add_slice(BasicRoutingStateReplicator ComponentServiceIf)
+asterisk_scf_component_add_slice(BasicRoutingStateReplicator BasicRoutingStateReplicationIf)
+asterisk_scf_component_add_slice(BasicRoutingStateReplicator RoutingIf)
+
+# This line allows us to use the templated state replicator
+# code. This statement is not the most
+# permanent of changes and assumes the directories are
+# structured in the way that gitall structures them.
+
+include_directories(${utils_dir}/StateReplicator/include)
+include_directories(${utils_dir}/SmartProxy/include)
+include_directories(${API_INCLUDE_DIR})
+
+asterisk_scf_component_add_file(BasicRoutingStateReplicator BasicRoutingStateReplicatorApp.cpp)
+asterisk_scf_component_add_slice(BasicRoutingStateReplicator ../local-slice/BasicRoutingStateReplicationIf.ice)
+
+asterisk_scf_component_add_ice_libraries(BasicRoutingStateReplicator IceStorm)
+
+asterisk_scf_component_build_icebox(BasicRoutingStateReplicator)
+
+target_link_libraries(BasicRoutingStateReplicator logging-client)
+target_link_libraries(BasicRoutingStateReplicator asterisk-scf-api)
+
+#asterisk_scf_component_install(BasicRoutingStateReplicator RUNTIME bin "Basic Routing Service State Replicator" BasicRoutingStateReplicator)
+
diff --git a/src/ConnectBridgedSessionsOperation.cpp b/src/ConnectBridgedSessionsOperation.cpp
new file mode 100644
index 0000000..55770a6
--- /dev/null
+++ b/src/ConnectBridgedSessionsOperation.cpp
@@ -0,0 +1,183 @@
+/*
+ * 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.
+ */
+#include <boost/shared_ptr.hpp>
+#include <boost/bind.hpp>
+
+#include <AsteriskSCF/logger.h>
+
+#include "ConnectBridgedSessionsOperation.h"
+
+using namespace AsteriskSCF;
+using namespace AsteriskSCF::Core::Routing::V1;
+using namespace AsteriskSCF::SessionCommunications::V1;
+using namespace AsteriskSCF::System::Logging;
+
+namespace
+{
+Logger lg = getLoggerFactory().getLogger("AsteriskSCF.BasicRoutingService");
+}
+
+namespace AsteriskSCF
+{
+namespace BasicRoutingService
+{
+
+/**
+ * This class represents an operation to replace one session in a Bridge with sessions
+ * from another bridge. No routing is actually performed. This operation exists here for consistency.
+ * This is a specialization of SessionRouterOperation<T> that handles the
+ * connectBridgedSessions() operation. The template parameter T is the type of
+ * the connectBridgedSessions() AMD callback handler to allow this object to send results to
+ * the initiator of this operation.
+ *
+ * This object is an instance of WorkQueue::Work so that it can enqueued to a worker thread.
+ */
+
+ConnectBridgedSessionsOperation::ConnectBridgedSessionsOperation
+ (const AMD_SessionRouter_connectBridgedSessionsPtr& cb,
+ const std::string& operationId,
+ const ::AsteriskSCF::SessionCommunications::V1::SessionPrx& sessionToReplace,
+ const ::AsteriskSCF::SessionCommunications::V1::SessionPrx& bridgedSession,
+ const ::Ice::Current& current,
+ const SessionContextPtr& context,
+ OperationsManager* const listener)
+ : SessionRouterOperation<AMD_SessionRouter_connectBridgedSessionsPtr,
+ ConnectBridgedSessionsOp::OperationState>
+ (cb,
+ context,
+ current,
+ listener,
+ ConnectBridgedSessionsOp::STATE_CONNECT,
+ operationId),
+ mSessionToReplace(sessionToReplace),
+ mBridgedSession(bridgedSession)
+{
+ mStateMachine.addState(ConnectBridgedSessionsOp::STATE_CONNECT,
+ boost::bind(&ConnectBridgedSessionsOperation::connectBridgedSessionsState, this));
+}
+
+/**
+ * Factory method for the operation.
+ */
+ConnectBridgedSessionsOperationPtr ConnectBridgedSessionsOperation::create
+ (const AMD_SessionRouter_connectBridgedSessionsPtr& cb,
+ const std::string& operationId,
+ const ::AsteriskSCF::SessionCommunications::V1::SessionPrx& sessionToReplace,
+ const ::AsteriskSCF::SessionCommunications::V1::SessionPrx& bridgedSession,
+ const ::Ice::Current& current,
+ const SessionContextPtr& context,
+ OperationsManager* const listener)
+{
+
+ ConnectBridgedSessionsOperationPtr ptr(new ConnectBridgedSessionsOperation(cb,
+ operationId,
+ sessionToReplace,
+ bridgedSession,
+ current,
+ context,
+ listener));
+
+ return ptr;
+}
+
+ConnectBridgedSessionsOperation::~ConnectBridgedSessionsOperation()
+{
+ lg(Debug) << "ConnectBridgedSessionsOperation() being destroyed." ;
+}
+
+/**
+ * This is a state handler for one of this operation's state machine.
+ * Replace one session in a Bridge with sessions from another bridge.
+ */
+void ConnectBridgedSessionsOperation::connectBridgedSessionsState()
+{
+ lg(Debug) << "connectBridgedSessions() entered... ";
+
+ // Get the bridge being merged into.
+ BridgePrx mergeBridge;
+ try
+ {
+ mergeBridge = getBridge(mSessionToReplace);
+ }
+ catch (const Ice::Exception &e)
+ {
+ finishWithException(e);
+ return;
+ }
+
+ SessionSeq preserveSessions;
+ try
+ {
+ preserveSessions = getSessionsInBridge(mergeBridge, mSessionToReplace);
+ }
+ catch (const Ice::Exception &e)
+ {
+ finishWithException(e);
+ return;
+ }
+
+ // Create a listener for the sessions not being replaced to handle early termination.
+ lg(Debug) << "connectBridgedSessions(): Adding listener to " << preserveSessions.size() << " session(s)." ;
+ mListenerManager = SessionListenerManagerPtr(
+ new SessionListenerManager(mSessionContext->adapter, preserveSessions));
+
+ // Get the bridge for the sessions being moved.
+ BridgePrx oldBridge;
+ try
+ {
+ oldBridge = getBridge(mBridgedSession);
+ }
+ catch (const Ice::Exception &e)
+ {
+ finishWithException(e);
+ return;
+ }
+
+ SessionSeq migratingSessions = removeSessionsFromBridge(oldBridge, mBridgedSession);
+
+ // Check for early termination by the source.
+ if (mListenerManager->getListener()->isTerminated())
+ {
+ lg(Notice) << BOOST_CURRENT_FUNCTION <<
+ ": Source ended session before transfer in connectBridgedSessions(). " ;
+ finishWithException(SourceTerminatedPreBridgingException(preserveSessions[0]->getEndpoint()->getId()));
+ return;
+ }
+
+ // We're through listening, and we will probably interfere with the Bridge's functionality if
+ // we keep listening.
+ lg(Debug) << "connectBridgedSessions(): Removing listener. " ;
+ mListenerManager->getListener()->unregister();
+
+ // Now replace the sessions.
+ try
+ {
+ lg(Debug) << BOOST_CURRENT_FUNCTION << ": Asking bridge to replace sessions." ;
+ mergeBridge->replaceSession(mSessionToReplace, migratingSessions);
+ }
+ catch(const Ice::Exception& e)
+ {
+ lg(Error) << BOOST_CURRENT_FUNCTION << ": Unable to replace session for bridge in connectBridgedSessions(). " ;
+ finishWithException(e);
+ return;
+ }
+
+ // This operation is complete. Send AMD responses.
+ finishAndSendResult();
+}
+
+} // end BasicRoutingService
+} // end AsteriskSCF
diff --git a/src/ConnectBridgedSessionsOperation.h b/src/ConnectBridgedSessionsOperation.h
new file mode 100644
index 0000000..f8304dc
--- /dev/null
+++ b/src/ConnectBridgedSessionsOperation.h
@@ -0,0 +1,97 @@
+/*
+ * 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/function.hpp>
+
+#include <AsteriskSCF/Core/Routing/RoutingIf.h>
+#include <AsteriskSCF/SessionCommunications/SessionCommunicationsIf.h>
+#include "AsteriskSCF/StateMachine/SimpleStateMachine.h"
+
+#include "SessionRouterOperation.h"
+
+namespace AsteriskSCF
+{
+namespace BasicRoutingService
+{
+// This namespace exists solely to avoid enumerators from different types
+// colliding.
+namespace ConnectBridgedSessionsOp
+{
+ /**
+ * This enum defines the states for the operation class below.
+ * It would have been defined within the class except that it is needed as a template
+ * parameter to the base class. (C++0x enum class will alleviate such nonsense.)
+ */
+ enum OperationState
+ {
+ STATE_CONNECT
+ };
+}
+
+class ConnectBridgedSessionsOperation;
+typedef boost::shared_ptr<ConnectBridgedSessionsOperation> ConnectBridgedSessionsOperationPtr;
+
+/**
+ * This class represents an operation to replace one session in a Bridge with sessions
+ * from another bridge. No routing is actually performed. This operation exists here for consistency.
+ * This is a specialization of SessionRouterOperation<T> that handles the
+ * connectBridgedSessions() operation. The template parameter T is the type of
+ * the connectBridgedSessions() AMD callback handler to allow this object to send results to
+ * the initiator of this operation.
+ *
+ * This object is an instance of WorkQueue::Work so that it can enqueued to a worker thread.
+ */
+class ConnectBridgedSessionsOperation : public SessionRouterOperation<AsteriskSCF::SessionCommunications::V1::AMD_SessionRouter_connectBridgedSessionsPtr,
+ ConnectBridgedSessionsOp::OperationState>
+{
+public:
+ static ConnectBridgedSessionsOperationPtr create(
+ const AsteriskSCF::SessionCommunications::V1::AMD_SessionRouter_connectBridgedSessionsPtr& cb,
+ const std::string& operationId,
+ const AsteriskSCF::SessionCommunications::V1::SessionPrx& sessionToReplace,
+ const AsteriskSCF::SessionCommunications::V1::SessionPrx& bridgedSession,
+ const Ice::Current& current,
+ const SessionContextPtr& context,
+ OperationsManager* const listener);
+
+ virtual ~ConnectBridgedSessionsOperation();
+
+protected:
+ ConnectBridgedSessionsOperation(const AsteriskSCF::SessionCommunications::V1::AMD_SessionRouter_connectBridgedSessionsPtr& cb,
+ const std::string& operationId,
+ const AsteriskSCF::SessionCommunications::V1::SessionPrx& sessionToReplace,
+ const AsteriskSCF::SessionCommunications::V1::SessionPrx& bridgedSession,
+ const Ice::Current& current,
+ const SessionContextPtr& context,
+ OperationsManager* const listener);
+
+private:
+ /**
+ * This is a state handler for one of this operation's states.
+ * Replace one session in a Bridge with sessions from another bridge.
+ */
+ void connectBridgedSessionsState();
+
+private:
+ // Operation input params.
+ AsteriskSCF::SessionCommunications::V1::SessionPrx mSessionToReplace;
+ AsteriskSCF::SessionCommunications::V1::SessionPrx mBridgedSession;
+
+}; // class ConnectBridgedSessionsOperation
+
+} // end BasicRoutingService
+} // end AsteriskSCF
diff --git a/src/ConnectBridgedSessionsWithDestinationOperation.cpp b/src/ConnectBridgedSessionsWithDestinationOperation.cpp
new file mode 100644
index 0000000..e7ca9d1
--- /dev/null
+++ b/src/ConnectBridgedSessionsWithDestinationOperation.cpp
@@ -0,0 +1,523 @@
+/*
+ * 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;
... 7191 lines suppressed ...
--
asterisk-scf/release/routing.git
More information about the asterisk-scf-commits
mailing list