[asterisk-scf-commits] asterisk-scf/integration/sip.git branch "threading" updated.
Commits to the Asterisk SCF project code repositories
asterisk-scf-commits at lists.digium.com
Tue May 10 18:14:12 CDT 2011
branch "threading" has been updated
via 78e5e7f41cc406ab6fd735cd3bc2684f2ededd38 (commit)
via 851a92b67b53c0f75bc703da381c4371d7aa0c23 (commit)
from 2e022453c47afd990d1f27e4403e813a45060e84 (commit)
Summary of changes:
src/PJSipSessionModule.cpp | 3 +-
src/SipSession.cpp | 272 +++++++++++++++++++++++++++++---------------
src/SipSession.h | 8 +-
3 files changed, 186 insertions(+), 97 deletions(-)
- Log -----------------------------------------------------------------
commit 78e5e7f41cc406ab6fd735cd3bc2684f2ededd38
Author: Mark Michelson <mmichelson at digium.com>
Date: Tue May 10 17:48:07 2011 -0500
Remove locking from SipSessions.
All uses of locking are being called from queued operations or
from places that I plan to turn into queued operations.
diff --git a/src/SipSession.cpp b/src/SipSession.cpp
index 3696ff5..eb7a450 100644
--- a/src/SipSession.cpp
+++ b/src/SipSession.cpp
@@ -131,14 +131,12 @@ public:
void addListener(const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener)
{
- boost::unique_lock<boost::shared_mutex> lock(mLock);
mListeners.push_back(listener);
if (mInviteSession)
{
PJSipSessionModInfo *session_mod_info =
static_cast<PJSipSessionModInfo*>(
mInviteSession->mod_data[mManager->getSessionModule()->getModule().id]);
- lock.unlock();
session_mod_info->updateSessionState(mInviteSession);
mManager->getSessionModule()->replicateState(NULL, NULL, session_mod_info);
}
@@ -242,11 +240,6 @@ public:
AsteriskSCF::System::Component::V1::ReplicaPtr mReplica;
SessionWorkPtr mSessionWork;
-
- /**
- * Shared mutex lock which protects some of the session.
- */
- boost::shared_mutex mLock;
};
/**
@@ -580,7 +573,6 @@ void SipSession::setBridge_async(
*/
void SipSession::setBridge(const AsteriskSCF::SessionCommunications::V1::BridgePrx& bridge)
{
- boost::unique_lock<boost::shared_mutex> lock(mImplPriv->mLock);
mImplPriv->mBridge = bridge;
}
@@ -1001,7 +993,6 @@ AsteriskSCF::Media::V1::StreamSinkSeq& SipSession::getSinks()
*/
std::vector<AsteriskSCF::SessionCommunications::V1::SessionListenerPrx> SipSession::getListeners()
{
- boost::shared_lock<boost::shared_mutex> lock(mImplPriv->mLock);
return mImplPriv->mListeners;
}
@@ -1034,7 +1025,6 @@ AsteriskSCF::Media::V1::SessionPrx& SipSession::getMediaSessionProxy()
*/
void SipSession::setListeners(const AsteriskSCF::SIP::V1::SessionListenerSeq& listeners)
{
- boost::unique_lock<boost::shared_mutex> lock(mImplPriv->mLock);
mImplPriv->mListeners = listeners;
}
commit 851a92b67b53c0f75bc703da381c4371d7aa0c23
Author: Mark Michelson <mmichelson at digium.com>
Date: Tue May 10 17:22:31 2011 -0500
Finish the remaining conversion to queueable session operations and use of AMD.
At this point everything compiles, but work could be done to ensure that everything
is implemented in the most logical way possible.
diff --git a/src/PJSipSessionModule.cpp b/src/PJSipSessionModule.cpp
index d2aa0a8..32ed923 100644
--- a/src/PJSipSessionModule.cpp
+++ b/src/PJSipSessionModule.cpp
@@ -260,8 +260,7 @@ void PJSipSessionModInfo::updateSessionState(pjsip_inv_session *inv_session)
mSessionState->mListeners = mSession->getListeners();
try
{
- Ice::Current current;
- mSessionState->mBridge = mSession->getBridge(current);
+ mSessionState->mBridge = mSession->getBridge();
}
catch (...)
{
diff --git a/src/SipSession.cpp b/src/SipSession.cpp
index caf5169..3696ff5 100644
--- a/src/SipSession.cpp
+++ b/src/SipSession.cpp
@@ -91,6 +91,82 @@ public:
: mAdapter(adapter), mDialog(0), mInviteSession(0), mEndpoint(endpoint), mDestination(destination),
mManager(manager), mServiceLocator(serviceLocator), mReplica(replica), mSessionWork(sessionWork) { };
+ AsteriskSCF::SessionCommunications::V1::SessionInfoPtr getInfo()
+ {
+ AsteriskSCF::SessionCommunications::V1::SessionInfoPtr sessioninfo =
+ new AsteriskSCF::SessionCommunications::V1::SessionInfo();
+
+ /* TODO: Utilize locking so this becomes safe
+ if (!mImplPriv->mInviteSession || mImplPriv->mInviteSession->state == PJSIP_INV_STATE_NULL)
+ {
+ sessioninfo->currentState = "ready";
+ }
+ else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_CALLING)
+ {
+ sessioninfo->currentState = "outbound";
+ }
+ else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_INCOMING)
+ {
+ sessioninfo->currentState = "inbound";
+ }
+ else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_EARLY)
+ {
+ sessioninfo->currentState = "early";
+ }
+ else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_CONNECTING)
+ {
+ sessioninfo->currentState = "connecting";
+ }
+ else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_CONFIRMED)
+ {
+ sessioninfo->currentState = "connected";
+ }
+ else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_DISCONNECTED)
+ {
+ sessioninfo->currentState = "disconnected";
+ }
+ */
+ return sessioninfo;
+ }
+
+ void addListener(const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener)
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+ mListeners.push_back(listener);
+ if (mInviteSession)
+ {
+ PJSipSessionModInfo *session_mod_info =
+ static_cast<PJSipSessionModInfo*>(
+ mInviteSession->mod_data[mManager->getSessionModule()->getModule().id]);
+ lock.unlock();
+ session_mod_info->updateSessionState(mInviteSession);
+ mManager->getSessionModule()->replicateState(NULL, NULL, session_mod_info);
+ }
+ }
+
+ void removeListener(const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener)
+ {
+ mListeners.erase(std::remove(mListeners.begin(), mListeners.end(),
+ listener), mListeners.end());
+ if (mInviteSession)
+ {
+ PJSipSessionModInfo *session_mod_info =
+ static_cast<PJSipSessionModInfo*>(
+ mInviteSession->mod_data[mManager->getSessionModule()->getModule().id]);
+ session_mod_info->updateSessionState(mInviteSession);
+ mManager->getSessionModule()->replicateState(NULL, NULL, session_mod_info);
+ }
+ }
+
+ AsteriskSCF::SessionCommunications::V1::BridgePrx getBridge()
+ {
+ if (mBridge == 0)
+ {
+ throw AsteriskSCF::SessionCommunications::V1::NotBridged();
+ }
+ return mBridge;
+ }
+
/**
* An instance of a media session.
*/
@@ -225,21 +301,35 @@ SipSession::SipSession(const Ice::ObjectAdapterPtr& adapter, const SipEndpointPt
mImplPriv->mSinks = sinks;
}
-AsteriskSCF::SessionCommunications::V1::SessionInfoPtr SipSession::addListener(
- const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener, const Ice::Current& current)
+class AddListenerOperation : public SuspendableWork
{
- boost::unique_lock<boost::shared_mutex> lock(mImplPriv->mLock);
- mImplPriv->mListeners.push_back(listener);
- if (mImplPriv->mInviteSession)
+public:
+ AddListenerOperation(
+ const AsteriskSCF::SessionCommunications::V1::AMD_Session_addListenerPtr& cb,
+ const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
+ const boost::shared_ptr<SipSessionPriv>& sessionPriv)
+ : mCb(cb), mListener(listener), mImplPriv(sessionPriv) { }
+
+ SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
{
- PJSipSessionModInfo *session_mod_info =
- static_cast<PJSipSessionModInfo*>(
- mImplPriv->mInviteSession->mod_data[mImplPriv->mManager->getSessionModule()->getModule().id]);
- lock.unlock();
- session_mod_info->updateSessionState(mImplPriv->mInviteSession);
- mImplPriv->mManager->getSessionModule()->replicateState(NULL, NULL, session_mod_info);
+ mImplPriv->addListener(mListener);
+ AsteriskSCF::SessionCommunications::V1::SessionInfoPtr sessionInfo =
+ mImplPriv->getInfo();
+ mCb->ice_response(sessionInfo);
+ return Complete;
}
- return getInfo(current);
+
+ AsteriskSCF::SessionCommunications::V1::AMD_Session_addListenerPtr mCb;
+ AsteriskSCF::SessionCommunications::V1::SessionListenerPrx mListener;
+ boost::shared_ptr<SipSessionPriv> mImplPriv;
+};
+
+void SipSession::addListener_async(
+ const AsteriskSCF::SessionCommunications::V1::AMD_Session_addListenerPtr& cb,
+ const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
+ const Ice::Current&)
+{
+ enqueueSessionWork(new AddListenerOperation(cb, listener, mImplPriv));
}
class IndicateOperation : public SuspendableWork
@@ -326,13 +416,17 @@ class GetEndpointOperation : public SuspendableWork
public:
GetEndpointOperation(
const AsteriskSCF::SessionCommunications::V1::AMD_Session_getEndpointPtr& cb,
- const boost::shared_ptr<SipSessionPriv> sessionPriv)
+ const boost::shared_ptr<SipSessionPriv>& sessionPriv)
: mCb(cb), mImplPriv(sessionPriv) { }
SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
{
mCb->ice_response(mImplPriv->mEndpoint->getEndpointProxy());
+ return Complete;
}
+
+ AsteriskSCF::SessionCommunications::V1::AMD_Session_getEndpointPtr mCb;
+ boost::shared_ptr<SipSessionPriv> mImplPriv;
};
/**
@@ -348,53 +442,22 @@ void SipSession::getEndpoint_async(
class GetInfoOperation : public SuspendableWork
{
public:
- //XXX This will likely need to be modified to take more parameters once we're
- //actually returning actual info.
GetInfoOperation(
- const AsteriskSCF::SessionCommunications::V1::AMD_Session_getInfoPtr& cb)
- : mCb(cb) { }
+ const AsteriskSCF::SessionCommunications::V1::AMD_Session_getInfoPtr& cb,
+ const boost::shared_ptr<SipSessionPriv>& sessionPriv)
+ : mCb(cb), mImplPriv(sessionPriv) { }
SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
{
- AsteriskSCF::SessionCommunications::V1::SessionInfoPtr sessioninfo =
- new AsteriskSCF::SessionCommunications::V1::SessionInfo();
-
- /* TODO: Utilize locking so this becomes safe
- if (!mImplPriv->mInviteSession || mImplPriv->mInviteSession->state == PJSIP_INV_STATE_NULL)
- {
- sessioninfo->currentState = "ready";
- }
- else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_CALLING)
- {
- sessioninfo->currentState = "outbound";
- }
- else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_INCOMING)
- {
- sessioninfo->currentState = "inbound";
- }
- else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_EARLY)
- {
- sessioninfo->currentState = "early";
- }
- else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_CONNECTING)
- {
- sessioninfo->currentState = "connecting";
- }
- else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_CONFIRMED)
- {
- sessioninfo->currentState = "connected";
- }
- else if (mImplPriv->mInviteSession->state == PJSIP_INV_STATE_DISCONNECTED)
- {
- sessioninfo->currentState = "disconnected";
- }
- */
+ AsteriskSCF::SessionCommunications::V1::SessionInfoPtr sessionInfo =
+ mImplPriv->getInfo();
mCb->ice_response(sessionInfo);
return Complete;
}
AsteriskSCF::SessionCommunications::V1::AMD_Session_getInfoPtr mCb;
+ boost::shared_ptr<SipSessionPriv> mImplPriv;
};
/**
@@ -404,13 +467,25 @@ void SipSession::getInfo_async(
const AsteriskSCF::SessionCommunications::V1::AMD_Session_getInfoPtr& cb,
const Ice::Current&)
{
- enqueueSessionWork(new GetInfoOperation(cb));
+ enqueueSessionWork(new GetInfoOperation(cb, mImplPriv));
}
class GetMediaSessionOperation : public SuspendableWork
{
public:
- //XXX PICK UP HERE TUESDAY
+ GetMediaSessionOperation(
+ const AsteriskSCF::SessionCommunications::V1::AMD_Session_getMediaSessionPtr& cb,
+ const boost::shared_ptr<SipSessionPriv>& sessionPriv)
+ : mCb(cb), mImplPriv(sessionPriv) { }
+
+ SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
+ {
+ mCb->ice_response(mImplPriv->mMediaSessionProxy);
+ return Complete;
+ }
+
+ AsteriskSCF::SessionCommunications::V1::AMD_Session_getMediaSessionPtr mCb;
+ boost::shared_ptr<SipSessionPriv> mImplPriv;
};
/**
@@ -420,28 +495,33 @@ void SipSession::getMediaSession_async(
const AsteriskSCF::SessionCommunications::V1::AMD_Session_getMediaSessionPtr& cb,
const Ice::Current&)
{
- return mImplPriv->mMediaSessionProxy;
+ enqueueSessionWork(new GetMediaSessionOperation(cb, mImplPriv));
}
class GetBridgeOperation : public SuspendableWork
{
- GetBridgeOperation(const AsteriskSCF::SessionCommunications::V1::AMD_Session_getBridgedPtr& cb,
+public:
+ GetBridgeOperation(const AsteriskSCF::SessionCommunications::V1::AMD_Session_getBridgePtr& cb,
const boost::shared_ptr<SipSessionPriv>& sessionPriv)
: mCb(cb), mImplPriv(sessionPriv) { }
SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
{
- if (mImplPriv->mBridge == 0)
+ AsteriskSCF::SessionCommunications::V1::BridgePrx bridge;
+ try
{
- mCb->ice_exception(AsteriskSCF::SessionCommunications::V1::NotBridged());
+ bridge = mImplPriv->getBridge();
+ }
+ catch (const AsteriskSCF::SessionCommunications::V1::NotBridged& ex)
+ {
+ mCb->ice_exception(ex);
return Complete;
}
-
- mCb->ice_response(mImplPriv->mBridge);
+ mCb->ice_response(bridge);
return Complete;
}
- AsteriskSCF::SessionCommunications::V1::AMD_Session_getBridgedPtr mCb;
+ AsteriskSCF::SessionCommunications::V1::AMD_Session_getBridgePtr mCb;
boost::shared_ptr<SipSessionPriv> mImplPriv;
};
@@ -461,21 +541,23 @@ public:
SetBridgeOperation(
const AsteriskSCF::SessionCommunications::V1::AMD_Session_setBridgePtr& cb,
const SipSessionPtr& session,
+ const boost::shared_ptr<SipSessionPriv>& sessionPriv,
const AsteriskSCF::SessionCommunications::V1::BridgePrx& bridge,
const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
const Ice::Current& current)
- : mAMDCb(cb), mSession(session), mBridge(bridge), mListener(listener), mCurrent(current) { }
+ : mCb(cb), mSession(session), mImplPriv(sessionPriv), mBridge(bridge), mListener(listener), mCurrent(current) { }
SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
{
mSession->setBridge(mBridge);
- mSession->addListener(mListener, mCurrent);
- mAMDCb->ice_response(mSession->getInfo(mCurrent));
+ mImplPriv->addListener(mListener);
+ mCb->ice_response(mImplPriv->getInfo());
return Complete;
}
- AsteriskSCF::SessionCommunications::V1::AMD_Session_setBridgePtr mAMDCb;
+ AsteriskSCF::SessionCommunications::V1::AMD_Session_setBridgePtr mCb;
SipSessionPtr mSession;
+ boost::shared_ptr<SipSessionPriv> mImplPriv;
AsteriskSCF::SessionCommunications::V1::BridgePrx mBridge;
AsteriskSCF::SessionCommunications::V1::SessionListenerPrx mListener;
Ice::Current mCurrent;
@@ -490,7 +572,7 @@ void SipSession::setBridge_async(
const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
const Ice::Current& current)
{
- enqueueSessionWork(new SetBridgeOperation(cb, this, bridge, listener, current));
+ enqueueSessionWork(new SetBridgeOperation(cb, this, mImplPriv, bridge, listener, current));
}
/**
@@ -502,15 +584,21 @@ void SipSession::setBridge(const AsteriskSCF::SessionCommunications::V1::BridgeP
mImplPriv->mBridge = bridge;
}
+AsteriskSCF::SessionCommunications::V1::BridgePrx SipSession::getBridge()
+{
+ return mImplPriv->getBridge();
+}
+
class RemoveBridgeOperation : public SuspendableWork
{
public:
RemoveBridgeOperation(
const AsteriskSCF::SessionCommunications::V1::AMD_Session_removeBridgePtr& cb,
+ const SipSessionPtr& session,
const boost::shared_ptr<SipSessionPriv>& sessionPriv,
const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
const Ice::Current& current)
- : mCb(cb), mImplPriv(sessionPriv), mListener(listener), mCurrent(current) { }
+ : mCb(cb), mSession(session), mImplPriv(sessionPriv), mListener(listener), mCurrent(current) { }
SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
{
@@ -520,16 +608,18 @@ public:
return Complete;
}
- setBridge(0);
+ mSession->setBridge(0);
- //XXX REMEMBER TO MAKE A LOCAL VERSION OF removeListener!
- removeListener(listener, current);
+ mImplPriv->removeListener(mListener);
+ mCb->ice_response();
+ return Complete;
}
- AsteriskSCF::SessionCommunications::V1::AMD_Session_removeBridgePtr cb;
- boost::shared_ptr<SipSessionPriv> sessionPriv;
- AsteriskSCF::SessionCommunications::V1::SessionListenerPrx listener;
- Ice::Current current;
+ AsteriskSCF::SessionCommunications::V1::AMD_Session_removeBridgePtr mCb;
+ SipSessionPtr mSession;
+ boost::shared_ptr<SipSessionPriv> mImplPriv;
+ AsteriskSCF::SessionCommunications::V1::SessionListenerPrx mListener;
+ Ice::Current mCurrent;
};
/**
@@ -540,26 +630,32 @@ void SipSession::removeBridge_async(
const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
const Ice::Current& current)
{
- enqueueSessionWork(new RemoveBridgeOperation(cb, mImplPriv, listener, current));
+ enqueueSessionWork(new RemoveBridgeOperation(cb, this, mImplPriv, listener, current));
}
+class RemoveListenerOperation : public SuspendableWork
+{
+public:
+ RemoveListenerOperation(
+ const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
+ const boost::shared_ptr<SipSessionPriv>& sessionPriv)
+ : mListener(listener), mImplPriv(sessionPriv) { }
+
+ SuspendableWorkResult execute(const SuspendableWorkListenerPtr&)
+ {
+ mImplPriv->removeListener(mListener);
+ return Complete;
+ }
+ AsteriskSCF::SessionCommunications::V1::SessionListenerPrx mListener;
+ boost::shared_ptr<SipSessionPriv> mImplPriv;
+};
+
/**
* An implementation of the removeListener method as defined in SessionCommunications.ice
*/
void SipSession::removeListener(const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener, const Ice::Current&)
{
- boost::unique_lock<boost::shared_mutex> lock(mImplPriv->mLock);
- mImplPriv->mListeners.erase(std::remove(mImplPriv->mListeners.begin(), mImplPriv->mListeners.end(),
- listener), mImplPriv->mListeners.end());
- if (mImplPriv->mInviteSession)
- {
- PJSipSessionModInfo *session_mod_info =
- static_cast<PJSipSessionModInfo*>(
- mImplPriv->mInviteSession->mod_data[mImplPriv->mManager->getSessionModule()->getModule().id]);
- lock.unlock();
- session_mod_info->updateSessionState(mImplPriv->mInviteSession);
- mImplPriv->mManager->getSessionModule()->replicateState(NULL, NULL, session_mod_info);
- }
+ enqueueSessionWork(new RemoveListenerOperation(listener, mImplPriv));
}
class StartOperation : public SuspendableWork
diff --git a/src/SipSession.h b/src/SipSession.h
index 8d608ef..7cd5a76 100644
--- a/src/SipSession.h
+++ b/src/SipSession.h
@@ -137,9 +137,11 @@ public:
const Ice::Current&);
void getBridge_async(
- const AsteriskSCF::SessionCommunications::V1:AMD_Session_getBridgePtr& cb,
+ const AsteriskSCF::SessionCommunications::V1::AMD_Session_getBridgePtr& cb,
const Ice::Current&);
+ AsteriskSCF::SessionCommunications::V1::BridgePrx getBridge();
+
void setBridge_async(
const AsteriskSCF::SessionCommunications::V1::AMD_Session_setBridgePtr& cb,
const AsteriskSCF::SessionCommunications::V1::BridgePrx&,
@@ -148,7 +150,7 @@ public:
void setBridge(const AsteriskSCF::SessionCommunications::V1::BridgePrx&);
void removeBridge_async(
- const AsteriskSCF::SessionCommunications::V1::Session_removeBridgePtr& cb,
+ const AsteriskSCF::SessionCommunications::V1::AMD_Session_removeBridgePtr& cb,
const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx&,
const Ice::Current&);
@@ -199,6 +201,8 @@ public:
SessionWorkPtr getSessionWork();
void setSessionWork(const SessionWorkPtr& sessionWork);
+
+ void enqueueSessionWork(const AsteriskSCF::System::WorkQueue::V1::SuspendableWorkPtr&);
private:
void requestRTPSessions(AsteriskSCF::Media::V1::FormatSeq& formats);
-----------------------------------------------------------------------
--
asterisk-scf/integration/sip.git
More information about the asterisk-scf-commits
mailing list