[asterisk-scf-commits] asterisk-scf/integration/bridging.git branch "media" created.
Commits to the Asterisk SCF project code repositories
asterisk-scf-commits at lists.digium.com
Mon Jun 13 15:40:35 CDT 2011
branch "media" has been created
at 687b55e261727b3a3fb910e20d4ad0715b7b42f2 (commit)
- Log -----------------------------------------------------------------
commit 687b55e261727b3a3fb910e20d4ad0715b7b42f2
Author: Joshua Colp <jcolp at digium.com>
Date: Sun Jun 12 17:39:27 2011 -0300
Partially update to new media slice.
diff --git a/src/MediaSplicer.cpp b/src/MediaSplicer.cpp
index 1a631b6..9d340a2 100755
--- a/src/MediaSplicer.cpp
+++ b/src/MediaSplicer.cpp
@@ -254,7 +254,8 @@ public:
{
try
{
- (*i)->setSink(StreamSinkPrx());
+// (*i)->setSink(StreamSinkPrx());
+// TODO: call removeSink
}
catch (const Ice::Exception& ex)
{
@@ -324,7 +325,8 @@ public:
{
try
{
- (*i)->setSink(StreamSinkPrx());
+// (*i)->setSink(StreamSinkPrx());
+// TODO: Call removeSink
}
catch (const Ice::Exception& ex)
{
@@ -790,7 +792,7 @@ protected:
{
}
- void setSinkCB()
+ void addSinkCB()
{
bool other = false;
{
@@ -863,7 +865,7 @@ protected:
for (OutgoingPairings::iterator i = mMaterials->outgoingPairings.begin(); i != mMaterials->outgoingPairings.end(); ++i)
{
OutgoingHelperPtr h(new OutgoingHelper(this, index));
- i->second->begin_setSink(i->first, AsteriskSCF::Media::V1::newCallback_StreamSource_setSink(h, &OutgoingHelper::setSinkCB, &OutgoingHelper::failed));
+ i->second->begin_addSink(i->first, AsteriskSCF::Media::V1::newCallback_StreamSource_addSink(h, &OutgoingHelper::addSinkCB, &OutgoingHelper::failed));
i->first->begin_setSource(i->second, AsteriskSCF::Media::V1::newCallback_StreamSink_setSource(h, &OutgoingHelper::setSourceCB, &OutgoingHelper::failed));
++index;
}
@@ -871,7 +873,7 @@ protected:
for (IncomingPairings::iterator i = mMaterials->incomingPairings.begin(); i != mMaterials->incomingPairings.end(); ++i)
{
IncomingHelperPtr h(new IncomingHelper(this, index));
- i->first->begin_setSink(i->second, AsteriskSCF::Media::V1::newCallback_StreamSource_setSink(h, &IncomingHelper::setSinkCB, &IncomingHelper::failed));
+ i->first->begin_addSink(i->second, AsteriskSCF::Media::V1::newCallback_StreamSource_addSink(h, &IncomingHelper::addSinkCB, &IncomingHelper::failed));
i->second->begin_setSource(i->first, AsteriskSCF::Media::V1::newCallback_StreamSink_setSource(h, &IncomingHelper::setSourceCB, &IncomingHelper::failed));
++index;
}
commit 6cffd17652efd700c25edd84d66659d69681b78f
Author: Brent Eagles <beagles at digium.com>
Date: Wed Jun 8 09:34:11 2011 -0230
Have only the standby component register with the locator.
diff --git a/src/Service.cpp b/src/Service.cpp
index fd3fcd6..addafb4 100644
--- a/src/Service.cpp
+++ b/src/Service.cpp
@@ -63,6 +63,10 @@ public:
{
(*i)->activated(ReplicaPrx::uncheckedCast(mAdapter->createDirectProxy(mAdapter->getCommunicator()->stringToIdentity(mReplicaId))));
}
+ //
+ // A nice thing to do here would be to refresh the proxy that is registered with the service locator. It may not
+ // have really changed (depends on the failover mechanism used) but in the event that it did...
+ //
return true;
}
@@ -274,23 +278,26 @@ void BridgingApp::start(const std::string& name, const Ice::CommunicatorPtr& com
manager->activate();
}
- bool registered = false;
- try
- {
- ServiceLocatorParamsPtr parameters(new ServiceLocatorParams);
- parameters->category = BridgeServiceDiscoveryCategory;
- mLocator =
- new LocatorRegistrationWrapper<BridgeManagerPrx>(communicator, serviceLocatorManagementProperty, bridgeManagerPrx,
- adapterName, parameters);
- registered = mLocator->registerService();
- }
- catch (const Ice::Exception&)
- {
- }
- if (!registered)
+ if (!onStandby)
{
- mRegisterThread = new RegisterThread<BridgeManagerPrx>(mLocator);
- mRegisterThread->start();
+ bool registered = false;
+ try
+ {
+ ServiceLocatorParamsPtr parameters(new ServiceLocatorParams);
+ parameters->category = BridgeServiceDiscoveryCategory;
+ mLocator =
+ new LocatorRegistrationWrapper<BridgeManagerPrx>(communicator, serviceLocatorManagementProperty, bridgeManagerPrx,
+ adapterName, parameters);
+ registered = mLocator->registerService();
+ }
+ catch (const Ice::Exception&)
+ {
+ }
+ if (!registered)
+ {
+ mRegisterThread = new RegisterThread<BridgeManagerPrx>(mLocator);
+ mRegisterThread->start();
+ }
}
//
// TODO: We need to know whether or not to activate!
commit 6595950f2059c14dc1dbc40f3ba76d4a27d72fd5
Author: Brent Eagles <beagles at digium.com>
Date: Mon Jun 6 12:17:40 2011 -0230
Simplify disconnection of the media splicing. To avoid race conditions, a
disconnection should not affect any of the other media sessions. The other
media sessions will be progressively altered either by internal bridge
operations or by the owning session itself.
Modified the listener to included a proactive reap to clean out disconnected
sessions.
diff --git a/src/MediaSplicer.cpp b/src/MediaSplicer.cpp
index 7812b64..1a631b6 100755
--- a/src/MediaSplicer.cpp
+++ b/src/MediaSplicer.cpp
@@ -133,36 +133,7 @@ public:
void unplug()
{
mLogger(Debug) << FUNLOG << ": called.";
- bool hadMedia = disconnectMedia();
- if (!hadMedia)
- {
- mLogger(Debug) << FUNLOG << ": we did not have any media. Contacting the peer connector and telling it to disconnect!";
- MediaConnectorPtr peer;
- {
- boost::unique_lock<boost::shared_mutex> lock(mLock);
- peer = mPeer;
- mPeer = 0;
- }
- if (peer)
- {
- peer->disconnectMedia();
- }
- }
- else
- {
- mLogger(Debug) << FUNLOG << ": media connections unplugged. Let's forget about our peer connector now!";
- MediaConnectorPtr peer;
- {
- boost::unique_lock<boost::shared_mutex> lock(mLock);
- peer = mPeer;
- mPeer = 0;
- }
- if (peer)
- {
- peer->clearConnections();
- }
- mPeer = 0;
- }
+ disconnectMedia();
mLogger(Debug) << FUNLOG << ": finished unplugging.";
if (mReplicator)
{
@@ -258,71 +229,54 @@ public:
//
void destroy()
{
- vector<OutgoingPairing> outgoing;
- vector<IncomingPairing> incoming;
SessionPairingPtr newState;
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
- outgoing = mOutgoing;
mOutgoing.clear();
- incoming = mIncoming;
mIncoming.clear();
mConnected = false;
newState = createUpdate();
}
pushUpdate(newState);
- if (outgoing.size() == 0 && incoming.size() == 0)
- {
- return;
- }
mLogger(Debug) << FUNLOG << ": unplugging sinks and sources";
//
- // Disconnect everybody, eating up exceptions in case things have gone away. This is a perfect spot
- // for oneways. We don't care about errors when destroying. We still catch Ice::Exceptions though since
- // a connection might be attempted, even on behalf of a oneway.
+ // NOTE: we used to talk to the peer and disconnect it too, but really
+ // that is fraught with danger when using non-synchronous or non-serial
+ // RPCs. We take the safe road and allow each connector to simply tell
+ // it's media session constituents to stop communicating with anything
+ // else. This also used to try one-ways, that's probably a bad idea too.
//
- for (vector<OutgoingPairing>::iterator i = outgoing.begin(); i != outgoing.end(); ++i)
+ try
{
- mLogger(Debug) << FUNLOG << ": disconnecting " << i->second->ice_toString() << " and "
- << i->first->ice_toString();
-
- try
+ StreamSourceSeq sources = mMedia->getSources();
+ for (StreamSourceSeq::const_iterator i = sources.begin(); i != sources.end(); ++i)
{
- tryOneWay(i->second)->setSink(0);
- }
- catch (const Ice::Exception& ex)
- {
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
- }
- try
- {
- tryOneWay(i->first)->setSource(0);
+ try
+ {
+ (*i)->setSink(StreamSinkPrx());
+ }
+ catch (const Ice::Exception& ex)
+ {
+ mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
+ }
}
- catch (const Ice::Exception& ex)
+ StreamSinkSeq sinks = mMedia->getSinks();
+ for (StreamSinkSeq::const_iterator i = sinks.begin(); i != sinks.end(); ++i)
{
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
+ try
+ {
+ (*i)->setSource(StreamSourcePrx());
+ }
+ catch (const Ice::Exception& ex)
+ {
+ mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
+ }
}
}
- for (vector<IncomingPairing>::iterator i = incoming.begin(); i != incoming.end(); ++i)
+ catch (const Ice::Exception& ex)
{
- mLogger(Debug) << FUNLOG << ": disconnecting " << i->first->ice_toString() << " and " << i->second->ice_toString();
- try
- {
- tryOneWay(i->first)->setSink(0);
- }
- catch (const Ice::Exception& ex)
- {
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
- }
- try
- {
- tryOneWay(i->second)->setSource(0);
- }
- catch (const Ice::Exception& ex)
- {
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
- }
+ mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
}
}
@@ -345,70 +299,56 @@ public:
bool disconnectMedia()
{
mLogger(Debug) << FUNLOG << ": unplugging sinks and sources.";
- vector<OutgoingPairing> outgoing;
- vector<IncomingPairing> incoming;
SessionPairingPtr newState;
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
- outgoing = mOutgoing;
mOutgoing.clear();
- incoming = mIncoming;
mIncoming.clear();
mConnected = false;
newState = createUpdate();
}
pushUpdate(newState);
- if (outgoing.size() == 0 && incoming.size() == 0)
- {
- return false;
- }
mLogger(Debug) << FUNLOG << ": unplugging sinks and sources";
//
- // TODO: Disconnect everybody, eating up exceptions in case things have gone away. This is a perfect spot for
- // oneways or, at the very least, AMI.
+ // NOTE: we used to talk to the peer and disconnect it too, but really
+ // that is fraught with danger when using non-synchronous or non-serial
+ // RPCs. We take the safe road and allow each connector to simply tell
+ // it's media session constituents to stop communicating with anything
+ // else. This also used to try one-ways, that's probably a bad idea too.
//
- for (vector<OutgoingPairing>::iterator i = outgoing.begin(); i != outgoing.end(); ++i)
+ try
{
- mLogger(Debug) << FUNLOG << ": disconnecting " << i->second->ice_toString() << " and " << i->first->ice_toString();
-
- try
- {
- i->second->setSink(0);
- }
- catch (const Ice::Exception& ex)
- {
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
- }
- try
+ StreamSourceSeq sources = mMedia->getSources();
+ for (StreamSourceSeq::const_iterator i = sources.begin(); i != sources.end(); ++i)
{
- i->first->setSource(0);
+ try
+ {
+ (*i)->setSink(StreamSinkPrx());
+ }
+ catch (const Ice::Exception& ex)
+ {
+ mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
+ }
}
- catch (const Ice::Exception& ex)
+ StreamSinkSeq sinks = mMedia->getSinks();
+ for (StreamSinkSeq::const_iterator i = sinks.begin(); i != sinks.end(); ++i)
{
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
+ try
+ {
+ (*i)->setSource(StreamSourcePrx());
+ }
+ catch (const Ice::Exception& ex)
+ {
+ mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
+ }
}
}
- for (vector<IncomingPairing>::iterator i = incoming.begin(); i != incoming.end(); ++i)
+ catch (const Ice::Exception& ex)
{
- mLogger(Debug) << FUNLOG << ": disconnecting " << i->first->ice_toString() << " and " << i->second->ice_toString();
- try
- {
- i->first->setSink(0);
- }
- catch (const Ice::Exception& ex)
- {
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
- }
- try
- {
- i->second->setSource(0);
- }
- catch (const Ice::Exception& ex)
- {
- mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
- }
+ mLogger(Debug) << FUNLOG << ":" << __LINE__ << ": exception, thought you would like to know " << ex.what();
}
+
return true;
}
diff --git a/src/SessionListener.cpp b/src/SessionListener.cpp
index 71c6d4d..bf5b450 100644
--- a/src/SessionListener.cpp
+++ b/src/SessionListener.cpp
@@ -183,6 +183,16 @@ public:
// A null response code pointer will prevent the stop indication from being sent back to this session.
//
session->shutdown(mListenerPrx, ResponseCodePtr());
+ mSessions->removeSession(session->getBridgedSession());
+
+ //
+ // It's a good idea to reap at this time to avoid having processes
+ // that might be disconnected from being considered in the
+ // visitSessions() run. TODO: Some operations shouldn't be
+ // performed on sessions in certain states. It's really the
+ // operation's responsibility to prevent that from happening.
+ //
+ mSessions->reap();
ShutdownSessionOperation shutdownOp(session->getSession(), mListenerPrx, stopped->response, mLogger);
mSessions->visitSessions(shutdownOp);
@@ -195,7 +205,6 @@ public:
{
try
{
- mSessions->reap();
SessionSeq currentSessions = mSessions->getSessionSeq();
//
commit b1411d90beaf59d710f87311c0471b57c29ab71f
Author: Brent Eagles <beagles at digium.com>
Date: Fri Jun 3 13:38:55 2011 -0230
Fixing build error
diff --git a/src/SessionCollection.h b/src/SessionCollection.h
index da7100c..745c7df 100644
--- a/src/SessionCollection.h
+++ b/src/SessionCollection.h
@@ -106,7 +106,7 @@ public:
}
catch (const std::exception& ex)
{
- mLogger(Error) << "visitSessions caught " << ex.what();
+ mLogger(AsteriskSCF::System::Logging::Error) << "visitSessions caught " << ex.what();
if (!ignoreExceptions)
{
throw;
commit 2e92a5419e33830e5896a7121eacb79d8450b28a
Author: Brent Eagles <beagles at digium.com>
Date: Fri Jun 3 13:33:57 2011 -0230
Fixes to address an issue where indications were being sent to sessions that
the bridge should have already disregarded.
diff --git a/src/BridgeImpl.cpp b/src/BridgeImpl.cpp
index fcfab3b..60b6bb2 100755
--- a/src/BridgeImpl.cpp
+++ b/src/BridgeImpl.cpp
@@ -784,6 +784,13 @@ void BridgeImpl::replaceSession_async(const AMD_Bridge_replaceSessionPtr& callba
SessionsTrackerPtr removeTracker(new SessionsTracker);
removeTracker->add(session->getSession());
session->shutdown(mSessionListenerPrx, new ResponseCode);
+
+ //
+ // Since the shutdown of the session is inherently asynchronous, the
+ // rest of this is going to happen anyways, so lets remove it from the
+ // list of known sessions now so it doesn't get in the way.
+ //
+ mSessions->removeSession(session->getBridgedSession());
SessionsTrackerPtr tracker(new SessionsTracker);
QueuedTasks tasks;
diff --git a/src/SessionCollection.h b/src/SessionCollection.h
index d961f87..da7100c 100644
--- a/src/SessionCollection.h
+++ b/src/SessionCollection.h
@@ -95,12 +95,23 @@ public:
*
**/
template <typename Func>
- Func& visitSessions(Func& op)
+ Func& visitSessions(Func& op, bool ignoreExceptions = true)
{
boost::shared_lock<boost::shared_mutex> lock(mLock);
for (SessionMap::iterator i = mMap.begin(); i != mMap.end(); ++i)
{
- op(i->second);
+ try
+ {
+ op(i->second);
+ }
+ catch (const std::exception& ex)
+ {
+ mLogger(Error) << "visitSessions caught " << ex.what();
+ if (!ignoreExceptions)
+ {
+ throw;
+ }
+ }
}
return op;
}
diff --git a/src/SessionListener.cpp b/src/SessionListener.cpp
index 0521181..71c6d4d 100644
--- a/src/SessionListener.cpp
+++ b/src/SessionListener.cpp
@@ -219,6 +219,7 @@ public:
//
// TODO: Who gets the ring notifications will likely depend on configuration, etc.
//
+ mSessions->reap();
RelayIndication relayer(source, mLogger, commandIndication, false);
mSessions->visitSessions(relayer);
}
commit 29d4f2f06467b3b03b09b79588d37ab7c5b452ca
Author: Brent Eagles <beagles at digium.com>
Date: Fri Jun 3 11:41:26 2011 -0230
Resolve a silly media connection issue where audio would not be connected properly.
diff --git a/src/MediaSplicer.cpp b/src/MediaSplicer.cpp
index 4bdcdae..7812b64 100755
--- a/src/MediaSplicer.cpp
+++ b/src/MediaSplicer.cpp
@@ -31,6 +31,10 @@
// * Cleanup of MediaConnectorI.
// * Conference bridging.
//
+// NOTES: I suspect there is a lot of unnecessary work being done here if some reasonable distributed object
+// protocols are followed throughout. The incoming and outgoing pairings and the disconnect/reconnect
+// logic is also likely fraught with race conditions. I will be looking at this more detail IMMEDIATELY.
+//
using namespace AsteriskSCF::System::Logging;
using namespace AsteriskSCF::Media::V1;
using namespace AsteriskSCF::Bridge::V1;
@@ -452,6 +456,10 @@ public:
MediaConnectorPtr createConnector(const SessionWrapperPtr& session, const MediaConnectorBuilderPtr& data)
{
MediaConnectorIPtr connector(new MediaConnectorI(data, mBridgeId, session->id(), mReplicator, mLogger));
+
+ //
+ // TODO: check to make sure this isn't mirroring!
+ //
if (data->peer)
{
data->peer->update(connector, data->outgoingPairings, data->incomingPairings);
@@ -533,7 +541,7 @@ public:
return result;
}
- vector<OutgoingPairing> findCompatiblePairings(const StreamSinkSeq& sinks)
+ vector<OutgoingPairing> findCompatiblePairings(const AsteriskSCF::Media::V1::SessionPrx& sessionToBeConnected, const StreamSinkSeq& sinks)
{
vector<StreamSourceSeq> allSources;
@@ -545,6 +553,10 @@ public:
//
for (MediaSessions::iterator i = mSessions.begin(); i != mSessions.end(); ++i)
{
+ if (i->mediaSession->ice_getIdentity() == sessionToBeConnected->ice_getIdentity())
+ {
+ continue;
+ }
allSources.push_back(i->mediaSession->getSources());
}
}
@@ -578,7 +590,7 @@ public:
return result;
}
- vector<IncomingPairing> findCompatiblePairings(const StreamSourceSeq& sources)
+ vector<IncomingPairing> findCompatiblePairings(const AsteriskSCF::Media::V1::SessionPrx& sessionToBeConnected, const StreamSourceSeq& sources)
{
vector<StreamSinkSeq> allSinks;
{
@@ -589,6 +601,10 @@ public:
//
for (MediaSessions::iterator i = mSessions.begin(); i != mSessions.end(); ++i)
{
+ if (i->mediaSession->ice_getIdentity() == sessionToBeConnected->ice_getIdentity())
+ {
+ continue;
+ }
allSinks.push_back(i->mediaSession->getSinks());
}
}
@@ -791,8 +807,8 @@ public:
protected:
bool executeImpl()
{
- mMaterials->incomingPairings = mSplicer->findCompatiblePairings(mMaterials->sources);
- mMaterials->outgoingPairings = mSplicer->findCompatiblePairings(mMaterials->sinks);
+ mMaterials->incomingPairings = mSplicer->findCompatiblePairings(mMaterials->mediaSession, mMaterials->sources);
+ mMaterials->outgoingPairings = mSplicer->findCompatiblePairings(mMaterials->mediaSession, mMaterials->sinks);
return true;
}
commit bc789938cc49587975f97a60d7bcb1a681ec981c
Author: Brent Eagles <beagles at digium.com>
Date: Thu Jun 2 23:50:40 2011 -0230
Solution for attended transfers. Basically the removeBridge implementation was
doing things it shouldn't to the session being removed.
diff --git a/src/BridgeImpl.cpp b/src/BridgeImpl.cpp
index 29e4849..fcfab3b 100755
--- a/src/BridgeImpl.cpp
+++ b/src/BridgeImpl.cpp
@@ -591,8 +591,8 @@ void BridgeImpl::removeSessions_async(const AMD_Bridge_removeSessionsPtr& callba
SessionWrapperPtr session = mSessions->getSession(*i);
if (session)
{
- session->shutdown(mSessionListenerPrx, new ResponseCode);
removed->add(session->getSession());
+ mSessions->removeSession(session->getBridgedSession());
}
}
QueuedTasks tasks;
commit a6d7d3da25de0c72ca58fd32b389de1810e4eedf
Author: Brent Eagles <beagles at digium.com>
Date: Thu Jun 2 21:37:14 2011 -0230
Add some logging.
diff --git a/src/SessionWrapper.cpp b/src/SessionWrapper.cpp
index 2450ccd..48ccfa8 100644
--- a/src/SessionWrapper.cpp
+++ b/src/SessionWrapper.cpp
@@ -73,7 +73,6 @@ public:
//
// These exceptions indicate something is wrong with the request on this object. We should not expect that
// it could ever exceed. ObjectNotExistException is included in this catch.
- // TODO- it might be reasonable to disconnect this session for this!
//
mLogger(Warning) << "exception when calling connect() on " << mSession->id() << ": " << x.what();
mSession->destroy();
@@ -181,10 +180,11 @@ class SessionStopTask : public QueuedTask
{
public:
SessionStopTask(const SessionWrapperPtr& session,
- const ResponseCodePtr& code) :
+ const ResponseCodePtr& code, const Logger& logger) :
QueuedTask("SessionStopTask"),
mSession(session),
- mCode(code)
+ mCode(code),
+ mLogger(logger)
{
}
@@ -202,12 +202,26 @@ protected:
mListener->succeeded();
}
- void failed(const Ice::Exception&)
+ void failed(const Ice::Exception& ex)
{
- //
- // TODO: Log exception.
- //
- mListener->failed();
+ try
+ {
+ ex.ice_throw();
+ }
+ catch (const Ice::ObjectNotExistException& x)
+ {
+ mLogger(Warning) << "received an ONE for " << mSession->id() << " when calling stop(), continuing...";
+ mListener->succeeded();
+ }
+ catch (const std::exception& x)
+ {
+ mLogger(Error) << "exception when calling connect() on " << mSession->id() << " : " << x.what();
+ mListener->failed();
+ }
+ catch (...)
+ {
+ mListener->failed();
+ }
}
void failImpl()
@@ -227,6 +241,7 @@ protected:
private:
SessionWrapperPtr mSession;
ResponseCodePtr mCode;
+ Logger mLogger;
};
class SetStateTask : public QueuedTask
@@ -337,7 +352,7 @@ private:
};
QueuedTasks createShutdownTasks(const SessionWrapperPtr& session, const SessionListenerPrx& listener,
- const ResponseCodePtr& code)
+ const ResponseCodePtr& code, const Logger& logger)
{
//
// We use this task instead of changing states because it will prevent the
@@ -353,7 +368,7 @@ QueuedTasks createShutdownTasks(const SessionWrapperPtr& session, const SessionL
{
tasks.push_back(new RemoveBridgeTask(session, listener, true));
tasks.push_back(new ShutdownMediaTask(session));
- tasks.push_back(new SessionStopTask(session, code));
+ tasks.push_back(new SessionStopTask(session, code, logger));
}
return tasks;
}
@@ -560,7 +575,7 @@ void SessionWrapper::setup()
void SessionWrapper::shutdown(const SessionListenerPrx& listener, const ResponseCodePtr& code)
{
mLogger(Debug) << FUNLOG << ": beginning shutdown of " << mId;
- QueuedTaskPtr shutdownRunner(new QueueableExecutor(createShutdownTasks(this, listener, code), mLogger));
+ QueuedTaskPtr shutdownRunner(new QueueableExecutor(createShutdownTasks(this, listener, code, mLogger), mLogger));
if (!code)
{
//
commit 7a581b9c43a91428b9f937509b06ea7cdc9aa31a
Author: Kevin P. Fleming <kpfleming at digium.com>
Date: Wed Jun 1 16:56:54 2011 -0500
More CMake script cleanup, and ensure bridge_unit_tests gets run.
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index f12a0f0..f2c83c3 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -1,9 +1,6 @@
-if(NOT logger_dir)
- message(FATAL_ERROR "The logger directory could not be found ${logger_dir}")
-endif()
-include_directories(${logger_dir}/include)
include_directories(${API_INCLUDE_DIR})
-include_directories(${utils_dir}/include)
+include_directories(${logger_dir}/include)
+include_directories(${ice-util-cpp_dir}/include)
asterisk_scf_slice_include_directories(${API_SLICE_DIR})
@@ -44,7 +41,6 @@ target_link_libraries(bridgeservice logging-client)
asterisk_scf_component_install(bridgeservice)
asterisk_scf_component_init(BridgeReplicator)
-include_directories(${utils_dir}/StateReplicator/include)
asterisk_scf_component_add_file(BridgeReplicator BridgeReplicatorService.cpp)
asterisk_scf_component_add_slice(BridgeReplicator ./BridgeReplicatorIf.ice)
asterisk_scf_component_add_ice_libraries(BridgeReplicator IceStorm)
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index b577e6e..b06bf75 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -1,17 +1,12 @@
+include_directories(${API_INCLUDE_DIR})
+include_directories(${logger_dir}/include)
+include_directories(${ice-util-cpp_dir}/include)
+
asterisk_scf_slice_include_directories(${API_SLICE_DIR})
-if(NOT integrated_build STREQUAL "true")
- add_subdirectory(channel_driver/local_slice)
-endif()
+
include_directories("../src")
-include_directories(${utils_dir}/include)
-if(NOT logger_dir)
- message(FATAL_ERROR "The logger directory could not be found ${logger_dir}")
-endif()
-include_directories(${logger_dir}/include)
-include_directories(${API_INCLUDE_DIR})
-if(integrated_build STREQUAL "true")
- include_directories(${CMAKE_CURRENT_BINARY_DIR}/../../test_channel/local_slice/generated)
-endif()
+
+include_directories(${CMAKE_CURRENT_BINARY_DIR}/../../test_channel/local_slice/generated)
asterisk_scf_component_init(bridge_component_test)
asterisk_scf_component_add_slice(bridge_component_test CommandsIf)
@@ -30,11 +25,7 @@ asterisk_scf_component_build_icebox(bridge_component_test)
target_link_libraries(bridge_component_test logging-client)
target_link_libraries(bridge_component_test asterisk-scf-api)
target_link_libraries(bridge_component_test test-channel-api)
-
-# component tests only work for integrated builds
-if(integrated_build STREQUAL "true")
- asterisk_scf_test_icebox(bridge_component_test config/test_bridging.conf)
-endif()
+asterisk_scf_test_icebox(bridge_component_test config/test_bridging.conf)
asterisk_scf_component_init(bridge_unit_tests)
asterisk_scf_component_add_slice(bridge_unit_tests ../src/BridgeReplicatorIf.ice)
@@ -43,8 +34,9 @@ asterisk_scf_component_add_file(bridge_unit_tests ../src/SessionOperations.cpp)
asterisk_scf_component_add_file(bridge_unit_tests ../src/SessionWrapper.cpp)
asterisk_scf_component_add_file(bridge_unit_tests ../src/MediaSplicer.cpp)
asterisk_scf_component_add_file(bridge_unit_tests UnitTests.cpp)
-asterisk_scf_component_add_ice_libraries(bridge_component_test Ice)
+asterisk_scf_component_add_ice_libraries(bridge_unit_tests Ice)
asterisk_scf_component_add_boost_libraries(bridge_unit_tests unit_test_framework thread)
asterisk_scf_component_build_standalone(bridge_unit_tests)
target_link_libraries(bridge_unit_tests logging-client)
target_link_libraries(bridge_unit_tests asterisk-scf-api)
+asterisk_scf_test_boost(bridge_unit_tests)
commit 0584204733e9b75b4e93fbc505d525f3855c2ba9
Author: Brent Eagles <beagles at digium.com>
Date: Wed Jun 1 11:26:28 2011 -0230
Allow replacing of the media connector through the "fundamental media
connection" function without checks to prior state. For the most part this
should be fine. Where it would get sticky is if there was an upstream
action that was being repeated in error. Even that might not be noticable from
the caller's point of view but would result in a lot of unnecessary work and
RPCs.
diff --git a/src/SessionWrapper.cpp b/src/SessionWrapper.cpp
index 23ee5b9..2450ccd 100644
--- a/src/SessionWrapper.cpp
+++ b/src/SessionWrapper.cpp
@@ -456,10 +456,13 @@ void SessionWrapper::setupMedia()
{
mLogger(Debug) << FUNLOG << " for " << mId;
boost::unique_lock<boost::shared_mutex> lock(mLock);
- if (!mConnector)
- {
- mSplicer->connect(this, mSession->session);
- }
+ //
+ // TODO: we don't really want there to be any precondition checks in this
+ // basic method. If we need to prevent this from recurring, then it needs
+ // to be in the caller's logic, not here.
+ //
+ mConnector = 0;
+ mSplicer->connect(this, mSession->session);
}
void SessionWrapper::setConnector(const MediaConnectorPtr& connector)
commit 87dbea8306aae28375fd8e0676f79825ba260dce
Author: Brent Eagles <beagles at digium.com>
Date: Tue May 31 15:30:42 2011 -0230
Added a "continue on exception flag to RemoveBridgeTask" to allow following tasks to continue.
diff --git a/src/SessionWrapper.cpp b/src/SessionWrapper.cpp
index cf7404f..23ee5b9 100644
--- a/src/SessionWrapper.cpp
+++ b/src/SessionWrapper.cpp
@@ -105,10 +105,11 @@ private:
class RemoveBridgeTask : public QueuedTask
{
public:
- RemoveBridgeTask(const SessionWrapperPtr& session, const SessionListenerPrx& listener) :
+ RemoveBridgeTask(const SessionWrapperPtr& session, const SessionListenerPrx& listener, bool continueOnEx) :
QueuedTask("RemoveBridgeTask"),
mSession(session),
- mSessionListener(listener)
+ mSessionListener(listener),
+ mContinueOnException(continueOnEx)
{
}
@@ -138,13 +139,21 @@ protected:
// This is most likely ok, it just means removeBridge was called
// twice.
//
+ mListener->succeeded();
}
catch (...)
{
- //
- // TODO: Log exception.
- //
- mListener->failed();
+ if (mContinueOnException)
+ {
+ mListener->succeeded();
+ }
+ else
+ {
+ //
+ // TODO: Log exception.
+ //
+ mListener->failed();
+ }
}
}
@@ -165,6 +174,7 @@ protected:
private:
SessionWrapperPtr mSession;
SessionListenerPrx mSessionListener;
+ bool mContinueOnException;
};
class SessionStopTask : public QueuedTask
@@ -341,7 +351,7 @@ QueuedTasks createShutdownTasks(const SessionWrapperPtr& session, const SessionL
tasks.push_back(new SetStateFromTask(session, ::AsteriskSCF::Bridge::V1::Disconnected, statesToContinueOn));
if (code)
{
- tasks.push_back(new RemoveBridgeTask(session, listener));
+ tasks.push_back(new RemoveBridgeTask(session, listener, true));
tasks.push_back(new ShutdownMediaTask(session));
tasks.push_back(new SessionStopTask(session, code));
}
commit e52dc2dbaa0b0afd56ef90dae7d8371eac83e871
Author: Brent Eagles <beagles at digium.com>
Date: Tue May 31 10:03:55 2011 -0230
Some logging fixes and adding some missing reaping that was complicating shutdown.
diff --git a/src/BridgeImpl.cpp b/src/BridgeImpl.cpp
index 0cf2807..29e4849 100755
--- a/src/BridgeImpl.cpp
+++ b/src/BridgeImpl.cpp
@@ -527,6 +527,7 @@ void BridgeImpl::addSessions_async(const AMD_Bridge_addSessionsPtr& callback, co
{
try
{
+ mSessions->reap();
if (sessions.empty())
{
if (callback)
@@ -575,6 +576,7 @@ void BridgeImpl::removeSessions_async(const AMD_Bridge_removeSessionsPtr& callba
}
checkSessions(sessions);
statePreCheck();
+ mSessions->reap();
//
// The shutdown of individual sessions are implemented as series of AMI requests. Once initiated,
@@ -619,6 +621,7 @@ SessionSeq BridgeImpl::listSessions(const Ice::Current& current)
void BridgeImpl::shutdown(const Ice::Current& current)
{
+ mSessions->reap();
//
// In an effort to allow some consistency with replicas, the shutdown operation is broken into
// two parts. In an asynchronous version, this would probably be a couple of queued tasks.
@@ -763,7 +766,7 @@ void BridgeImpl::replaceSession_async(const AMD_Bridge_replaceSessionPtr& callba
try
{
mLogger(Debug) << FUNLOG << ":" << objectIdFromCurrent(current);
-
+ mSessions->reap();
checkSessions(newSessions);
statePreCheck();
diff --git a/src/BridgeServiceConfig.h b/src/BridgeServiceConfig.h
index 6f61e68..5452f5e 100644
--- a/src/BridgeServiceConfig.h
+++ b/src/BridgeServiceConfig.h
@@ -60,7 +60,7 @@ private:
};
#define FUNLOG \
- __FILE__ << ':' << __LINE__ << '(' << BOOST_CURRENT_FUNCTION << ')'
+ '(' << BOOST_CURRENT_FUNCTION << ')'
inline std::string objectIdFromCurrent(const Ice::Current& current)
{
diff --git a/src/SessionCollection.cpp b/src/SessionCollection.cpp
index fbe07bb..bf9db83 100644
--- a/src/SessionCollection.cpp
+++ b/src/SessionCollection.cpp
@@ -25,14 +25,6 @@ using namespace AsteriskSCF::Bridge::V1;
using namespace AsteriskSCF;
using namespace std;
-/**
- *
- * NOTE: The lack of reaping throughout is deliberate. Sessions that are in a terminal state are of some interest to the
- * service internals. Reaping is consequently a manual process. The visitSessions method + various SessionOperations can
- * be used to perform state-dependent queries.
- *
- **/
-
SessionCollection::SessionCollection(const Ice::CommunicatorPtr& comm, const string& bridgeId,
const ReplicatorSmartPrx& replicator,
const Logger& logger) :
diff --git a/src/SessionListener.cpp b/src/SessionListener.cpp
index 5fffe12..0521181 100644
--- a/src/SessionListener.cpp
+++ b/src/SessionListener.cpp
@@ -124,7 +124,7 @@ public:
{
string proxyString = source->ice_toString();
mLogger(Debug) << FUNLOG << ": session connected " << proxyString;
-
+ mSessions->reap();
try
{
SessionWrapperPtr session = mSessions->getSession(source);
@@ -195,6 +195,7 @@ public:
{
try
{
+ mSessions->reap();
SessionSeq currentSessions = mSessions->getSessionSeq();
//
diff --git a/src/Tasks.h b/src/Tasks.h
index 0b13188..c3ce557 100644
--- a/src/Tasks.h
+++ b/src/Tasks.h
@@ -235,7 +235,7 @@ public:
**/
void start()
{
- mLogger(AsteriskSCF::System::Logging::Debug) << mName << ": starting running";
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " starting running.";
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
if (!mStopped)
@@ -258,7 +258,7 @@ public:
**/
void stop()
{
- mLogger(AsteriskSCF::System::Logging::Debug) << mName << ": setting stop";
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " setting stop";
boost::unique_lock<boost::shared_mutex> lock(mLock);
mStopped = true;
}
@@ -286,7 +286,7 @@ public:
QueuedTaskPtr current = popNextTask();
while (current)
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " executing " << current->name();
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " executing " << current->name();
//
// Properly implemented tasks will return true if the next task should
@@ -313,11 +313,11 @@ public:
**/
void failed()
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " entering failed state";
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " entering failed state";
QueuedTaskPtr current = popNextTask();
while (current)
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " failing " << current->name();
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " failing " << current->name();
current->fail();
current = popNextTask();
}
@@ -334,14 +334,14 @@ public:
**/
void destroy()
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " entering destroyed state";
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " entering destroyed state";
//
// NOTE: a request that is in progress will not be destroyed.
//
QueuedTaskPtr current = popNextTask();
while (current)
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " destroying " << current->name();
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " destroying " << current->name();
current->destroy();
current = popNextTask();
}
commit 59387df4b80bb1ccc1b84c13a130979698a2dc1f
Author: Brent Eagles <beagles at digium.com>
Date: Wed May 25 17:15:13 2011 -0230
Fix some more session/setup teardown issues and added some logging so its
possible to track which queues are being failed, destroyed or executing.
Also added an exclusionary form of SessionShutdownOperation to allow
us to skip a session that has indicated that it is stopped. It would
work without it, but it saves some extra calls.
diff --git a/src/SessionListener.cpp b/src/SessionListener.cpp
index 7aee4df..5fffe12 100644
--- a/src/SessionListener.cpp
+++ b/src/SessionListener.cpp
@@ -184,7 +184,7 @@ public:
//
session->shutdown(mListenerPrx, ResponseCodePtr());
- ShutdownSessionOperation shutdownOp(mListenerPrx, stopped->response, mLogger);
+ ShutdownSessionOperation shutdownOp(session->getSession(), mListenerPrx, stopped->response, mLogger);
mSessions->visitSessions(shutdownOp);
//
diff --git a/src/SessionOperations.cpp b/src/SessionOperations.cpp
index 0ef489a..d8ab630 100644
--- a/src/SessionOperations.cpp
+++ b/src/SessionOperations.cpp
@@ -40,18 +40,33 @@ void ConnectSessionOperation::operator()(const SessionWrapperPtr& s)
}
}
+ShutdownSessionOperation::ShutdownSessionOperation(const SessionPrx& exclude, const SessionListenerPrx& listener,
+ const ResponseCodePtr& response,
+ const Logger& logger) :
+ mExclude(exclude->ice_getIdentity()),
+ mListener(listener),
+ mResponse(response),
+ mLogger(logger),
+ mSkipExclude(false)
+{
+}
+
ShutdownSessionOperation::ShutdownSessionOperation(const SessionListenerPrx& listener,
const ResponseCodePtr& response,
const Logger& logger) :
mListener(listener),
mResponse(response),
- mLogger(logger)
+ mLogger(logger),
+ mSkipExclude(true)
{
}
void ShutdownSessionOperation::operator()(const SessionWrapperPtr& wrapper)
{
- wrapper->shutdown(mListener, mResponse);
+ if (mSkipExclude || wrapper->getSession()->ice_getIdentity() != mExclude)
+ {
+ wrapper->shutdown(mListener, mResponse);
+ }
}
RelayIndication::RelayIndication(const SessionPrx& exclude, const Logger& logger, const IndicationPtr& indication, bool includeConnected) :
diff --git a/src/SessionOperations.h b/src/SessionOperations.h
index cdf4b82..89e6c8b 100644
--- a/src/SessionOperations.h
+++ b/src/SessionOperations.h
@@ -42,15 +42,22 @@ private:
class ShutdownSessionOperation : public std::unary_function<SessionWrapperPtr, void>
{
public:
+ ShutdownSessionOperation(const AsteriskSCF::SessionCommunications::V1::SessionPrx& exclude,
+ const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
+ const AsteriskSCF::SessionCommunications::V1::ResponseCodePtr& response,
+ const AsteriskSCF::System::Logging::Logger& logger);
+
ShutdownSessionOperation(const AsteriskSCF::SessionCommunications::V1::SessionListenerPrx& listener,
- const AsteriskSCF::SessionCommunications::V1::ResponseCodePtr& response,
- const AsteriskSCF::System::Logging::Logger& logger);
+ const AsteriskSCF::SessionCommunications::V1::ResponseCodePtr& response,
+ const AsteriskSCF::System::Logging::Logger& logger);
void operator()(const SessionWrapperPtr& wrapper);
private:
+ Ice::Identity mExclude;
AsteriskSCF::SessionCommunications::V1::SessionListenerPrx mListener;
AsteriskSCF::SessionCommunications::V1::ResponseCodePtr mResponse;
AsteriskSCF::System::Logging::Logger mLogger;
+ bool mSkipExclude;
};
class RelayIndication : public std::unary_function<SessionWrapperPtr, void>
diff --git a/src/SessionWrapper.cpp b/src/SessionWrapper.cpp
index 883c614..cf7404f 100644
--- a/src/SessionWrapper.cpp
+++ b/src/SessionWrapper.cpp
@@ -261,9 +261,8 @@ protected:
{
BridgedSessionState oldState = mSession->setState(mState);
if (mStartStates.find(oldState) == mStartStates.end())
- {
+ {
mListener->failed();
- return false;
}
return true;
}
@@ -340,9 +339,9 @@ QueuedTasks createShutdownTasks(const SessionWrapperPtr& session, const SessionL
QueuedTasks tasks;
tasks.push_back(new SetStateFromTask(session, ::AsteriskSCF::Bridge::V1::Disconnected, statesToContinueOn));
- tasks.push_back(new RemoveBridgeTask(session, listener));
if (code)
{
+ tasks.push_back(new RemoveBridgeTask(session, listener));
tasks.push_back(new ShutdownMediaTask(session));
tasks.push_back(new SessionStopTask(session, code));
}
@@ -549,9 +548,14 @@ void SessionWrapper::shutdown(const SessionListenerPrx& listener, const Response
{
mLogger(Debug) << FUNLOG << ": beginning shutdown of " << mId;
QueuedTaskPtr shutdownRunner(new QueueableExecutor(createShutdownTasks(this, listener, code), mLogger));
- //
- // TODO: determine if the pending activites should be shutdown first.
- //
+ if (!code)
+ {
+ //
+ // This only happens if the session itself indicates that it is done, so we should
+ // make every attempt to not do any more.
+ //
+ mActivities->destroy();
+ }
mActivities->append(shutdownRunner);
//
// If shutdown is going to preempt stuff we need to uncomment this.
diff --git a/src/Tasks.h b/src/Tasks.h
index 57898b3..0b13188 100644
--- a/src/Tasks.h
+++ b/src/Tasks.h
@@ -16,6 +16,7 @@
#pragma once
#include <Ice/Ice.h>
+#include <IceUtil/UUID.h>
#include <boost/thread/locks.hpp>
#include <list>
#include <AsteriskSCF/logger.h>
@@ -95,6 +96,7 @@ public:
catch(...)
{
mListener->failed();
+ return false;
}
return true;
}
@@ -159,7 +161,7 @@ protected:
std::string mName;
QueuedTask() :
- mName("<unnamed>")
+ mName("QueuedTask")
{
}
@@ -205,14 +207,16 @@ class Executor : virtual public TaskListener
public:
Executor(const AsteriskSCF::System::Logging::Logger& logger) :
mLogger(logger),
- mStopped(true)
+ mStopped(true),
+ mName(IceUtil::generateUUID())
{
}
Executor(const QueuedTasks& tasks, const AsteriskSCF::System::Logging::Logger& logger) :
mTasks(tasks),
mLogger(logger),
- mStopped(true)
+ mStopped(true),
+ mName(IceUtil::generateUUID())
{
//
// As indicated above, the tasks are associated with this "executor" instance
@@ -231,6 +235,7 @@ public:
**/
void start()
{
+ mLogger(AsteriskSCF::System::Logging::Debug) << mName << ": starting running";
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
if (!mStopped)
@@ -253,6 +258,7 @@ public:
**/
void stop()
{
+ mLogger(AsteriskSCF::System::Logging::Debug) << mName << ": setting stop";
boost::unique_lock<boost::shared_mutex> lock(mLock);
mStopped = true;
}
@@ -280,7 +286,7 @@ public:
QueuedTaskPtr current = popNextTask();
while (current)
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": executing " << current->name();
+ mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " executing " << current->name();
//
// Properly implemented tasks will return true if the next task should
@@ -307,10 +313,11 @@ public:
**/
void failed()
{
+ mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " entering failed state";
QueuedTaskPtr current = popNextTask();
while (current)
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": failing " << current->name();
+ mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " failing " << current->name();
current->fail();
current = popNextTask();
}
@@ -327,13 +334,14 @@ public:
**/
void destroy()
{
+ mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " entering destroyed state";
//
// NOTE: a request that is in progress will not be destroyed.
//
QueuedTaskPtr current = popNextTask();
while (current)
{
- mLogger(AsteriskSCF::System::Logging::Debug) << ": destroying " << current->name();
+ mLogger(AsteriskSCF::System::Logging::Debug) << ": " << mName << " destroying " << current->name();
current->destroy();
current = popNextTask();
}
@@ -359,6 +367,7 @@ public:
**/
void append(const QueuedTaskPtr& newTask)
{
+ mLogger(AsteriskSCF::System::Logging::Debug) << " " << mName << " Appending new task " << newTask->name();
bool restart = false;
{
newTask->setListener(this);
@@ -397,6 +406,7 @@ protected:
QueuedTasks mTasks;
AsteriskSCF::System::Logging::Logger mLogger;
bool mStopped;
+ std::string mName;
};
typedef IceUtil::Handle<Executor> ExecutorPtr;
@@ -434,6 +444,7 @@ public:
void failed()
{
mListener->failed();
+ Executor::failed();
}
protected:
commit 9f78b40761e71cccf0e0e85e0f89235a2f77583f
Author: Brent Eagles <beagles at digium.com>
Date: Wed May 25 12:09:35 2011 -0230
Fixing bugs with indications.
- Added indication of translations and passes through indications that do not require further action on
the bridging side.
- Fixed terminal session wrapper operation so that it would not repeat work if the wrapper was already
set to the disconnected state.
- Changed order of operations on shutdown so media would be disconnected before the stop indication was already
passed along
diff --git a/src/BridgeImpl.cpp b/src/BridgeImpl.cpp
index 15dd93a..0cf2807 100755
--- a/src/BridgeImpl.cpp
+++ b/src/BridgeImpl.cpp
@@ -651,7 +651,12 @@ void BridgeImpl::shutdown(const Ice::Current& current)
pushUpdate(update);
update = 0;
{
- ShutdownSessionOperation shutdownOp(mSessionListenerPrx, new ResponseCode, mLogger);
+ //
+ // Currently the slice defines the response "Normal Clearing" as the default
+ // value so we shouldn't need to set anything here.
+ //
+ ResponseCodePtr responseCode = new ResponseCode;
+ ShutdownSessionOperation shutdownOp(mSessionListenerPrx, responseCode, mLogger);
mSessions->visitSessions(shutdownOp);
mListeners->stopped();
boost::unique_lock<boost::shared_mutex> lock(mLock);
diff --git a/src/SessionListener.cpp b/src/SessionListener.cpp
index 585f9f4..7aee4df 100644
--- a/src/SessionListener.cpp
+++ b/src/SessionListener.cpp
@@ -28,6 +28,77 @@ using namespace std;
namespace
{
+template <class FromT, class ToT>
+void indicationCopy(const FromT&, const ToT&)
+{
+ //
+ // NO-OP.
+ //
+}
+
+template <>
+void indicationCopy(const ProgressingIndicationPtr& source, const ProgressIndicationPtr& destination)
+{
+ destination->response = source->response;
+}
+
+template <class FromT, class ToT>
+ToT translateImpl(const IndicationPtr& f)
+{
+ FromT candidate = FromT::dynamicCast(f);
+ if (candidate)
+ {
+ typedef typename ToT::element_type Tval;
+ ToT result = new Tval;
+ indicationCopy(candidate, result);
+ return result;
+ }
+ return 0;
+}
+
+IndicationPtr translate(const AsteriskSCF::SessionCommunications::V1::IndicationPtr& source)
+{
+ {
+ FlashIndicationPtr indication = translateImpl<FlashedIndicationPtr, FlashIndicationPtr>(source);
+ if (indication)
+ {
+ return indication;
+ }
+ }
+ {
+ HoldIndicationPtr indication = translateImpl<HeldIndicationPtr, HoldIndicationPtr>(source);
+ if (indication)
+ {
+ return indication;
+ }
+ }
+ {
+ UnholdIndicationPtr indication = translateImpl<UnheldIndicationPtr, UnholdIndicationPtr>(source);
+ if (indication)
+ {
+ return indication;
+ }
+ }
+ {
+ RingIndicationPtr indication = translateImpl<RingingIndicationPtr, RingIndicationPtr>(source);
+ if (indication)
+ {
+ return indication;
+ }
+ }
+ {
+ ProgressIndicationPtr indication = translateImpl<ProgressingIndicationPtr, ProgressIndicationPtr>(source);
+ if (indication)
+ {
+ return indication;
+ }
+ }
+ //
+ // TODO: This probably shoudn't happen.
+ //
+ return source;
+}
+
//
// For events that require modification to the bridge, we use helper methods on the bridge itself.
// For events result in distribution to the bridge sessions, we copy the current sessions and
@@ -75,14 +146,6 @@ public:
throw;
}
}
- else if ((ringing = AsteriskSCF::SessionCommunications::V1::RingingIndicationPtr::dynamicCast(indication)))
- {
- //
- // TODO: Who gets the ring notifications will likely depend on configuration, etc.
- //
- RingSessionOperation ringer(source, mLogger);
- mSessions->visitSessions(ringer);
- }
else if ((stopped = AsteriskSCF::SessionCommunications::V1::StoppedIndicationPtr::dynamicCast(indication)))
{
string proxyString = source->ice_toString();
@@ -114,7 +177,6 @@ public:
//
// This returns only the active sessions.
//
- SessionSeq currentSessions = mSessions->getSessionSeq();
//
// Shutdown is handled asynchronously, so there won't be any exceptions that need to be caught here.
@@ -122,6 +184,9 @@ public:
//
session->shutdown(mListenerPrx, ResponseCodePtr());
+ ShutdownSessionOperation shutdownOp(mListenerPrx, stopped->response, mLogger);
+ mSessions->visitSessions(shutdownOp);
+
//
// Ideally, we'd wait for an affirmative response from the session being removed from the bridge
// before doing this next part.
@@ -130,11 +195,13 @@ public:
{
try
{
+ SessionSeq currentSessions = mSessions->getSessionSeq();
+
//
// TODO: Should be determined by policy. 2 is the proper upper limit because
// the session in question will not have been removed from the bridge yet.
//
- if (currentSessions.size() <= 2)
+ if (currentSessions.size() < 2)
{
mBridgePrx->shutdown();
}
@@ -145,6 +212,15 @@ public:
}
}
}
+ else
+ {
+ IndicationPtr commandIndication = translate(indication);
+ //
+ // TODO: Who gets the ring notifications will likely depend on configuration, etc.
+ //
+ RelayIndication relayer(source, mLogger, commandIndication, false);
+ mSessions->visitSessions(relayer);
+ }
}
void onActivate(const BridgePrx& bridgePrx, const SessionListenerPrx& listenerPrx)
diff --git a/src/SessionOperations.cpp b/src/SessionOperations.cpp
index 56a5c5a..0ef489a 100644
--- a/src/SessionOperations.cpp
+++ b/src/SessionOperations.cpp
@@ -54,23 +54,25 @@ void ShutdownSessionOperation::operator()(const SessionWrapperPtr& wrapper)
wrapper->shutdown(mListener, mResponse);
}
-RingSessionOperation::RingSessionOperation(const SessionPrx& exclude, const Logger& logger) :
+RelayIndication::RelayIndication(const SessionPrx& exclude, const Logger& logger, const IndicationPtr& indication, bool includeConnected) :
mExclude(exclude->ice_getIdentity()),
- mLogger(logger)
+ mLogger(logger),
+ mIndication(indication),
+ mIncludeConnected(includeConnected)
{
}
-void RingSessionOperation::operator()(const SessionWrapperPtr& session)
+void RelayIndication::operator()(const SessionWrapperPtr& session)
{
SessionPrx s = session->getSession();
- if (s->ice_getIdentity() != mExclude)
+ if (s->ice_getIdentity() != mExclude && (mIncludeConnected || !session->isConnected()))
{
try
{
//
// TODO: AMI.. or would this be better as a oneway. Do we care if we get a response etc?
//
- s->indicate(new AsteriskSCF::SessionCommunications::V1::RingIndication());
+ s->indicate(mIndication);
}
catch (const Ice::ObjectNotExistException& ex)
{
diff --git a/src/SessionOperations.h b/src/SessionOperations.h
index 1976059..cdf4b82 100644
--- a/src/SessionOperations.h
+++ b/src/SessionOperations.h
@@ -53,17 +53,21 @@ private:
AsteriskSCF::System::Logging::Logger mLogger;
};
-class RingSessionOperation : public std::unary_function<SessionWrapperPtr, void>
+class RelayIndication : public std::unary_function<SessionWrapperPtr, void>
{
public:
- RingSessionOperation(const AsteriskSCF::SessionCommunications::V1::SessionPrx& exclude,
- const AsteriskSCF::System::Logging::Logger& logger);
+ RelayIndication(const AsteriskSCF::SessionCommunications::V1::SessionPrx& exclude,
+ const AsteriskSCF::System::Logging::Logger& logger,
+ const AsteriskSCF::SessionCommunications::V1::IndicationPtr& indication,
+ bool includeConnected);
void operator()(const SessionWrapperPtr& session);
private:
Ice::Identity mExclude;
AsteriskSCF::System::Logging::Logger mLogger;
+ AsteriskSCF::SessionCommunications::V1::IndicationPtr mIndication;
+ bool mIncludeConnected;
};
class IfStateCriteria
diff --git a/src/SessionWrapper.cpp b/src/SessionWrapper.cpp
index c447486..883c614 100644
--- a/src/SessionWrapper.cpp
+++ b/src/SessionWrapper.cpp
@@ -18,6 +18,7 @@
#include <Ice/Ice.h>
#include <AsteriskSCF/logger.h>
#include "ServiceUtil.h"
+#include <set>
using namespace AsteriskSCF::System::Logging;
using namespace AsteriskSCF::SessionCommunications::V1;
@@ -28,6 +29,9 @@ using namespace std;
namespace
{
+
+typedef set<BridgedSessionState> StateSet;
+
//
// NOTE: This object is used in support of making connect() calls on sessions via AMI. It's a fair candidate for a
// singleton as you could pass the session as a cookie to the AMI call. However, it is desirable to have some logging,
@@ -124,7 +128,6 @@ protected:
void failed(const Ice::Exception& ex)
{
-
try
{
ex.ice_throw();
@@ -178,7 +181,6 @@ public:
protected:
bool executeImpl()
{
-
mSession->getSession()->begin_stop(mCode,
newCallback_Session_stop(this, &SessionStopTask::removed,
&SessionStopTask::failed));
@@ -241,6 +243,38 @@ private:
AsteriskSCF::Bridge::V1::BridgedSessionState mOldState;
};
+class SetStateFromTask : public QueuedTask
+{
+public:
+ SetStateFromTask(const SessionWrapperPtr& session,
+ const AsteriskSCF::Bridge::V1::BridgedSessionState newState,
+ const StateSet& startStates) :
+ QueuedTask("SetStateFromTask"),
+ mSession(session),
+ mState(newState),
+ mStartStates(startStates)
+ {
+ }
+
+protected:
+ bool executeImpl()
+ {
+ BridgedSessionState oldState = mSession->setState(mState);
+ if (mStartStates.find(oldState) == mStartStates.end())
+ {
+ mListener->failed();
+ return false;
+ }
+ return true;
+ }
+
+private:
+ SessionWrapperPtr mSession;
+ AsteriskSCF::Bridge::V1::BridgedSessionState mState;
+ AsteriskSCF::Bridge::V1::BridgedSessionState mOldState;
+ StateSet mStartStates;
+};
+
class ShutdownMediaTask : public QueuedTask
{
public:
@@ -250,25 +284,16 @@ public:
}
protected:
+ //
+ // NOTE: This has been changed from async to sync for the time being.. the unplugMedia
+ // call is inherently async, so we can just go on.
+ //
bool executeImpl()
{
mSession->unplugMedia();
- return false;
- }
-
- void done()
- {
- mListener->succeeded();
+ return true;
}
- void failed(const Ice::Exception&)
- {
- //
- // TODO: Log exception.
- //
- mListener->failed();
- }
-
private:
SessionWrapperPtr mSession;
};
@@ -306,26 +331,28 @@ QueuedTasks createShutdownTasks(const SessionWrapperPtr& session, const SessionL
const ResponseCodePtr& code)
{
//
- // Tasks are a queued, so they go in order that they will be processed.
+ // We use this task instead of changing states because it will prevent the
+ // queue from continuing if the session is already connected or done.
//
+ StateSet statesToContinueOn;
+ statesToContinueOn.insert(::AsteriskSCF::Bridge::V1::Added);
+ statesToContinueOn.insert(::AsteriskSCF::Bridge::V1::Connected);
+
QueuedTasks tasks;
- tasks.push_back(new SetStateTask(session, Disconnected));
+ tasks.push_back(new SetStateFromTask(session, ::AsteriskSCF::Bridge::V1::Disconnected, statesToContinueOn));
tasks.push_back(new RemoveBridgeTask(session, listener));
if (code)
{
+ tasks.push_back(new ShutdownMediaTask(session));
tasks.push_back(new SessionStopTask(session, code));
}
- //
- // TODO: These two tasks should be reversed really.
- //
- tasks.push_back(new ShutdownMediaTask(session));
return tasks;
}
QueuedTasks createSetupTasks(const SessionWrapperPtr& session)
{
QueuedTasks tasks;
- tasks.push_back(new SetStateTask(session, Connected));
+ tasks.push_back(new SetStateTask(session, ::AsteriskSCF::Bridge::V1::Connected));
tasks.push_back(new ConnectMediaTask(session));
return tasks;
}
@@ -532,19 +559,19 @@ void SessionWrapper::shutdown(const SessionListenerPrx& listener, const Response
// shutdownRunner->start();
}
-void SessionWrapper::setState(const AsteriskSCF::Bridge::V1::BridgedSessionState newState)
+AsteriskSCF::Bridge::V1::BridgedSessionState SessionWrapper::setState(const AsteriskSCF::Bridge::V1::BridgedSessionState newState)
{
mLogger(Debug) << FUNLOG << ": updating state " << mId;
BridgedSessionPtr copyOfNewState;
+ AsteriskSCF::Bridge::V1::BridgedSessionState oldState;
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
- //
- // TODO:
- //
+ oldState = mSession->currentState;
mSession->currentState = newState;
copyOfNewState = createUpdate();
}
pushUpdate(copyOfNewState);
+ return oldState;
}
void SessionWrapper::unplugMedia()
diff --git a/src/SessionWrapper.h b/src/SessionWrapper.h
index 0c98112..e0a8835 100644
--- a/src/SessionWrapper.h
+++ b/src/SessionWrapper.h
@@ -127,7 +127,7 @@ public:
//
// TODO: Refactor so these methods don't need to be exposed.
//
- void setState(const AsteriskSCF::Bridge::V1::BridgedSessionState newState);
+ AsteriskSCF::Bridge::V1::BridgedSessionState setState(const AsteriskSCF::Bridge::V1::BridgedSessionState newState);
/**
* Disconnection helper.
diff --git a/test/TestBridging.cpp b/test/TestBridging.cpp
index a2ad429..f5569e9 100644
--- a/test/TestBridging.cpp
+++ b/test/TestBridging.cpp
@@ -420,6 +420,7 @@ public:
BOOST_CHECK(servant->createCalls() == 1);
+ IceUtil::ThreadControl::sleep(IceUtil::Time::seconds(2));
channel.commands()->getlog(idA, log);
bool findStop = find(log, "stop");
if (!findStop)
@@ -615,6 +616,7 @@ public:
BOOST_CHECK(bridgeListener->addedCount() == 2);
bridge->shutdown();
+ IceUtil::ThreadControl::sleep(IceUtil::Time::seconds(2));
BOOST_CHECK(servant->createCalls() == 1);
channel.commands()->getlog(idA, log);
bool findStop = find(log, "stop");
commit 231250c599d86a6bec9c0cc6038ed2e33a9feb3f
Author: Brent Eagles <beagles at digium.com>
Date: Mon May 23 12:10:41 2011 -0230
Add missing configuration line. It's necessary to have more than one
client thread in the ami/amd enabled bridging service.
diff --git a/config/bridging.conf b/config/bridging.conf
index 7ba3738..9fa634f 100644
--- a/config/bridging.conf
+++ b/config/bridging.conf
@@ -2,6 +2,7 @@
# Adapter parameters for this component
BridgeManager.BridgeService.Endpoints=default
+Ice.ThreadPool.Client.Size=4
# A proxy to the service locator management service
ServiceLocatorManagementProxy=LocatorServiceManagement:tcp -p 4422
commit bf3094a3314b4cc4bedeb59df947080b744df9ed
Author: Brent Eagles <beagles at digium.com>
Date: Fri May 20 17:27:11 2011 -0230
Modified test suite so it should allow connect indications to process properly.
Previously it had been forcing both sessions to send a Connected indication
into the bridge, rendering them both "fully connected" in the bridge's view and
frequently preventing the corresponding "indicate(Connected)" from being called
on the peer.
Also added some missing locking in the splicer class.
diff --git a/src/MediaSplicer.cpp b/src/MediaSplicer.cpp
index deac0b7..4bdcdae 100755
--- a/src/MediaSplicer.cpp
+++ b/src/MediaSplicer.cpp
@@ -457,6 +457,7 @@ public:
data->peer->update(connector, data->outgoingPairings, data->incomingPairings);
}
connector->initialUpdate();
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
mSessions.push_back(MediaSessionStruct(data->mediaSession, connector));
return connector;
}
@@ -536,13 +537,16 @@ public:
{
vector<StreamSourceSeq> allSources;
- //
- // TODO: This is not really correct at all. When the bridge implements conferencing chances, everything
- // will come into a mix and go out.
- //
- for (MediaSessions::iterator i = mSessions.begin(); i != mSessions.end(); ++i)
{
- allSources.push_back(i->mediaSession->getSources());
+ boost::shared_lock<boost::shared_mutex> lock(mLock);
+ //
+ // TODO: This is not really correct at all. When the bridge implements conferencing chances, everything
+ // will come into a mix and go out.
+ //
+ for (MediaSessions::iterator i = mSessions.begin(); i != mSessions.end(); ++i)
+ {
+ allSources.push_back(i->mediaSession->getSources());
+ }
}
vector<OutgoingPairing> result;
@@ -577,13 +581,16 @@ public:
vector<IncomingPairing> findCompatiblePairings(const StreamSourceSeq& sources)
{
vector<StreamSinkSeq> allSinks;
- //
- // TODO: This is not really correct at all. When the bridge implements conferencing chances, everything
- // will come into a mix and go out.
- //
- for (MediaSessions::iterator i = mSessions.begin(); i != mSessions.end(); ++i)
{
- allSinks.push_back(i->mediaSession->getSinks());
+ boost::shared_lock<boost::shared_mutex> lock(mLock);
+ //
+ // TODO: This is not really correct at all. When the bridge implements conferencing chances, everything
+ // will come into a mix and go out.
+ //
+ for (MediaSessions::iterator i = mSessions.begin(); i != mSessions.end(); ++i)
+ {
+ allSinks.push_back(i->mediaSession->getSinks());
+ }
}
vector<IncomingPairing> result;
@@ -805,9 +812,7 @@ public:
MakeConnections(const MediaConnectorBuilderPtr& materials) :
QueuedTask("MakeConnections"),
mMaterials(materials),
- mIncomingPairings(mMaterials->incomingPairings.size(), 0),
- mOutgoingPairings(mMaterials->outgoingPairings.size(), 0),
- mTotalCount(mIncomingPairings.size() + mOutgoingPairings.size())
+ mCurrentCount(0)
{
}
@@ -889,6 +894,9 @@ protected:
//
return true;
}
+ mIncomingPairings.resize(mMaterials->incomingPairings.size());
+ mOutgoingPairings.resize(mMaterials->outgoingPairings.size());
+ mTotalCount = (mIncomingPairings.size() + mOutgoingPairings.size());
//
// If there are lots of pairings, this could result in a "flurry" of AMI requests. I'm not sure that it is
@@ -951,6 +959,10 @@ protected:
}
if (done)
{
+ if (mMaterials->connector)
+ {
+ mMaterials->connector->update(mMaterials->peer, mMaterials->outgoingPairings, mMaterials->incomingPairings);
+ }
mListener->succeeded();
}
}
@@ -976,7 +988,7 @@ private:
Ice::IntSeq mIncomingPairings;
Ice::IntSeq mOutgoingPairings;
size_t mCurrentCount;
- const size_t mTotalCount;
+ size_t mTotalCount;
};
//
diff --git a/test/TestBridging.cpp b/test/TestBridging.cpp
index b5546db..a2ad429 100644
--- a/test/TestBridging.cpp
+++ b/test/TestBridging.cpp
@@ -402,12 +402,14 @@ public:
BOOST_CHECK(!find(log, "start"));
a->start();
- b->start();
channel.commands()->getlog(idA, log);
BOOST_CHECK(find(log, "start"));
+
+ IceUtil::ThreadControl::sleep(IceUtil::Time::seconds(2));
channel.commands()->getlog(idB, log);
- BOOST_CHECK(find(log, "start"));
+ dumplog(log);
+ BOOST_CHECK(find(log, "Connect"));
//
// Should result in a media hookup!
@@ -498,11 +500,10 @@ public:
channel.commands()->getlog(idB, log);
BOOST_CHECK(!find(log, "start"));
- a->start();
b->start();
-
+ IceUtil::ThreadControl::sleep(IceUtil::Time::seconds(2));
channel.commands()->getlog(idA, log);
- BOOST_CHECK(find(log, "start"));
+ BOOST_CHECK(find(log, "Connect"));
channel.commands()->getlog(idB, log);
BOOST_CHECK(find(log, "start"));
@@ -597,11 +598,10 @@ public:
channel.commands()->getlog(idB, log);
BOOST_CHECK(!find(log, "start"));
- a->start();
b->start();
-
+ IceUtil::ThreadControl::sleep(IceUtil::Time::seconds(2));
channel.commands()->getlog(idA, log);
- BOOST_CHECK(find(log, "start"));
+ BOOST_CHECK(find(log, "Connect"));
channel.commands()->getlog(idB, log);
BOOST_CHECK(find(log, "start"));
commit 8cd95c68bbaf4a0fe63b1b4d91587ae4aec1aa36
Author: Brent Eagles <beagles at digium.com>
Date: Fri May 20 14:30:25 2011 -0230
Fix a bug that would prevent media sessions from being setup properly.
Also changed the order of tasks for media setup to create and register
the connectors prior to attempting to make connections. It fits a bit
better with the asynchronous nature of how everything works.
diff --git a/src/MediaSplicer.cpp b/src/MediaSplicer.cpp
index de7b7c8..deac0b7 100755
--- a/src/MediaSplicer.cpp
+++ b/src/MediaSplicer.cpp
@@ -22,6 +22,7 @@
#include <vector>
#include "ServiceUtil.h"
#include "SessionWrapper.h"
+#include <boost/enable_shared_from_this.hpp>
//
// TODO:
@@ -45,7 +46,7 @@ namespace BridgeService
class MediaSplicerI;
class MediaConnectorI;
typedef IceUtil::Handle<MediaConnectorI> MediaConnectorIPtr;
-typedef IceUtil::Handle<MediaSplicerI> MediaSplicerIPtr;
+typedef boost::shared_ptr<MediaSplicerI> MediaSplicerIPtr;
//
// Some types that are used throughout.
@@ -67,6 +68,7 @@ struct MediaConnectorBuilder : public IceUtil::Shared
AsteriskSCF::Media::V1::StreamSourceSeq sources;
AsteriskSCF::Media::V1::StreamSinkSeq sinks;
MediaConnectorIPtr peer;
+ MediaConnectorIPtr connector;
OutgoingPairings outgoingPairings;
IncomingPairings incomingPairings;
};
@@ -432,7 +434,7 @@ QueuedTasks createMediaConnectTasks(const SessionWrapperPtr& session,
//
// TODO: This needs to register the streams with an active threaded mixing element.
//
-class MediaSplicerI : public IceUtil::Shared
+class MediaSplicerI : public boost::enable_shared_from_this<MediaSplicerI>
{
public:
MediaSplicerI(const Ice::CommunicatorPtr& communicator, const string& bridgeId, const ReplicatorSmartPrx& replicator, const Logger& logger) :
@@ -496,7 +498,8 @@ public:
// We do not bother tracking the executor for now. A tidy shutdown would probably want to clean this up.
// An alternative is to pass back the queued tasks to the caller and let them start and stop the process.
//
- ExecutorPtr taskExecutor(new Executor(createMediaConnectTasks(session, sessionPrx, existing, this), mLogger));
+ ExecutorPtr taskExecutor(new Executor(createMediaConnectTasks(session, sessionPrx, existing,
+ shared_from_this()), mLogger));
taskExecutor->start();
}
@@ -671,8 +674,6 @@ private:
}
};
-typedef IceUtil::Handle<MediaSplicerI> MediaSplicerIPtr;
-
class GetMediaSession : public QueuedTask
... 831 lines suppressed ...
--
asterisk-scf/integration/bridging.git
More information about the asterisk-scf-commits
mailing list