[asterisk-scf-commits] asterisk-scf/integration/sip.git branch "session-decorator" updated.
Commits to the Asterisk SCF project code repositories
asterisk-scf-commits at lists.digium.com
Fri Jul 15 13:52:30 CDT 2011
branch "session-decorator" has been updated
via e19711fb128b2538bd793dd6993fa1a89a99909e (commit)
via 4f79a16c7cb94d0b6aa5ed13c3bfdc81fc764641 (commit)
via 6577d75e59c8b2f4241103f48fd114da433c8597 (commit)
via e528d7be81f34ac1789b74ad7415de949a1c19ed (commit)
via b61e242edd548a82eb5a492c7ffb16835746ceaa (commit)
via 32c65526dd56d0bc7e03148dfd407bfa40423fff (commit)
via ef373b282be4a218a92e9981fc0c3163c181603b (commit)
via 0a84ba53e4041c3d0afcab7caf954529b60b6f63 (commit)
via 768256f0d324d42905eb9f3832c054193cd750b2 (commit)
via 162d8a50bf6df93a6bc24d06e9042b5026c088ee (commit)
via 41e944174785ff3370b4e165d491e8ff92cc0827 (commit)
via f860396ac01e1581dfe5f56ca30434c8b01b4270 (commit)
via be2641c8edb8a41556c4a0d9c2d86fb023917792 (commit)
via 1ea02fa8ec59f386d62f04a0ef3c1adb77fd5319 (commit)
via 3fa68dd21a3f9a62ca80400e48832d0e6225a2ea (commit)
via fc2f64b78c4c57c4c3b8fea3233a9fd3f52147f6 (commit)
via 575abed0a58ba2e9f321b8ccc121a32356c7288d (commit)
via b0d2423c8b01c4432c7289f9648213e16406174a (commit)
via 1dda9776c5a1f55037ac61f5290593e25a5677cb (commit)
via 34797f06098f619838f094d9af7ec1a8c3ad4724 (commit)
via cfa23a3f19760c5266a9f4be0ad9302c977bfe8a (commit)
via a6bfcd47c9f662ac6dd4a2318d96579737a12837 (commit)
via 57993cd407779a00004dc426cb282f06bccb5bd5 (commit)
via 413750665895171f8eae4c8d86ff672b04a8f7cd (commit)
via b874caeeddf70e7b3911a4312494c8ae6de1ffd4 (commit)
via 89e5e5522e81f9f9be4351eb6aab5f346a35586d (commit)
via 448088a4f28ccdf90471cb60642d28bd4fdfd589 (commit)
via abbaa0220ce9faf1e97fd4845ee7f78076436193 (commit)
via 0ffdaa56376184435e2ebb6b59dac42fbde51ab9 (commit)
via 47ce4d4e87b887297289ff6511bde2feb807a476 (commit)
via e2736843dca94388a5ad562b7096a53451e2d037 (commit)
via 9394b46cf8c83c0cfa83e73a34886aeb9f63913b (commit)
via 66b54622b2edd8c6f41fb76950e81121813804e2 (commit)
via 66c548481a50dbbaaa4f886d41af64cc39a5bfa3 (commit)
via b694e978ae5146b198914f8e114fff6333a75662 (commit)
via 1f56ea5e07117170b407e17cc298ea9a79b60860 (commit)
via 706a6986e046909d0035038412d759ba1e6c199c (commit)
via 1624b232fe8cd35ab4a88ae0aa7d74498d98e3c9 (commit)
via 591d2f8183dbed4dce5734ae09e5c2d2cb2de8c3 (commit)
via ae5e868cf90a06e489748299eca32564f248d52b (commit)
via 1a6d54ba8e42b022e2dea6c696ab26ebac0345a2 (commit)
via c5ed036baf4b2eea39a2496e8c8267be96766dc2 (commit)
via 433163526c9c788f53fb52ae4b389e3c77f7dc31 (commit)
via fa87159c9237db447843c6523f1daa7da2a67607 (commit)
via 9ab8adb83f3038c03b8c1619bebcabbffc4e352e (commit)
via cb2abe96f218864eb0088c343b3a91ffa6e578e8 (commit)
via 480dca6e1d5a251dd2ab680f776b46f37a052660 (commit)
via 871d682ad8f89f934e68326fa6b3d861479321da (commit)
via 2e6f61b3697981bb4a1af0a4f2738ff786668269 (commit)
via f8ba9bd5baffb67df87d1726c8635bb7d5d63ff2 (commit)
via 49064db0f2c505427f826c7fc9608e39d3a2ffe2 (commit)
via 72fce78916b051f6d76a4e61de9fb5d26eec10de (commit)
via 51d7a0addacb9eae0fc88bea5b010a0559b4b887 (commit)
via df0ebe1c9b2b5eb5b99f1a5e4140dd13b3a14622 (commit)
via c5eff00eb9e4e64bac262fbc8f0ff7631c9b5cf8 (commit)
via 5aa5e56f84e645b9c5238215ef4fdeb0eba28f20 (commit)
via dd490ab030f402537f7edac3d1fcd443bc602e68 (commit)
from 26ec91cf1ad7ae654b38a57e355d85cd7b2a6495 (commit)
Summary of changes:
.../SipSessionManager/SipStateReplicationIf.ice | 17 +
src/CMakeLists.txt | 5 +
src/PJSipManager.cpp | 25 +
src/PJSipManager.h | 18 +
src/PJSipRegistrarModule.cpp | 831 ++++++++++++++++++++
src/PJSipRegistrarModule.h | 203 +++++
src/PJSipRegistrarModuleConstruction.cpp | 104 +++
src/PJSipSessionModule.cpp | 12 +-
src/PJSipSessionModule.h | 20 +-
src/PJSipSessionModuleConstruction.cpp | 18 +-
src/SipEndpoint.cpp | 9 +
src/SipEndpoint.h | 2 +
src/SipEndpointFactory.cpp | 13 +-
src/SipEndpointFactory.h | 2 +
src/SipRegistrarListener.cpp | 113 +++
src/SipRegistrarListener.h | 46 ++
src/SipSession.cpp | 38 +-
src/SipSession.h | 2 +-
src/SipSessionManagerApp.cpp | 32 +-
src/SipSessionManagerEndpointLocator.cpp | 14 +-
src/SipStateReplicator.h | 1 +
src/SipStateReplicatorListener.cpp | 172 ++++
22 files changed, 1641 insertions(+), 56 deletions(-)
create mode 100644 src/PJSipRegistrarModule.cpp
create mode 100644 src/PJSipRegistrarModule.h
create mode 100644 src/PJSipRegistrarModuleConstruction.cpp
create mode 100644 src/SipRegistrarListener.cpp
create mode 100644 src/SipRegistrarListener.h
- Log -----------------------------------------------------------------
commit e19711fb128b2538bd793dd6993fa1a89a99909e
Author: Mark Michelson <mmichelson at digium.com>
Date: Fri Jul 15 13:43:39 2011 -0500
Remove remaining references to the word "decorator"
diff --git a/src/PJSipSessionModule.cpp b/src/PJSipSessionModule.cpp
index 294a8f6..f282cf0 100644
--- a/src/PJSipSessionModule.cpp
+++ b/src/PJSipSessionModule.cpp
@@ -173,23 +173,23 @@ SessionWorkPtr PJSipSessionModInfo::getSessionWork()
return mSessionWork;
}
-void SipSessionDecorator::addDecorator(const SessionDecoratorHookPrx& hook, const Ice::Current&)
+void SipSessionCreationExtensionPoint::addSessionCreationHook(const SessionCreationHookPrx& hook, const Ice::Current&)
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
mHooks.push_back(hook);
}
-void SipSessionDecorator::removeDecorator(const SessionDecoratorHookPrx& hook, const Ice::Current&)
+void SipSessionCreationExtensionPoint::removeSessionCreationHook(const SessionCreationHookPrx& hook, const Ice::Current&)
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
mHooks.erase(std::find(mHooks.begin(), mHooks.end(), hook));
}
-void SipSessionDecorator::clearDecorators(const Ice::Current&)
+void SipSessionCreationExtensionPoint::clearSessionCreationHooks(const Ice::Current&)
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
mHooks.clear();
}
-SessionDecoratorHookSeq SipSessionDecorator::getHooks()
+SessionCreationHookSeq SipSessionCreationExtensionPoint::getHooks()
{
boost::shared_lock<boost::shared_mutex> lock(mLock);
return mHooks;
@@ -1542,9 +1542,9 @@ QueuePtr PJSipSessionModule::getThreadPoolQueue()
return mPoolQueue;
}
-SessionDecoratorHookSeq PJSipSessionModule::getSessionDecoratorHooks()
+SessionCreationHookSeq PJSipSessionModule::getSessionCreationHooks()
{
- return mSessionDecorator->getHooks();
+ return mSessionCreationExtensionPoint->getHooks();
}
void PJSipSessionModule::enqueueSessionWork(const SuspendableWorkPtr& work, pjsip_inv_session *inv)
diff --git a/src/PJSipSessionModule.h b/src/PJSipSessionModule.h
index e3a72b0..7ea7e5d 100644
--- a/src/PJSipSessionModule.h
+++ b/src/PJSipSessionModule.h
@@ -82,20 +82,20 @@ private:
typedef IceUtil::Handle<PJSipSessionModuleThreadPoolListener> PJSipSessionModuleThreadPoolListenerPtr;
-class SipSessionDecorator : public AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorExtensionPoint
+class SipSessionCreationExtensionPoint : public AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationExtensionPoint
{
public:
- void addDecorator(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookPrx& hook, const Ice::Current&);
- void removeDecorator(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookPrx& hook, const Ice::Current&);
- void clearDecorators(const Ice::Current&);
- AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookSeq getHooks();
+ void addSessionCreationHook(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookPrx& hook, const Ice::Current&);
+ void removeSessionCreationHook(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookPrx& hook, const Ice::Current&);
+ void clearSessionCreationHooks(const Ice::Current&);
+ AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookSeq getHooks();
private:
- AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookSeq mHooks;
+ AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookSeq mHooks;
boost::shared_mutex mLock;
};
-typedef IceUtil::Handle<SipSessionDecorator> SipSessionDecoratorPtr;
+typedef IceUtil::Handle<SipSessionCreationExtensionPoint> SipSessionCreationExtensionPointPtr;
class PJSipSessionModule : public PJSipModule
{
@@ -135,7 +135,7 @@ public:
AsteriskSCF::System::WorkQueue::V1::QueuePtr getThreadPoolQueue();
- AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookSeq getSessionDecoratorHooks();
+ AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookSeq getSessionCreationHooks();
private:
void handleNewInvite(pjsip_rx_data *rdata);
@@ -153,9 +153,9 @@ private:
AsteriskSCF::System::ThreadPool::V1::PoolPtr mPool;
AsteriskSCF::System::WorkQueue::V1::QueuePtr mPoolQueue;
AsteriskSCF::System::ThreadPool::V1::PoolListenerPtr mPoolListener;
- SipSessionDecoratorPtr mSessionDecorator;
+ SipSessionCreationExtensionPointPtr mSessionCreationExtensionPoint;
Ice::ObjectAdapterPtr mAdapter;
- AsteriskSCF::Core::Discovery::V1::ServiceManagementPrx mDecoratorService;
+ AsteriskSCF::Core::Discovery::V1::ServiceManagementPrx mSessionCreationExtensionPointService;
};
typedef IceUtil::Handle<PJSipSessionModule> PJSipSessionModulePtr;
diff --git a/src/PJSipSessionModuleConstruction.cpp b/src/PJSipSessionModuleConstruction.cpp
index 459cd9b..8bf443b 100644
--- a/src/PJSipSessionModuleConstruction.cpp
+++ b/src/PJSipSessionModuleConstruction.cpp
@@ -25,7 +25,7 @@ using namespace AsteriskSCF::System::Logging;
namespace
{
Logger lg = getLoggerFactory().getLogger("AsteriskSCF.SipSessionManager");
-const std::string DecoratorId("SipSessionDecorator");
+const std::string SessionCreationExtensionPointId("SipSessionCreationExtensionPoint");
}
namespace AsteriskSCF
@@ -118,7 +118,7 @@ PJSipSessionModule::PJSipSessionModule(pjsip_endpoint *endpt,
: mName(moduleName), mEndpointFactory(endpointFactoryPtr),
mSessionRouter(sessionRouter), mServiceLocator(serviceLocator),
mStateReplicator(stateReplicator), mReplica(replica), mEndpoint(endpt),
- mSessionDecorator(new SipSessionDecorator()), mAdapter(adapter)
+ mSessionCreationExtensionPoint(new SipSessionCreationExtensionPoint()), mAdapter(adapter)
{
sessionModule = this;
mModule.name = pj_str(moduleName);
@@ -133,12 +133,12 @@ PJSipSessionModule::PJSipSessionModule(pjsip_endpoint *endpt,
mModule.on_tx_response = NULL;
mModule.on_tsx_state = NULL;
- Ice::ObjectPrx decoratorObjPrx = mAdapter->add(mSessionDecorator, mAdapter->getCommunicator()->stringToIdentity(DecoratorId));
- SessionDecoratorExtensionPointPrx decoratorPrx =
- SessionDecoratorExtensionPointPrx::uncheckedCast(decoratorObjPrx);
+ Ice::ObjectPrx extensionPointObjPrx = mAdapter->add(mSessionCreationExtensionPoint, mAdapter->getCommunicator()->stringToIdentity(SessionCreationExtensionPointId));
+ SessionCreationExtensionPointPrx extensionPointPrx =
+ SessionCreationExtensionPointPrx::uncheckedCast(extensionPointObjPrx);
- mDecoratorService = serviceLocatorManagement->addService(decoratorPrx, DecoratorId);
- mDecoratorService->addLocatorParams(new AsteriskSCF::Core::Discovery::V1::ServiceLocatorParams(SessionDecoratorLocatorCategory), "");
+ mSessionCreationExtensionPointService = serviceLocatorManagement->addService(extensionPointPrx, SessionCreationExtensionPointId);
+ mSessionCreationExtensionPointService->addLocatorParams(new AsteriskSCF::Core::Discovery::V1::ServiceLocatorParams(SessionCreationHookLocatorCategory), "");
mPoolQueue = new AsteriskSCF::WorkQueue::WorkQueue();
mPoolListener = new PJSipSessionModuleThreadPoolListener();
@@ -181,8 +181,8 @@ PJSipSessionModule::~PJSipSessionModule()
try
{
mPoolQueue->shutdown();
- mDecoratorService->unregister();
- mAdapter->remove(mAdapter->getCommunicator()->stringToIdentity(DecoratorId));
+ mSessionCreationExtensionPointService->unregister();
+ mAdapter->remove(mAdapter->getCommunicator()->stringToIdentity(SessionCreationExtensionPointId));
}
catch (const AsteriskSCF::System::WorkQueue::V1::ShuttingDown&)
{
diff --git a/src/SipSession.cpp b/src/SipSession.cpp
index 0b54281..f733e56 100644
--- a/src/SipSession.cpp
+++ b/src/SipSession.cpp
@@ -196,9 +196,11 @@ public:
AsteriskSCF::SessionCommunications::V1::SessionPrx mSessionProxy;
/**
- * A proxy to the decorator for this session.
+ * When someone asks for our session proxy, we give them this. This may
+ * be the same as mSessionProxy, or it may be a different one set by
+ * a session creation hook.
*/
- AsteriskSCF::SessionCommunications::V1::SessionPrx mSessionDecoratorProxy;
+ AsteriskSCF::SessionCommunications::V1::SessionPrx mPublicSessionProxy;
/**
* The Ice object adapter that our proxies were added to.
@@ -348,22 +350,23 @@ void SipSession::initializePJSIPStructs()
setInviteSession(inviteSession);
}
-void SipSession::activateIceObjects(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookSeq& hooks)
+void SipSession::activateIceObjects(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookSeq& hooks)
{
mImplPriv->mSessionProxy =
AsteriskSCF::SessionCommunications::V1::SessionPrx::uncheckedCast(mImplPriv->mAdapter->addWithUUID(this));
- bool decoratorSet = false;
- for (AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookSeq::const_iterator iter = hooks.begin();
+ AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookDataPtr initial(new AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookData(mImplPriv->mSessionProxy, mImplPriv->mListeners));
+
+ AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookDataPtr in = initial;
+
+ for (AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookSeq::const_iterator iter = hooks.begin();
iter != hooks.end(); ++iter)
{
- AsteriskSCF::SessionCommunications::V1::SessionPrx decorator;
- AsteriskSCF::SessionCommunications::V1::SessionListenerPrx decoratorListener;
AsteriskSCF::System::Hook::V1::HookResult result;
-
+ AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookDataPtr out;
try
{
- result = (*iter)->decorateSession(mImplPriv->mSessionProxy, mImplPriv->mListeners, decorator, decoratorListener);
+ result = (*iter)->decorateSession(in, out);
}
catch (const Ice::Exception& ex)
{
@@ -374,12 +377,11 @@ void SipSession::activateIceObjects(const AsteriskSCF::SessionCommunications::Ex
if (result.status == AsteriskSCF::System::Hook::V1::Succeeded)
{
- lg(Debug) << "Setting our session proxy to be the decorator " << decorator->ice_getIdentity().name;
- mImplPriv->mSessionDecoratorProxy = decorator;
+ lg(Debug) << "Setting our session proxy to be " << out->session->ice_getIdentity().name;
+ mImplPriv->mPublicSessionProxy = out->session;
mImplPriv->mListeners.clear();
- mImplPriv->mListeners.push_back(decoratorListener);
- decoratorSet = true;
- break;
+ mImplPriv->mListeners = out->listeners;
+ in = out;
}
}
}
@@ -398,7 +400,7 @@ SipSession::SipSession(const Ice::ObjectAdapterPtr& adapter, const SipEndpointPt
mImplPriv->mListeners.push_back(listener);
}
- activateIceObjects(mImplPriv->mManager->getSessionModule()->getSessionDecoratorHooks());
+ activateIceObjects(mImplPriv->mManager->getSessionModule()->getSessionCreationHooks());
mImplPriv->mMediaSession = new SipMediaSession(this);
mImplPriv->mMediaSessionProxy =
@@ -422,7 +424,7 @@ SipSession::SipSession(const Ice::ObjectAdapterPtr& adapter, const SipEndpointPt
const AsteriskSCF::System::Component::V1::ReplicaPtr& replica, bool isUAC)
: mImplPriv(new SipSessionPriv(adapter, endpoint, destination, manager, serviceLocator, replica))
{
- activateIceObjects(mImplPriv->mManager->getSessionModule()->getSessionDecoratorHooks());
+ activateIceObjects(mImplPriv->mManager->getSessionModule()->getSessionCreationHooks());
mImplPriv->mMediaSession = new SipMediaSession(this);
mImplPriv->mMediaSessionProxy =
@@ -1619,9 +1621,9 @@ SipEndpointPtr SipSession::getEndpoint()
*/
AsteriskSCF::SessionCommunications::V1::SessionPrx& SipSession::getSessionProxy()
{
- if (mImplPriv->mSessionDecoratorProxy)
+ if (mImplPriv->mPublicSessionProxy)
{
- return mImplPriv->mSessionDecoratorProxy;
+ return mImplPriv->mPublicSessionProxy;
}
return mImplPriv->mSessionProxy;
}
diff --git a/src/SipSession.h b/src/SipSession.h
index 99fadd7..91b4f6c 100644
--- a/src/SipSession.h
+++ b/src/SipSession.h
@@ -156,7 +156,7 @@ public:
const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx&,
const Ice::Current&);
- void activateIceObjects(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionDecoratorHookSeq& hooks);
+ void activateIceObjects(const AsteriskSCF::SessionCommunications::ExtensionPoints::V1::SessionCreationHookSeq& hooks);
void hold(const Ice::Current&);
void progress(const AsteriskSCF::SessionCommunications::V1::ResponseCodePtr&, const Ice::Current&);
commit 4f79a16c7cb94d0b6aa5ed13c3bfdc81fc764641
Merge: 26ec91c 6577d75
Author: Mark Michelson <mmichelson at digium.com>
Date: Fri Jul 15 10:40:51 2011 -0500
Merge branch 'master' into session-decorator
Conflicts:
src/SipEndpointFactory.h
diff --cc src/SipEndpointFactory.h
index 6dca59f,ce8c59b..b6ccbd4
--- a/src/SipEndpointFactory.h
+++ b/src/SipEndpointFactory.h
@@@ -21,7 -21,7 +21,8 @@@
#include <boost/enable_shared_from_this.hpp>
#include <AsteriskSCF/Core/Endpoint/EndpointIf.h>
+#include <AsteriskSCF/SessionCommunications/SessionCommunicationsExtensionPointsIf.h>
+ #include <AsteriskSCF/SIP/SIPRegistrarIf.h>
namespace AsteriskSCF
{
commit 6577d75e59c8b2f4241103f48fd114da433c8597
Merge: e528d7b fc8f711
Author: Mark Michelson <mmichelson at digium.com>
Date: Thu Jul 14 17:07:46 2011 -0500
Merge branch 'master' into registrar
commit e528d7be81f34ac1789b74ad7415de949a1c19ed
Author: Mark Michelson <mmichelson at digium.com>
Date: Tue Jul 5 13:34:47 2011 -0500
s/updateBindings/addAndRemoveBindings/
The name change helps to differentiate it from updateBinding() and
it also makes it more clear what is actually being done.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index 67697b7..b01832a 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -471,10 +471,10 @@ void PJSipRegistrarModule::replicateState(
//XXX This and the method that queues this work should
//have different names. It's easy to get confused with
//UpdateBinding, and this could stand to be more descriptive
-class UpdateBindings : public Work
+class AddAndRemoveBindings : public Work
{
public:
- UpdateBindings(
+ AddAndRemoveBindings(
const RegistrarIPtr& registrar,
const std::string& aor,
const BindingWrapperSeq& newBindings,
@@ -532,9 +532,9 @@ private:
BindingWrapperSeq mRemovedBindings;
};
-void RegistrarI::updateBindings(const std::string &aor, BindingWrapperSeq& newBindings, BindingWrapperSeq& removedBindings, const QueuePtr& queue)
+void RegistrarI::addAndRemoveBindings(const std::string &aor, BindingWrapperSeq& newBindings, BindingWrapperSeq& removedBindings, const QueuePtr& queue)
{
- queue->enqueueWork(new UpdateBindings(this, aor, newBindings, removedBindings));
+ queue->enqueueWork(new AddAndRemoveBindings(this, aor, newBindings, removedBindings));
}
BindingUpdateSeq RegistrarI::createBindingUpdateSeq(const std::string& aor, BindingWrapperSeq& bindings)
@@ -784,7 +784,7 @@ pj_bool_t PJSipRegistrarModule::on_rx_request(pjsip_rx_data *rdata)
}
}
- mRegistrar->updateBindings(aor, newBindings, removedBindings, mRegistrar->getQueue());
+ mRegistrar->addAndRemoveBindings(aor, newBindings, removedBindings, mRegistrar->getQueue());
// We enqueue the SIP response to make sure we send it and replicate state AFTER we have updated
// our internal bindings.
diff --git a/src/PJSipRegistrarModule.h b/src/PJSipRegistrarModule.h
index c29e24b..6105e5d 100644
--- a/src/PJSipRegistrarModule.h
+++ b/src/PJSipRegistrarModule.h
@@ -64,7 +64,7 @@ public:
void removeBinding(BindingWrapperSeq& currentBindings, BindingWrapperPtr wrapper);
void removeAOR(const std::string& aor);
- void updateBindings(
+ void addAndRemoveBindings(
const std::string &aor,
BindingWrapperSeq& newBindings,
BindingWrapperSeq& removedBindings,
diff --git a/src/SipStateReplicatorListener.cpp b/src/SipStateReplicatorListener.cpp
index c25b800..41f1fd9 100644
--- a/src/SipStateReplicatorListener.cpp
+++ b/src/SipStateReplicatorListener.cpp
@@ -126,7 +126,7 @@ public:
//that we've already expired and removed. No biggie. Just
//keep going.
}
- regModule->getRegistrar()->updateBindings(dictIter->first, emptySeq, toRemove, regModule->getRegistrar()->getQueue());
+ regModule->getRegistrar()->addAndRemoveBindings(dictIter->first, emptySeq, toRemove, regModule->getRegistrar()->getQueue());
}
}
@@ -153,7 +153,7 @@ public:
void setNewRegistrationState(const PJSipRegistrarModulePtr& regModule, const SipRegistrarStateItemPtr& regItem)
{
//For new registrations, we need to create a sequence of BindingWrappers and call the registrar's
- //updateBindings() function.
+ //addAndRemoveBindings() function.
for (BindingDict::iterator dictIter = regItem->bindings.begin();
dictIter != regItem->bindings.end(); ++dictIter)
{
@@ -166,7 +166,7 @@ public:
BindingWrapperPtr wrapper = createBindingWrapper(dictIter->first, *seqIter, regModule);
wrapperSeq.push_back(wrapper);
}
- regModule->getRegistrar()->updateBindings(dictIter->first, wrapperSeq, emptySeq, regModule->getRegistrar()->getQueue());
+ regModule->getRegistrar()->addAndRemoveBindings(dictIter->first, wrapperSeq, emptySeq, regModule->getRegistrar()->getQueue());
}
}
@@ -203,7 +203,7 @@ public:
}
if (!newBindings.empty())
{
- regModule->getRegistrar()->updateBindings(dictIter->first, newBindings, emptySeq, regModule->getRegistrar()->getQueue());
+ regModule->getRegistrar()->addAndRemoveBindings(dictIter->first, newBindings, emptySeq, regModule->getRegistrar()->getQueue());
}
}
}
commit b61e242edd548a82eb5a492c7ffb16835746ceaa
Author: Mark Michelson <mmichelson at digium.com>
Date: Tue Jul 5 13:19:51 2011 -0500
Fix the addListener method to actually return content.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index a8e18f7..67697b7 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -177,9 +177,7 @@ public:
lg(Debug) << "Adding Listener " << mListener->ice_getIdentity().name << " to registrar";
listeners.push_back(mListener);
}
- //We should maybe throw an exception if someone tries to insert a duplicate?
- //XXX Need to update this to actually return something.
- mCB->ice_response(ContactDict());
+ mCB->ice_response(mRegistrar->createContactDict(mRegistrar->getBindings()));
}
private:
AMD_Registrar_addListenerPtr mCB;
@@ -337,6 +335,22 @@ void RegistrarI::removeAOR(const std::string& aor)
mBindings.erase(aor);
}
+ContactDict RegistrarI::createContactDict(const BindingWrapperDict& bindings)
+{
+ Ice::StringSeq returnedContacts;
+ ContactDict dict;
+ for (BindingWrapperDict::const_iterator dictIter = bindings.begin();
+ dictIter != bindings.end(); ++dictIter)
+ {
+ for (BindingWrapperSeq::const_iterator seqIter = dictIter->second.begin();
+ seqIter != dictIter->second.end(); ++seqIter)
+ {
+ dict[dictIter->first].push_back((*seqIter)->mBinding->contact);
+ }
+ }
+ return dict;
+}
+
BindingDict RegistrarI::createBindingDict(const std::string& aor, const BindingWrapperSeq& bindings)
{
BindingSeq returnedBindings;
diff --git a/src/PJSipRegistrarModule.h b/src/PJSipRegistrarModule.h
index 76b422a..c29e24b 100644
--- a/src/PJSipRegistrarModule.h
+++ b/src/PJSipRegistrarModule.h
@@ -76,6 +76,7 @@ public:
AsteriskSCF::SIP::Registration::V1::BindingDict createBindingDict(const std::string& aor, const BindingWrapperSeq& bindings);
+ AsteriskSCF::SIP::Registration::V1::ContactDict createContactDict(const BindingWrapperDict& bindings);
std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx>& getListeners();
commit 32c65526dd56d0bc7e03148dfd407bfa40423fff
Author: Mark Michelson <mmichelson at digium.com>
Date: Tue Jul 5 11:09:27 2011 -0500
Get state replication and the SIP response to the REGISTER properly queued.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index 8fd413e..a8e18f7 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -21,6 +21,7 @@
#include <AsteriskSCF/WorkQueue/DefaultQueueListener.h>
#include "PJSipRegistrarModule.h"
+#include "PJSipManager.h"
using namespace AsteriskSCF::SIP::Registration::V1;
using namespace AsteriskSCF::System::Logging;
@@ -82,7 +83,6 @@ BindingWrapper::BindingWrapper(
pj_timer_entry_init(&mEntry, expirationId++, this, registrationExpired);
lg(Debug) << "Scheduling binding " << mBinding->contact << " for expiration in " << expiration << " seconds.";
scheduleRegistrationExpiration(expiration);
- pjsip_endpt_schedule_timer(mEndpoint, &mEntry, &delay);
}
BindingWrapper::~BindingWrapper()
@@ -143,7 +143,7 @@ RegistrarI::RegistrarI(const RegistrarListenerPrx& defaultListener)
{
lg(Debug) << "In RegistrarI constructor, should be adding a listener...";
mQueue = new AsteriskSCF::WorkQueue::WorkQueue();
- AsteriskSCF::WorkQueue::DefaultQueueListenerPtr listener(new AsteriskSCF::WorkQueue::DefaultQueueListener(mQueue));
+ AsteriskSCF::WorkQueue::DefaultQueueListenerPtr listener(new AsteriskSCF::WorkQueue::DefaultQueueListener(mQueue, new pjlibHook()));
mQueue->setListener(listener);
mListeners.push_back(defaultListener);
}
@@ -365,21 +365,80 @@ QueuePtr RegistrarI::getQueue()
return mQueue;
}
-//XXX
-//This function is called both from the work queue and from outside the
-//work queue. Either this needs to be made into its own queueable operation,
-//or it needs locking. Preferably, this will just be another queueable operation
-//though.
+class ReplicateState : public Work
+{
+public:
+ ReplicateState(
+ const AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx& stateReplicator,
+ const std::string &aor,
+ const BindingWrapperSeq& existingBindings,
+ const BindingWrapperSeq& newBindings,
+ const BindingWrapperSeq& removedBindings,
+ const RegistrarIPtr& registrar)
+ : mStateReplicator(stateReplicator), mAOR(aor), mExistingBindings(existingBindings), mNewBindings(newBindings),
+ mRemovedBindings(removedBindings), mRegistrar(registrar) { }
+
+ void execute()
+ {
+ try
+ {
+ if (!mNewBindings.empty())
+ {
+ BindingDict newDict = mRegistrar->createBindingDict(mAOR, mNewBindings);
+ SipRegistrarStateItemPtr newItem(new SipRegistrarStateItem(mAOR, mAOR, newDict, true));
+ SipStateItemSeq items;
+ items.push_back(newItem);
+ mStateReplicator->setState(items);
+ }
+
+ if (!mExistingBindings.empty())
+ {
+ BindingDict existingDict = mRegistrar->createBindingDict(mAOR, mExistingBindings);
+ SipRegistrarStateItemPtr existingItem(new SipRegistrarStateItem(mAOR, mAOR, existingDict, false));
+ SipStateItemSeq items;
+ items.push_back(existingItem);
+ mStateReplicator->setState(items);
+ }
+
+ if (!mRemovedBindings.empty())
+ {
+ BindingDict removedDict = mRegistrar->createBindingDict(mAOR, mRemovedBindings);
+ SipRegistrarStateItemPtr removedItem(new SipRegistrarStateItem(mAOR, mAOR, removedDict, false));
+ SipStateItemSeq items;
+ items.push_back(removedItem);
+ mStateReplicator->removeStateForItems(items);
+ }
+ }
+ catch (const Ice::TwowayOnlyException&)
+ {
+ lg(Error) << "State replication operations are not oneway";
+ }
+ catch (const Ice::Exception& ex)
+ {
+ lg(Error) << "Exception caught attempting to replicate state in registrar\n" << ex.what();
+ }
+
+ }
+private:
+ AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx mStateReplicator;
+ const std::string mAOR;
+ BindingWrapperSeq mExistingBindings;
+ BindingWrapperSeq mNewBindings;
+ BindingWrapperSeq mRemovedBindings;
+ RegistrarIPtr mRegistrar;
+};
+
void PJSipRegistrarModule::replicateState(
const std::string &aor,
const BindingWrapperSeq& existingBindings,
const BindingWrapperSeq& newBindings,
const BindingWrapperSeq& removedBindings)
{
+
+ AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx oneWayStateReplicator;
try
{
- AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx oneWayStateReplicator =
- AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx::uncheckedCast(mStateReplicator->ice_oneway());
+ oneWayStateReplicator = AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx::uncheckedCast(mStateReplicator->ice_oneway());
}
catch (const Ice::NoEndpointException&)
{
@@ -392,43 +451,7 @@ void PJSipRegistrarModule::replicateState(
return;
}
- try
- {
- if (!newBindings.empty())
- {
- BindingDict newDict = mRegistrar->createBindingDict(aor, newBindings);
- SipRegistrarStateItemPtr newItem(new SipRegistrarStateItem(aor, aor, newDict, true));
- SipStateItemSeq items;
- items.push_back(newItem);
- mStateReplicator->setState(items);
- }
-
- if (!existingBindings.empty())
- {
- BindingDict existingDict = mRegistrar->createBindingDict(aor, existingBindings);
- SipRegistrarStateItemPtr existingItem(new SipRegistrarStateItem(aor, aor, existingDict, false));
- SipStateItemSeq items;
- items.push_back(existingItem);
- mStateReplicator->setState(items);
- }
-
- if (!removedBindings.empty())
- {
- BindingDict removedDict = mRegistrar->createBindingDict(aor, removedBindings);
- SipRegistrarStateItemPtr removedItem(new SipRegistrarStateItem(aor, aor, removedDict, false));
- SipStateItemSeq items;
- items.push_back(removedItem);
- mStateReplicator->removeStateForItems(items);
- }
- }
- catch (const Ice::TwowayOnlyException&)
- {
- lg(Error) << "State replication operations are not oneway";
- }
- catch (const Ice::Exception& ex)
- {
- lg(Error) << "Exception caught attempting to replicate state in registrar\n" << ex.what();
- }
+ mRegistrar->getQueue()->enqueueWork(new ReplicateState(oneWayStateReplicator, aor, existingBindings, newBindings, removedBindings, mRegistrar));
}
//XXX This and the method that queues this work should
@@ -495,10 +518,8 @@ private:
BindingWrapperSeq mRemovedBindings;
};
-//XXX Potential queueable operation
void RegistrarI::updateBindings(const std::string &aor, BindingWrapperSeq& newBindings, BindingWrapperSeq& removedBindings, const QueuePtr& queue)
{
-
queue->enqueueWork(new UpdateBindings(this, aor, newBindings, removedBindings));
}
@@ -661,6 +682,20 @@ BindingWrapperPtr PJSipRegistrarModule::createNewBinding(pjsip_contact_hdr *cont
return wrapper;
}
+class Response : public Work
+{
+public:
+ Response(pjsip_transaction *tsx, pjsip_tx_data *tdata)
+ : mTsx(tsx), mTdata(tdata) { }
+ void execute()
+ {
+ pjsip_tsx_send_msg(mTsx, mTdata);
+ }
+private:
+ pjsip_transaction *mTsx;
+ pjsip_tx_data *mTdata;
+};
+
pj_bool_t PJSipRegistrarModule::on_rx_request(pjsip_rx_data *rdata)
{
if (rdata->msg_info.msg->line.req.method.id != PJSIP_REGISTER_METHOD)
@@ -737,11 +772,13 @@ pj_bool_t PJSipRegistrarModule::on_rx_request(pjsip_rx_data *rdata)
mRegistrar->updateBindings(aor, newBindings, removedBindings, mRegistrar->getQueue());
+ // We enqueue the SIP response to make sure we send it and replicate state AFTER we have updated
+ // our internal bindings.
pjsip_tx_data *tdata;
pjsip_endpt_create_response(tsx->endpt, rdata, 200, NULL, &tdata);
- pjsip_tsx_send_msg(tsx, tdata);
+ mRegistrar->getQueue()->enqueueWork(new Response(tsx, tdata));
- replicateState(aor, newBindings, existingBindings, removedBindings);
+ replicateState(aor, existingBindings, newBindings, removedBindings);
return PJ_TRUE;
}
commit ef373b282be4a218a92e9981fc0c3163c181603b
Author: Mark Michelson <mmichelson at digium.com>
Date: Sun Jun 26 23:41:22 2011 -0500
Add XXX comments so I know what to tackle next.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index f6b2fa3..8fd413e 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -66,8 +66,6 @@ private:
static void registrationExpired(pj_timer_heap_t *, struct pj_timer_entry *entry)
{
BindingWrapper *wrapper = static_cast<BindingWrapper *>(entry->user_data);
- //Create a BindingWrapperPtr so that when we remove the wrapper from the
- //registrar, it doesn't get deleted.
wrapper->mRegModule->getRegistrar()->getQueue()->enqueueWork(new RegistrationExpired(wrapper));
}
@@ -82,8 +80,8 @@ BindingWrapper::BindingWrapper(
: mRegModule(regModule), mBinding(binding), mEndpoint(endpt), mAOR(aor)
{
pj_timer_entry_init(&mEntry, expirationId++, this, registrationExpired);
- pj_time_val delay = {expiration, 0};
lg(Debug) << "Scheduling binding " << mBinding->contact << " for expiration in " << expiration << " seconds.";
+ scheduleRegistrationExpiration(expiration);
pjsip_endpt_schedule_timer(mEndpoint, &mEntry, &delay);
}
@@ -367,6 +365,11 @@ QueuePtr RegistrarI::getQueue()
return mQueue;
}
+//XXX
+//This function is called both from the work queue and from outside the
+//work queue. Either this needs to be made into its own queueable operation,
+//or it needs locking. Preferably, this will just be another queueable operation
+//though.
void PJSipRegistrarModule::replicateState(
const std::string &aor,
const BindingWrapperSeq& existingBindings,
@@ -428,6 +431,9 @@ void PJSipRegistrarModule::replicateState(
}
}
+//XXX This and the method that queues this work should
+//have different names. It's easy to get confused with
+//UpdateBinding, and this could stand to be more descriptive
class UpdateBindings : public Work
{
public:
commit 0a84ba53e4041c3d0afcab7caf954529b60b6f63
Author: Mark Michelson <mmichelson at digium.com>
Date: Sun Jun 26 23:21:40 2011 -0500
Remove XXX "Queueable operation comments."
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index 0425179..f6b2fa3 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -189,7 +189,6 @@ private:
RegistrarIPtr mRegistrar;
};
-//XXX Queueable operation
void RegistrarI::addListener_async(
const AMD_Registrar_addListenerPtr& cb,
const RegistrarListenerPrx& listener,
@@ -222,7 +221,6 @@ private:
};
-//XXX Queueable operation
void RegistrarI::removeListener_async(
const AMD_Registrar_removeListenerPtr& cb,
const RegistrarListenerPrx& listener,
@@ -267,7 +265,6 @@ private:
RegistrarIPtr mRegistrar;
};
-//XXX Queueable operation
void RegistrarI::getAllBindings_async(
const AMD_Registrar_getAllBindingsPtr& cb,
const Ice::Current&)
@@ -308,7 +305,6 @@ private:
RegistrarIPtr mRegistrar;
};
-//XXX Queueable Operation
void RegistrarI::getAORBindings_async(
const AMD_Registrar_getAORBindingsPtr& cb,
const std::string& aor,
commit 768256f0d324d42905eb9f3832c054193cd750b2
Author: Mark Michelson <mmichelson at digium.com>
Date: Sun Jun 26 23:14:25 2011 -0500
Use a work queue for registrar methods.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index 5363154..0425179 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -161,6 +161,33 @@ RegistrarI::~RegistrarI()
}
}
+class AddListener : public Work
+{
+public:
+ AddListener(
+ const AMD_Registrar_addListenerPtr& cb,
+ const RegistrarListenerPrx& listener,
+ const RegistrarIPtr& registrar)
+ : mCB(cb), mListener(listener), mRegistrar(registrar) { }
+
+ void execute()
+ {
+ std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx>& listeners(mRegistrar->getListeners());
+
+ if (std::find(listeners.begin(), listeners.end(), mListener) == listeners.end())
+ {
+ lg(Debug) << "Adding Listener " << mListener->ice_getIdentity().name << " to registrar";
+ listeners.push_back(mListener);
+ }
+ //We should maybe throw an exception if someone tries to insert a duplicate?
+ //XXX Need to update this to actually return something.
+ mCB->ice_response(ContactDict());
+ }
+private:
+ AMD_Registrar_addListenerPtr mCB;
+ RegistrarListenerPrx mListener;
+ RegistrarIPtr mRegistrar;
+};
//XXX Queueable operation
void RegistrarI::addListener_async(
@@ -168,16 +195,32 @@ void RegistrarI::addListener_async(
const RegistrarListenerPrx& listener,
const Ice::Current&)
{
- if (std::find(mListeners.begin(), mListeners.end(), listener) == mListeners.end())
+ mQueue->enqueueWork(new AddListener(cb, listener, this));
+}
+
+class RemoveListener : public Work
+{
+public:
+ RemoveListener(
+ const AMD_Registrar_removeListenerPtr& cb,
+ const RegistrarListenerPrx& listener,
+ const RegistrarIPtr& registrar)
+ : mCB(cb), mListener(listener), mRegistrar(registrar) { }
+
+ void execute()
{
- lg(Debug) << "Adding Listener " << listener->ice_getIdentity().name << " to registrar";
- mListeners.push_back(listener);
+ std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx>& listeners(mRegistrar->getListeners());
+
+ listeners.erase(std::remove(listeners.begin(), listeners.end(), mListener));
+ mCB->ice_response();
}
- //We should maybe throw an exception if someone tries to insert a duplicate?
- //XXX Are we ever even populating mContacts??
- cb->ice_response(mContacts);
- return;
-}
+
+private:
+ AMD_Registrar_removeListenerPtr mCB;
+ RegistrarListenerPrx mListener;
+ RegistrarIPtr mRegistrar;
+
+};
//XXX Queueable operation
void RegistrarI::removeListener_async(
@@ -185,54 +228,93 @@ void RegistrarI::removeListener_async(
const RegistrarListenerPrx& listener,
const Ice::Current&)
{
- mListeners.erase(std::remove(mListeners.begin(), mListeners.end(), listener));
- cb->ice_response();
+ mQueue->enqueueWork(new RemoveListener(cb, listener, this));
}
+class GetAllBindings : public Work
+{
+public:
+ GetAllBindings(
+ const AMD_Registrar_getAllBindingsPtr& cb,
+ const RegistrarIPtr& registrar)
+ : mCB(cb), mRegistrar(registrar) { }
+
+ void execute()
+ {
+ BindingDict returnedBindings;
+ //XXX
+ //This is suboptimal. What may work better is to maintain
+ //parallel maps of BindingWrappers and Bindings so that
+ //we can quickly return the appropriate information instead of
+ //having to construct a BindingDict on each call.
+
+ BindingWrapperDict& regBindings = mRegistrar->getBindings();
+ for (BindingWrapperDict::iterator iter = regBindings.begin();
+ iter != regBindings.end(); ++iter)
+ {
+ BindingSeq bindings;
+ for (BindingWrapperSeq::iterator seqIter = iter->second.begin();
+ seqIter != iter->second.end(); ++seqIter)
+ {
+ bindings.push_back((*seqIter)->mBinding);
+ }
+ returnedBindings.insert(make_pair(iter->first, bindings));
+ }
+ mCB->ice_response(returnedBindings);
+ }
+private:
+ AMD_Registrar_getAllBindingsPtr mCB;
+ RegistrarIPtr mRegistrar;
+};
+
//XXX Queueable operation
void RegistrarI::getAllBindings_async(
const AMD_Registrar_getAllBindingsPtr& cb,
const Ice::Current&)
{
- BindingDict returnedBindings;
- //XXX
- //This is suboptimal. What may work better is to maintain
- //parallel maps of BindingWrappers and Bindings so that
- //we can quickly return the appropriate information instead of
- //having to construct a BindingDict on each call.
- for (BindingWrapperDict::iterator iter = mBindings.begin();
- iter != mBindings.end(); ++iter)
+ mQueue->enqueueWork(new GetAllBindings(cb, this));
+}
+
+class GetAORBindings : public Work
+{
+public:
+ GetAORBindings(
+ const AMD_Registrar_getAORBindingsPtr& cb,
+ const std::string &aor,
+ const RegistrarIPtr& registrar)
+ : mCB(cb), mAOR(aor), mRegistrar(registrar) { }
+
+ void execute()
{
- BindingSeq bindings;
- for (BindingWrapperSeq::iterator seqIter = iter->second.begin();
- seqIter != iter->second.end(); ++seqIter)
+ BindingWrapperDict& regBindings = mRegistrar->getBindings();
+ BindingWrapperDict::iterator iter = regBindings.find(mAOR);
+ if (iter != regBindings.end())
{
- bindings.push_back((*seqIter)->mBinding);
+ BindingSeq bindings;
+ for (BindingWrapperSeq::iterator seqIter = iter->second.begin();
+ seqIter != iter->second.end(); ++seqIter)
+ {
+ bindings.push_back((*seqIter)->mBinding);
+ }
+ mCB->ice_response(bindings);
+ return;
}
- returnedBindings.insert(make_pair(iter->first, bindings));
+ mCB->ice_response(BindingSeq());
}
- cb->ice_response(returnedBindings);
-}
+
+private:
+ AMD_Registrar_getAORBindingsPtr mCB;
+ const std::string mAOR;
+ RegistrarIPtr mRegistrar;
+};
//XXX Queueable Operation
void RegistrarI::getAORBindings_async(
const AMD_Registrar_getAORBindingsPtr& cb,
- const std::string &aor,
+ const std::string& aor,
const Ice::Current&)
{
- BindingWrapperDict::iterator iter = mBindings.find(aor);
- if (iter != mBindings.end())
- {
- BindingSeq bindings;
- for (BindingWrapperSeq::iterator seqIter = iter->second.begin();
- seqIter != iter->second.end(); ++seqIter)
- {
- bindings.push_back((*seqIter)->mBinding);
- }
- cb->ice_response(bindings);
- return;
- }
- cb->ice_response(BindingSeq());
+ mQueue->enqueueWork(new GetAORBindings(cb, aor, this));
}
static BindingWrapperSeq Empty;
commit 162d8a50bf6df93a6bc24d06e9042b5026c088ee
Author: Mark Michelson <mmichelson at digium.com>
Date: Sun Jun 26 22:38:51 2011 -0500
Use AMD for Registrar operations.
I'm not actually using the work queue yet. That's next.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index 243c1e0..5363154 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -163,7 +163,10 @@ RegistrarI::~RegistrarI()
//XXX Queueable operation
-ContactDict RegistrarI::addListener(const RegistrarListenerPrx& listener, const Ice::Current&)
+void RegistrarI::addListener_async(
+ const AMD_Registrar_addListenerPtr& cb,
+ const RegistrarListenerPrx& listener,
+ const Ice::Current&)
{
if (std::find(mListeners.begin(), mListeners.end(), listener) == mListeners.end())
{
@@ -171,17 +174,25 @@ ContactDict RegistrarI::addListener(const RegistrarListenerPrx& listener, const
mListeners.push_back(listener);
}
//We should maybe throw an exception if someone tries to insert a duplicate?
- return mContacts;
+ //XXX Are we ever even populating mContacts??
+ cb->ice_response(mContacts);
+ return;
}
//XXX Queueable operation
-void RegistrarI::removeListener(const RegistrarListenerPrx& listener, const Ice::Current&)
+void RegistrarI::removeListener_async(
+ const AMD_Registrar_removeListenerPtr& cb,
+ const RegistrarListenerPrx& listener,
+ const Ice::Current&)
{
mListeners.erase(std::remove(mListeners.begin(), mListeners.end(), listener));
+ cb->ice_response();
}
//XXX Queueable operation
-BindingDict RegistrarI::getAllBindings(const Ice::Current&)
+void RegistrarI::getAllBindings_async(
+ const AMD_Registrar_getAllBindingsPtr& cb,
+ const Ice::Current&)
{
BindingDict returnedBindings;
//XXX
@@ -200,11 +211,14 @@ BindingDict RegistrarI::getAllBindings(const Ice::Current&)
}
returnedBindings.insert(make_pair(iter->first, bindings));
}
- return returnedBindings;
+ cb->ice_response(returnedBindings);
}
//XXX Queueable Operation
-BindingSeq RegistrarI::getAORBindings(const std::string &aor, const Ice::Current&)
+void RegistrarI::getAORBindings_async(
+ const AMD_Registrar_getAORBindingsPtr& cb,
+ const std::string &aor,
+ const Ice::Current&)
{
BindingWrapperDict::iterator iter = mBindings.find(aor);
if (iter != mBindings.end())
@@ -215,9 +229,10 @@ BindingSeq RegistrarI::getAORBindings(const std::string &aor, const Ice::Current
{
bindings.push_back((*seqIter)->mBinding);
}
- return bindings;
+ cb->ice_response(bindings);
+ return;
}
- return BindingSeq();
+ cb->ice_response(BindingSeq());
}
static BindingWrapperSeq Empty;
diff --git a/src/PJSipRegistrarModule.h b/src/PJSipRegistrarModule.h
index 28ac125..76b422a 100644
--- a/src/PJSipRegistrarModule.h
+++ b/src/PJSipRegistrarModule.h
@@ -39,10 +39,25 @@ class RegistrarI : public AsteriskSCF::SIP::Registration::V1::Registrar
public:
RegistrarI(const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& defaultListener);
~RegistrarI();
- AsteriskSCF::SIP::Registration::V1::ContactDict addListener(const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& listener, const Ice::Current&);
- void removeListener(const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& listener, const Ice::Current&);
- AsteriskSCF::SIP::Registration::V1::BindingDict getAllBindings(const Ice::Current&);
- AsteriskSCF::SIP::Registration::V1::BindingSeq getAORBindings(const std::string &aor, const Ice::Current&);
+
+ void addListener_async(
+ const AsteriskSCF::SIP::Registration::V1::AMD_Registrar_addListenerPtr& cb,
+ const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& listener,
+ const Ice::Current&);
+
+ void removeListener_async(
+ const AsteriskSCF::SIP::Registration::V1::AMD_Registrar_removeListenerPtr& cb,
+ const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& listener,
+ const Ice::Current&);
+
+ void getAllBindings_async(
+ const AsteriskSCF::SIP::Registration::V1::AMD_Registrar_getAllBindingsPtr& cb,
+ const Ice::Current&);
+
+ void getAORBindings_async(
+ const AsteriskSCF::SIP::Registration::V1::AMD_Registrar_getAORBindingsPtr& cb,
+ const std::string &aor,
+ const Ice::Current&);
BindingWrapperSeq& getAORBindingWrappers(const std::string &aor);
commit 41e944174785ff3370b4e165d491e8ff92cc0827
Author: Mark Michelson <mmichelson at digium.com>
Date: Sun Jun 26 21:49:05 2011 -0500
Switch the workQueue to live on the registrar instead of the PJSIP module.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index 15f7897..243c1e0 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -17,6 +17,8 @@
#include <AsteriskSCF/SIP/SIPRegistrarIf.h>
#include <AsteriskSCF/logger.h>
#include <AsteriskSCF/System/WorkQueue/WorkQueueIf.h>
+#include <AsteriskSCF/WorkQueue/WorkQueue.h>
+#include <AsteriskSCF/WorkQueue/DefaultQueueListener.h>
#include "PJSipRegistrarModule.h"
@@ -66,7 +68,7 @@ static void registrationExpired(pj_timer_heap_t *, struct pj_timer_entry *entry)
BindingWrapper *wrapper = static_cast<BindingWrapper *>(entry->user_data);
//Create a BindingWrapperPtr so that when we remove the wrapper from the
//registrar, it doesn't get deleted.
- wrapper->mRegModule->getQueue()->enqueueWork(new RegistrationExpired(wrapper));
+ wrapper->mRegModule->getRegistrar()->getQueue()->enqueueWork(new RegistrationExpired(wrapper));
}
static int expirationId;
@@ -142,9 +144,24 @@ void BindingWrapper::updateBinding(const std::string& callID, int cSeq, int expi
RegistrarI::RegistrarI(const RegistrarListenerPrx& defaultListener)
{
lg(Debug) << "In RegistrarI constructor, should be adding a listener...";
+ mQueue = new AsteriskSCF::WorkQueue::WorkQueue();
+ AsteriskSCF::WorkQueue::DefaultQueueListenerPtr listener(new AsteriskSCF::WorkQueue::DefaultQueueListener(mQueue));
+ mQueue->setListener(listener);
mListeners.push_back(defaultListener);
}
+RegistrarI::~RegistrarI()
+{
+ try
+ {
+ mQueue->shutdown();
+ }
+ catch (const AsteriskSCF::System::WorkQueue::V1::ShuttingDown&)
+ {
+ }
+}
+
+
//XXX Queueable operation
ContactDict RegistrarI::addListener(const RegistrarListenerPrx& listener, const Ice::Current&)
{
@@ -252,7 +269,7 @@ std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx>& Registrar
return mListeners;
}
-QueuePtr PJSipRegistrarModule::getQueue()
+QueuePtr RegistrarI::getQueue()
{
return mQueue;
}
@@ -606,7 +623,7 @@ pj_bool_t PJSipRegistrarModule::on_rx_request(pjsip_rx_data *rdata)
}
else
{
- (*bindingToUpdate)->updateBinding(callID, cSeq, expiration, mQueue);
+ (*bindingToUpdate)->updateBinding(callID, cSeq, expiration, mRegistrar->getQueue());
lg(Debug) << "Maintaining " << (*bindingToUpdate)->mBinding->contact << " in our existing bindings";
existingBindings.push_back(*bindingToUpdate);
}
@@ -619,7 +636,7 @@ pj_bool_t PJSipRegistrarModule::on_rx_request(pjsip_rx_data *rdata)
}
}
- mRegistrar->updateBindings(aor, newBindings, removedBindings, mQueue);
+ mRegistrar->updateBindings(aor, newBindings, removedBindings, mRegistrar->getQueue());
pjsip_tx_data *tdata;
pjsip_endpt_create_response(tsx->endpt, rdata, 200, NULL, &tdata);
diff --git a/src/PJSipRegistrarModule.h b/src/PJSipRegistrarModule.h
index 9637f5c..28ac125 100644
--- a/src/PJSipRegistrarModule.h
+++ b/src/PJSipRegistrarModule.h
@@ -38,6 +38,7 @@ class RegistrarI : public AsteriskSCF::SIP::Registration::V1::Registrar
{
public:
RegistrarI(const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& defaultListener);
+ ~RegistrarI();
AsteriskSCF::SIP::Registration::V1::ContactDict addListener(const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& listener, const Ice::Current&);
void removeListener(const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& listener, const Ice::Current&);
AsteriskSCF::SIP::Registration::V1::BindingDict getAllBindings(const Ice::Current&);
@@ -62,12 +63,16 @@ public:
AsteriskSCF::SIP::Registration::V1::BindingDict createBindingDict(const std::string& aor, const BindingWrapperSeq& bindings);
std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx>& getListeners();
+
BindingWrapperDict& getBindings();
+ AsteriskSCF::System::WorkQueue::V1::QueuePtr getQueue();
+
private:
BindingWrapperDict mBindings;
AsteriskSCF::SIP::Registration::V1::ContactDict mContacts;
std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx> mListeners;
+ AsteriskSCF::System::WorkQueue::V1::QueuePtr mQueue;
};
typedef IceUtil::Handle<RegistrarI> RegistrarIPtr;
@@ -79,7 +84,6 @@ public:
pjsip_endpoint *endpt,
const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& defaultListener,
const AsteriskSCF::Discovery::SmartProxy<AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx>& stateReplicator);
- ~PJSipRegistrarModule();
pj_status_t load(pjsip_endpoint *endpoint);
pj_status_t start();
pj_status_t stop();
@@ -91,8 +95,6 @@ public:
void on_tsx_state(pjsip_transaction *tsx, pjsip_event *event);
RegistrarIPtr getRegistrar();
- AsteriskSCF::System::WorkQueue::V1::QueuePtr getQueue();
-
/**
* Replicate state of bindings currently held by this registrar.
*
@@ -136,7 +138,6 @@ private:
pjsip_endpoint *mEndpoint;
RegistrarIPtr mRegistrar;
AsteriskSCF::Discovery::SmartProxy<AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx> mStateReplicator;
- AsteriskSCF::System::WorkQueue::V1::QueuePtr mQueue;
};
typedef IceUtil::Handle<PJSipRegistrarModule> PJSipRegistrarModulePtr;
diff --git a/src/PJSipRegistrarModuleConstruction.cpp b/src/PJSipRegistrarModuleConstruction.cpp
index 660ccd6..45958bc 100644
--- a/src/PJSipRegistrarModuleConstruction.cpp
+++ b/src/PJSipRegistrarModuleConstruction.cpp
@@ -16,11 +16,7 @@
#include "PJSipRegistrarModule.h"
-#include <AsteriskSCF/WorkQueue/WorkQueue.h>
-#include <AsteriskSCF/WorkQueue/DefaultQueueListener.h>
-
using namespace AsteriskSCF::SIP::Registration::V1;
-using namespace AsteriskSCF::WorkQueue;
namespace AsteriskSCF
{
@@ -101,22 +97,8 @@ PJSipRegistrarModule::PJSipRegistrarModule(
mModule.on_tx_response = registrarOnTxResponse;
mModule.on_tsx_state = registrarOnTsxState;
- mQueue = new AsteriskSCF::WorkQueue::WorkQueue();
- AsteriskSCF::WorkQueue::DefaultQueueListenerPtr listener(new AsteriskSCF::WorkQueue::DefaultQueueListener(mQueue));
- mQueue->setListener(listener);
pjsip_endpt_register_module(mEndpoint, &mModule);
}
-PJSipRegistrarModule::~PJSipRegistrarModule()
-{
- try
- {
- mQueue->shutdown();
- }
- catch (const AsteriskSCF::System::WorkQueue::V1::ShuttingDown&)
- {
- }
-}
-
};
};
diff --git a/src/SipStateReplicatorListener.cpp b/src/SipStateReplicatorListener.cpp
index 8a3dd2a..c25b800 100644
--- a/src/SipStateReplicatorListener.cpp
+++ b/src/SipStateReplicatorListener.cpp
@@ -126,7 +126,7 @@ public:
//that we've already expired and removed. No biggie. Just
//keep going.
}
- regModule->getRegistrar()->updateBindings(dictIter->first, emptySeq, toRemove, regModule->getQueue());
+ regModule->getRegistrar()->updateBindings(dictIter->first, emptySeq, toRemove, regModule->getRegistrar()->getQueue());
}
}
@@ -166,7 +166,7 @@ public:
BindingWrapperPtr wrapper = createBindingWrapper(dictIter->first, *seqIter, regModule);
wrapperSeq.push_back(wrapper);
}
- regModule->getRegistrar()->updateBindings(dictIter->first, wrapperSeq, emptySeq, regModule->getQueue());
+ regModule->getRegistrar()->updateBindings(dictIter->first, wrapperSeq, emptySeq, regModule->getRegistrar()->getQueue());
}
}
@@ -198,12 +198,12 @@ public:
}
else
{
- (*bindingToUpdate)->updateBinding((*seqIter)->callid, (*seqIter)->cseq, (*seqIter)->expiration - time(NULL), regModule->getQueue());
+ (*bindingToUpdate)->updateBinding((*seqIter)->callid, (*seqIter)->cseq, (*seqIter)->expiration - time(NULL), regModule->getRegistrar()->getQueue());
}
}
if (!newBindings.empty())
{
- regModule->getRegistrar()->updateBindings(dictIter->first, newBindings, emptySeq, regModule->getQueue());
+ regModule->getRegistrar()->updateBindings(dictIter->first, newBindings, emptySeq, regModule->getRegistrar()->getQueue());
}
}
}
commit f860396ac01e1581dfe5f56ca30434c8b01b4270
Author: Mark Michelson <mmichelson at digium.com>
Date: Sun Jun 26 21:38:30 2011 -0500
Initial moves toward using a work queue in the registrar.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index eba02df..15f7897 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -16,11 +16,13 @@
#include <AsteriskSCF/SIP/SIPRegistrarIf.h>
#include <AsteriskSCF/logger.h>
+#include <AsteriskSCF/System/WorkQueue/WorkQueueIf.h>
#include "PJSipRegistrarModule.h"
using namespace AsteriskSCF::SIP::Registration::V1;
using namespace AsteriskSCF::System::Logging;
+using namespace AsteriskSCF::System::WorkQueue::V1;
namespace
{
@@ -33,26 +35,38 @@ namespace AsteriskSCF
namespace SipSessionManager
{
-//XXX QueueableOperation
+class RegistrationExpired : public Work
+{
+public:
+ RegistrationExpired(const BindingWrapperPtr& wrapper)
+ : mWrapper(wrapper) { }
+
+ void execute()
+ {
+ lg(Debug) << "Detected expiration of binding " << mWrapper->mBinding->contact;
+ mWrapper->cancelRegistrationExpiration();
+ RegistrarIPtr registrar = mWrapper->mRegModule->getRegistrar();
+ BindingWrapperSeq& currentBindings = registrar->getAORBindingWrappers(mWrapper->mAOR);
+ registrar->removeBinding(currentBindings, mWrapper);
+ if (currentBindings.empty())
+ {
+ registrar->removeAOR(mWrapper->mAOR);
+ }
+ BindingWrapperSeq emptySeq;
+ BindingWrapperSeq toRemove;
+ toRemove.push_back(mWrapper);
+ mWrapper->mRegModule->replicateState(mWrapper->mAOR, emptySeq, emptySeq, toRemove);
+ }
+private:
+ BindingWrapperPtr mWrapper;
+};
+
static void registrationExpired(pj_timer_heap_t *, struct pj_timer_entry *entry)
{
BindingWrapper *wrapper = static_cast<BindingWrapper *>(entry->user_data);
//Create a BindingWrapperPtr so that when we remove the wrapper from the
//registrar, it doesn't get deleted.
- BindingWrapperPtr wrapperPtr(new BindingWrapper(*wrapper));
- lg(Debug) << "Detected expiration of binding " << wrapperPtr->mBinding->contact;
- wrapperPtr->cancelRegistrationExpiration();
- RegistrarIPtr registrar = wrapperPtr->mRegModule->getRegistrar();
- BindingWrapperSeq& currentBindings = registrar->getAORBindingWrappers(wrapperPtr->mAOR);
- registrar->removeBinding(currentBindings, wrapperPtr);
- if (currentBindings.empty())
- {
- registrar->removeAOR(wrapperPtr->mAOR);
- }
- BindingWrapperSeq emptySeq;
- BindingWrapperSeq toRemove;
- toRemove.push_back(wrapperPtr);
- wrapperPtr->mRegModule->replicateState(wrapperPtr->mAOR, emptySeq, emptySeq, toRemove);
+ wrapper->mRegModule->getQueue()->enqueueWork(new RegistrationExpired(wrapper));
}
static int expirationId;
@@ -97,15 +111,32 @@ void BindingWrapper::scheduleRegistrationExpiration(int expiration)
pjsip_endpt_schedule_timer(mEndpoint, &mEntry, &delay);
}
-//XXX Queueable operation
-void BindingWrapper::updateBinding(const std::string &callID, int cSeq, int expiration)
+class UpdateBinding : public Work
{
- mBinding->callid = callID;
- mBinding->cseq = cSeq;
- mBinding->expiration = time(NULL) + expiration;
- cancelRegistrationExpiration();
- lg(Debug) << "Renewing binding " << mBinding->contact << " for another " << expiration << " seconds.";
- scheduleRegistrationExpiration(expiration);
+public:
+ UpdateBinding(const BindingWrapperPtr& wrapper, const std::string& callID, int cSeq, int expiration)
+ : mWrapper(wrapper), mCallID(callID), mCSeq(cSeq), mExpiration(expiration) { }
+
+ void execute()
+ {
+ mWrapper->mBinding->callid = mCallID;
+ mWrapper->mBinding->cseq = mCSeq;
+ mWrapper->mBinding->expiration = time(NULL) + mExpiration;
+ mWrapper->cancelRegistrationExpiration();
+ lg(Debug) << "Renewing binding " << mWrapper->mBinding->contact << " for another " << mExpiration << " seconds.";
+ mWrapper->scheduleRegistrationExpiration(mExpiration);
+ }
+
+private:
+ BindingWrapperPtr mWrapper;
+ const std::string mCallID;
+ const int mCSeq;
+ const int mExpiration;
+};
+
+void BindingWrapper::updateBinding(const std::string& callID, int cSeq, int expiration, const QueuePtr& queue)
+{
+ queue->enqueueWork(new UpdateBinding(this, callID, cSeq, expiration));
}
RegistrarI::RegistrarI(const RegistrarListenerPrx& defaultListener)
@@ -211,6 +242,21 @@ BindingDict RegistrarI::createBindingDict(const std::string& aor, const BindingW
return dict;
}
+BindingWrapperDict& RegistrarI::getBindings()
+{
+ return mBindings;
+}
+
+std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx>& RegistrarI::getListeners()
+{
+ return mListeners;
+}
+
+QueuePtr PJSipRegistrarModule::getQueue()
+{
+ return mQueue;
+}
+
void PJSipRegistrarModule::replicateState(
const std::string &aor,
const BindingWrapperSeq& existingBindings,
@@ -272,49 +318,72 @@ void PJSipRegistrarModule::replicateState(
}
}
-//XXX Potential queueable operation
-void RegistrarI::updateBindings(const std::string &aor, BindingWrapperSeq& newBindings, BindingWrapperSeq& removedBindings)
+class UpdateBindings : public Work
{
+public:
+ UpdateBindings(
+ const RegistrarIPtr& registrar,
+ const std::string& aor,
+ const BindingWrapperSeq& newBindings,
+ const BindingWrapperSeq& removedBindings)
+ : mRegistrar(registrar), mAOR(aor), mNewBindings(newBindings), mRemovedBindings(removedBindings) {}
- BindingWrapperDict::iterator aorBindings = mBindings.find(aor);
-
- if (aorBindings == mBindings.end())
- {
- if (!newBindings.empty())
- {
- lg(Debug) << "Adding new bindings for aor " << aor;
- mBindings.insert(make_pair(aor, newBindings));
- }
- }
- else
+ void execute()
{
- BindingWrapperSeq& currentBindings = aorBindings->second;
- if (!newBindings.empty())
+ BindingWrapperDict& regBindings = mRegistrar->getBindings();
+ BindingWrapperDict::iterator aorBindings = regBindings.find(mAOR);
+
+ if (aorBindings == regBindings.end())
{
- lg(Debug) << "Adding new bindings for aor " << aor;
- currentBindings.insert(currentBindings.end(), newBindings.begin(), newBindings.end());
+ if (!mNewBindings.empty())
+ {
+ lg(Debug) << "Adding new bindings for aor " << mAOR;
+ regBindings.insert(make_pair(mAOR, mNewBindings));
+ }
}
- for (BindingWrapperSeq::iterator iter = removedBindings.begin(); iter != removedBindings.end(); ++iter)
+ else
{
- removeBinding(currentBindings, *iter);
+ BindingWrapperSeq& currentBindings = aorBindings->second;
+ if (!mNewBindings.empty())
+ {
+ lg(Debug) << "Adding new bindings for aor " << mAOR;
+ currentBindings.insert(currentBindings.end(), mNewBindings.begin(), mNewBindings.end());
+ }
+ for (BindingWrapperSeq::iterator iter = mRemovedBindings.begin(); iter != mRemovedBindings.end(); ++iter)
+ {
+ mRegistrar->removeBinding(currentBindings, *iter);
+ }
+ if (currentBindings.empty())
+ {
+ mRegistrar->removeAOR(mAOR);
+ }
}
- if (currentBindings.empty())
+
+ // We have our internal stuff all sorted out. Now we need to notify listeners and replicas.
+ BindingUpdateSeq newBindingUpdateSeq(mRegistrar->createBindingUpdateSeq(mAOR, mNewBindings));
+ BindingUpdateSeq removedBindingUpdateSeq(mRegistrar->createBindingUpdateSeq(mAOR, mRemovedBindings));
+ std::vector<RegistrarListenerPrx>& listeners = mRegistrar->getListeners();
+ for (std::vector<RegistrarListenerPrx>::iterator iter = listeners.begin();
+ iter != listeners.end(); ++iter)
{
- removeAOR(aor);
+ //Listeners are only concerned with new and removed bindings. There's no need to tell them of existing ones.
+ lg(Debug) << "Alerting listener " << (*iter)->ice_getIdentity().name << " about changes to bindings";
+ (*iter)->contactsAdded(newBindingUpdateSeq);
+ (*iter)->contactsRemoved(removedBindingUpdateSeq);
}
}
+private:
+ RegistrarIPtr mRegistrar;
+ const std::string mAOR;
+ BindingWrapperSeq mNewBindings;
+ BindingWrapperSeq mRemovedBindings;
+};
- // We have our internal stuff all sorted out. Now we need to notify listeners and replicas.
- BindingUpdateSeq newBindingUpdateSeq(createBindingUpdateSeq(aor, newBindings));
- BindingUpdateSeq removedBindingUpdateSeq(createBindingUpdateSeq(aor, removedBindings));
- for (std::vector<RegistrarListenerPrx>::iterator iter = mListeners.begin();
- iter != mListeners.end(); ++iter)
- {
- //Listeners are only concerned with new and removed bindings. There's no need to tell them of existing ones.
- lg(Debug) << "Alerting listener " << (*iter)->ice_getIdentity().name << " about changes to bindings";
- (*iter)->contactsAdded(newBindingUpdateSeq);
- (*iter)->contactsRemoved(removedBindingUpdateSeq);
- }
+//XXX Potential queueable operation
+void RegistrarI::updateBindings(const std::string &aor, BindingWrapperSeq& newBindings, BindingWrapperSeq& removedBindings, const QueuePtr& queue)
+{
+
+ queue->enqueueWork(new UpdateBindings(this, aor, newBindings, removedBindings));
}
BindingUpdateSeq RegistrarI::createBindingUpdateSeq(const std::string& aor, BindingWrapperSeq& bindings)
@@ -537,7 +606,7 @@ pj_bool_t PJSipRegistrarModule::on_rx_request(pjsip_rx_data *rdata)
}
else
{
- (*bindingToUpdate)->updateBinding(callID, cSeq, expiration);
+ (*bindingToUpdate)->updateBinding(callID, cSeq, expiration, mQueue);
lg(Debug) << "Maintaining " << (*bindingToUpdate)->mBinding->contact << " in our existing bindings";
existingBindings.push_back(*bindingToUpdate);
}
@@ -550,7 +619,7 @@ pj_bool_t PJSipRegistrarModule::on_rx_request(pjsip_rx_data *rdata)
}
}
- mRegistrar->updateBindings(aor, newBindings, removedBindings);
+ mRegistrar->updateBindings(aor, newBindings, removedBindings, mQueue);
pjsip_tx_data *tdata;
pjsip_endpt_create_response(tsx->endpt, rdata, 200, NULL, &tdata);
diff --git a/src/PJSipRegistrarModule.h b/src/PJSipRegistrarModule.h
index 005657b..9637f5c 100644
--- a/src/PJSipRegistrarModule.h
+++ b/src/PJSipRegistrarModule.h
@@ -30,7 +30,7 @@ namespace SipSessionManager
{
class BindingWrapper;
-typedef boost::shared_ptr<BindingWrapper> BindingWrapperPtr;
+typedef IceUtil::Handle<BindingWrapper> BindingWrapperPtr;
typedef std::vector<BindingWrapperPtr> BindingWrapperSeq;
typedef std::map<std::string, BindingWrapperSeq> BindingWrapperDict;
@@ -51,7 +51,8 @@ public:
void updateBindings(
const std::string &aor,
BindingWrapperSeq& newBindings,
- BindingWrapperSeq& removedBindings);
+ BindingWrapperSeq& removedBindings,
+ const AsteriskSCF::System::WorkQueue::V1::QueuePtr& queue);
AsteriskSCF::SIP::Registration::V1::BindingUpdateSeq createBindingUpdateSeq(
const std::string& aor,
@@ -60,6 +61,9 @@ public:
AsteriskSCF::SIP::Registration::V1::BindingDict createBindingDict(const std::string& aor, const BindingWrapperSeq& bindings);
+ std::vector<AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx>& getListeners();
+ BindingWrapperDict& getBindings();
+
private:
BindingWrapperDict mBindings;
AsteriskSCF::SIP::Registration::V1::ContactDict mContacts;
@@ -87,6 +91,8 @@ public:
void on_tsx_state(pjsip_transaction *tsx, pjsip_event *event);
RegistrarIPtr getRegistrar();
+ AsteriskSCF::System::WorkQueue::V1::QueuePtr getQueue();
+
/**
* Replicate state of bindings currently held by this registrar.
*
@@ -135,7 +141,7 @@ private:
typedef IceUtil::Handle<PJSipRegistrarModule> PJSipRegistrarModulePtr;
-class BindingWrapper
+class BindingWrapper : public IceUtil::Shared
{
public:
BindingWrapper(
@@ -164,7 +170,7 @@ public:
* Update the binding in the wrapper with new information. This will
* result in rescheduling destruction of the binding.
*/
- void updateBinding(const std::string &callID, int cSeq, int expiration);
+ void updateBinding(const std::string &callID, int cSeq, int expiration, const AsteriskSCF::System::WorkQueue::V1::QueuePtr& queue);
bool operator==(const BindingWrapper& rhs);
bool operator==(const AsteriskSCF::SIP::Registration::V1::BindingPtr& rhs);
diff --git a/src/SipStateReplicatorListener.cpp b/src/SipStateReplicatorListener.cpp
index bf52fe9..8a3dd2a 100644
--- a/src/SipStateReplicatorListener.cpp
+++ b/src/SipStateReplicatorListener.cpp
@@ -126,7 +126,7 @@ public:
//that we've already expired and removed. No biggie. Just
//keep going.
}
- regModule->getRegistrar()->updateBindings(dictIter->first, emptySeq, toRemove);
+ regModule->getRegistrar()->updateBindings(dictIter->first, emptySeq, toRemove, regModule->getQueue());
}
}
@@ -166,7 +166,7 @@ public:
BindingWrapperPtr wrapper = createBindingWrapper(dictIter->first, *seqIter, regModule);
wrapperSeq.push_back(wrapper);
}
- regModule->getRegistrar()->updateBindings(dictIter->first, wrapperSeq, emptySeq);
+ regModule->getRegistrar()->updateBindings(dictIter->first, wrapperSeq, emptySeq, regModule->getQueue());
}
}
@@ -198,12 +198,12 @@ public:
}
else
{
- (*bindingToUpdate)->updateBinding((*seqIter)->callid, (*seqIter)->cseq, (*seqIter)->expiration - time(NULL));
+ (*bindingToUpdate)->updateBinding((*seqIter)->callid, (*seqIter)->cseq, (*seqIter)->expiration - time(NULL), regModule->getQueue());
}
}
if (!newBindings.empty())
{
- regModule->getRegistrar()->updateBindings(dictIter->first, newBindings, emptySeq);
+ regModule->getRegistrar()->updateBindings(dictIter->first, newBindings, emptySeq, regModule->getQueue());
}
}
}
commit be2641c8edb8a41556c4a0d9c2d86fb023917792
Author: Mark Michelson <mmichelson at digium.com>
Date: Sat Jun 25 17:58:15 2011 -0500
Mark potential queuable operations.
diff --git a/src/PJSipRegistrarModule.cpp b/src/PJSipRegistrarModule.cpp
index 7e52733..eba02df 100644
--- a/src/PJSipRegistrarModule.cpp
+++ b/src/PJSipRegistrarModule.cpp
@@ -33,6 +33,7 @@ namespace AsteriskSCF
namespace SipSessionManager
{
+//XXX QueueableOperation
static void registrationExpired(pj_timer_heap_t *, struct pj_timer_entry *entry)
{
BindingWrapper *wrapper = static_cast<BindingWrapper *>(entry->user_data);
@@ -96,6 +97,7 @@ void BindingWrapper::scheduleRegistrationExpiration(int expiration)
pjsip_endpt_schedule_timer(mEndpoint, &mEntry, &delay);
}
+//XXX Queueable operation
void BindingWrapper::updateBinding(const std::string &callID, int cSeq, int expiration)
{
mBinding->callid = callID;
@@ -112,6 +114,7 @@ RegistrarI::RegistrarI(const RegistrarListenerPrx& defaultListener)
mListeners.push_back(defaultListener);
}
+//XXX Queueable operation
ContactDict RegistrarI::addListener(const RegistrarListenerPrx& listener, const Ice::Current&)
{
if (std::find(mListeners.begin(), mListeners.end(), listener) == mListeners.end())
@@ -123,11 +126,13 @@ ContactDict RegistrarI::addListener(const RegistrarListenerPrx& listener, const
return mContacts;
}
+//XXX Queueable operation
void RegistrarI::removeListener(const RegistrarListenerPrx& listener, const Ice::Current&)
{
mListeners.erase(std::remove(mListeners.begin(), mListeners.end(), listener));
}
+//XXX Queueable operation
BindingDict RegistrarI::getAllBindings(const Ice::Current&)
{
BindingDict returnedBindings;
@@ -150,6 +155,7 @@ BindingDict RegistrarI::getAllBindings(const Ice::Current&)
return returnedBindings;
}
+//XXX Queueable Operation
BindingSeq RegistrarI::getAORBindings(const std::string &aor, const Ice::Current&)
{
BindingWrapperDict::iterator iter = mBindings.find(aor);
@@ -266,6 +272,7 @@ void PJSipRegistrarModule::replicateState(
}
}
+//XXX Potential queueable operation
void RegistrarI::updateBindings(const std::string &aor, BindingWrapperSeq& newBindings, BindingWrapperSeq& removedBindings)
{
commit 1ea02fa8ec59f386d62f04a0ef3c1adb77fd5319
Author: Mark Michelson <mmichelson at digium.com>
Date: Sat Jun 25 17:52:34 2011 -0500
Add a WorkQueue onto the registrar module.
diff --git a/src/PJSipRegistrarModule.h b/src/PJSipRegistrarModule.h
index 8607570..005657b 100644
--- a/src/PJSipRegistrarModule.h
+++ b/src/PJSipRegistrarModule.h
@@ -18,6 +18,7 @@
#include <AsteriskSCF/Discovery/SmartProxy.h>
#include <AsteriskSCF/SIP/SIPRegistrarIf.h>
+#include <AsteriskSCF/System/WorkQueue/WorkQueueIf.h>
#include "SipStateReplicator.h"
#include "PJSipModule.h"
@@ -74,6 +75,7 @@ public:
pjsip_endpoint *endpt,
const AsteriskSCF::SIP::Registration::V1::RegistrarListenerPrx& defaultListener,
const AsteriskSCF::Discovery::SmartProxy<AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx>& stateReplicator);
+ ~PJSipRegistrarModule();
pj_status_t load(pjsip_endpoint *endpoint);
pj_status_t start();
pj_status_t stop();
@@ -128,6 +130,7 @@ private:
pjsip_endpoint *mEndpoint;
RegistrarIPtr mRegistrar;
AsteriskSCF::Discovery::SmartProxy<AsteriskSCF::Replication::SipSessionManager::V1::SipStateReplicatorPrx> mStateReplicator;
+ AsteriskSCF::System::WorkQueue::V1::QueuePtr mQueue;
};
typedef IceUtil::Handle<PJSipRegistrarModule> PJSipRegistrarModulePtr;
diff --git a/src/PJSipRegistrarModuleConstruction.cpp b/src/PJSipRegistrarModuleConstruction.cpp
index 46a3b10..660ccd6 100644
--- a/src/PJSipRegistrarModuleConstruction.cpp
+++ b/src/PJSipRegistrarModuleConstruction.cpp
@@ -16,7 +16,11 @@
#include "PJSipRegistrarModule.h"
+#include <AsteriskSCF/WorkQueue/WorkQueue.h>
+#include <AsteriskSCF/WorkQueue/DefaultQueueListener.h>
+
using namespace AsteriskSCF::SIP::Registration::V1;
+using namespace AsteriskSCF::WorkQueue;
namespace AsteriskSCF
{
@@ -96,8 +100,23 @@ PJSipRegistrarModule::PJSipRegistrarModule(
mModule.on_tx_request = registrarOnTxRequest;
mModule.on_tx_response = registrarOnTxResponse;
mModule.on_tsx_state = registrarOnTsxState;
+
+ mQueue = new AsteriskSCF::WorkQueue::WorkQueue();
+ AsteriskSCF::WorkQueue::DefaultQueueListenerPtr listener(new AsteriskSCF::WorkQueue::DefaultQueueListener(mQueue));
+ mQueue->setListener(listener);
pjsip_endpt_register_module(mEndpoint, &mModule);
}
+PJSipRegistrarModule::~PJSipRegistrarModule()
+{
+ try
+ {
+ mQueue->shutdown();
+ }
+ catch (const AsteriskSCF::System::WorkQueue::V1::ShuttingDown&)
+ {
+ }
+}
+
};
};
commit 3fa68dd21a3f9a62ca80400e48832d0e6225a2ea
Merge: fc2f64b c977ec5
Author: Mark Michelson <mmichelson at digium.com>
Date: Sat Jun 25 17:33:56 2011 -0500
Merge branch 'master' into registrar
Conflicts:
config/SipConfigurator.py
commit fc2f64b78c4c57c4c3b8fea3233a9fd3f52147f6
Merge: 575abed 4cc73e7
Author: Mark Michelson <mmichelson at digium.com>
Date: Sat Jun 25 16:21:37 2011 -0500
Merge branch 'master' into registrar
commit 575abed0a58ba2e9f321b8ccc121a32356c7288d
Author: Mark Michelson <mmichelson at digium.com>
Date: Fri Jun 24 16:18:55 2011 -0500
Adjust for changes made by build system reorganization.
diff --git a/config/SipConfigurator.py b/config/SipConfigurator.py
index 980cded..31793e8 100755
--- a/config/SipConfigurator.py
+++ b/config/SipConfigurator.py
@@ -22,23 +22,23 @@
import Ice, Configurator, sys
# Load our component specific configuration definitions
-Ice.loadSlice('-I. -I/opt/Ice-3.4.1/slice -I../../slice --all ../local-slice/SipConfigurationIf.ice')
-import AsteriskSCF.SIP.V1
+Ice.loadSlice('-I. -I/opt/Ice-3.4.1/slice -I../../slice/slice --all ../slice/AsteriskSCF/Configuration/SipSessionManager/SipConfigurationIf.ice')
+import AsteriskSCF.Configuration.SipSessionManager.V1
# Add our own visitor implementations for the sections we support
class SipSectionVisitors(Configurator.SectionVisitors):
def visit_general(self, config, section):
- group = AsteriskSCF.SIP.V1.SipGeneralGroup()
+ group = AsteriskSCF.Configuration.SipSessionManager.V1.SipGeneralGroup()
group.configurationItems = { }
self.groups.append(group)
def visit_transport_udp(self, config, section):
- group = AsteriskSCF.SIP.V1.SipUDPTransportGroup()
+ group = AsteriskSCF.Configuration.SipSessionManager.V1.SipUDPTransportGroup()
group.name = section
group.configurationItems = { }
mapper = Configurator.OptionMapper()
- item = AsteriskSCF.SIP.V1.SipHostItem()
+ item = AsteriskSCF.Configuration.SipSessionManager.V1.SipHostItem()
mapper.map('host', item, 'host', 'address', config.get, None)
mapper.map('port', item, 'port', 'address', config.getint, 5060)
for option in config.options(section):
@@ -49,12 +49,12 @@ class SipSectionVisitors(Configurator.SectionVisitors):
self.groups.append(group)
def visit_transport_tcp(self, config, section):
- group = AsteriskSCF.SIP.V1.SipTCPTransportGroup()
+ group = AsteriskSCF.Configuration.SipSessionManager.V1.SipTCPTransportGroup()
group.name = section
group.configurationItems = { }
mapper = Configurator.OptionMapper()
- item = AsteriskSCF.SIP.V1.SipHostItem()
+ item = AsteriskSCF.Configuration.SipSessionManager.V1.SipHostItem()
mapper.map('host', item, 'host', 'address', config.get, None)
mapper.map('port', item, 'port', 'address', config.getint, 5060)
for option in config.options(section):
@@ -65,12 +65,12 @@ class SipSectionVisitors(Configurator.SectionVisitors):
self.groups.append(group)
def visit_transport_tls(self, config, section):
- group = AsteriskSCF.SIP.V1.SipTLSTransportGroup()
+ group = AsteriskSCF.Configuration.SipSessionManager.V1.SipTLSTransportGroup()
group.name = section
group.configurationItems = { }
mapper = Configurator.OptionMapper()
- item = AsteriskSCF.SIP.V1.SipHostItem()
+ item = AsteriskSCF.Configuration.SipSessionManager.V1.SipHostItem()
mapper.map('host', item, 'host', 'address', config.get, None)
mapper.map('port', item, 'port', 'address', config.getint, 5060)
for option in config.options(section):
@@ -81,19 +81,19 @@ class SipSectionVisitors(Configurator.SectionVisitors):
self.groups.append(group)
def visit_endpoint(self, config, section):
- group = AsteriskSCF.SIP.V1.SipEndpointGroup()
+ group = AsteriskSCF.Configuration.SipSessionManager.V1.SipEndpointGroup()
group.name = section
group.configurationItems = { }
mapper = Configurator.OptionMapper()
- mapper.map('routing', AsteriskSCF.SIP.V1.SipRoutingItem(), 'routingServiceName', 'routingService', config.get, None)
+ mapper.map('routing', AsteriskSCF.Configuration.SipSessionManager.V1.SipRoutingItem(), 'routingServiceName', 'routingService', config.get, None)
... 6014 lines suppressed ...
--
asterisk-scf/integration/sip.git
More information about the asterisk-scf-commits
mailing list