[asterisk-scf-commits] asterisk-scf/integration/bridging.git branch "bridge-replication-issues" created.
Commits to the Asterisk SCF project code repositories
asterisk-scf-commits at lists.digium.com
Thu Dec 22 10:28:43 CST 2011
branch "bridge-replication-issues" has been created
at 934164aaf68112a21e589db5df578eecaee8483c (commit)
- Log -----------------------------------------------------------------
commit 934164aaf68112a21e589db5df578eecaee8483c
Author: Brent Eagles <beagles at digium.com>
Date: Thu Dec 22 12:54:00 2011 -0330
- Fix up some logging messages.
- Null handle checks in some of the tasks (exceptions were showing up in
the test suite.
- Quick hack for setting log level by confiugration.. the component's
logger isn't being used, instead the file scoped copy is.. .wrong! Still,
didn't change that.. enough unrelated changes made in this commit already.
- A lot of stuff was not calling removeState when it was going away.
- The test suite doesn't really do replication correctly.. which might be
part of the problem.
diff --git a/config/test_bridging.conf b/config/test_bridging.conf
index 5ec14f2..8e7ead3 100644
--- a/config/test_bridging.conf
+++ b/config/test_bridging.conf
@@ -82,6 +82,7 @@ TestBridge.ServiceAdapter.ThreadPool.Size=2
TestBridge.BackplaneAdapter.Endpoints=default -h 127.0.0.1 -p 57001
TestBridge.BackplaneAdapter.ThreadPool.Size=2
TestBridge.Standby=false
+TestBridge.LogLevel=0
TestBridge2.InstanceName=TestBridge2
TestBridge2.BridgeManagerObjectId=TestBridgeManager2
@@ -90,6 +91,9 @@ TestBridge2.ServiceAdapter.Endpoints=default -h 127.0.0.1 -p 57010
TestBridge2.ServiceAdapter.ThreadPool.Size=4
TestBridge2.BackplaneAdapter.Endpoints=default -h 127.0.0.1 -p 57011
TestBridge2.BackplaneAdapter.ThreadPool.Size=4
+TestBridge2.LogLevel=0
+#Ice.Trace.Network=1
+#Ice.Trace.Protocol=1
#
# Configuration for the bridge state replicator.
diff --git a/src/BridgeImpl.cpp b/src/BridgeImpl.cpp
index 5f2c03a..1d10928 100755
--- a/src/BridgeImpl.cpp
+++ b/src/BridgeImpl.cpp
@@ -304,22 +304,34 @@ public:
protected:
bool executeImpl()
{
+ mLogger(Trace) << FUNLOG;
+ if (!mSourceSession)
+ {
+ mLogger(Trace) << FUNLOG << " Source session is nil, returning early with true";
+ return true;
+ }
try
{
- mLogger(Trace) << FUNLOG;
// Forward the ConnectedLine to each bridged session.
SessionSeq sessions = mBridge->getSessions()->getSessionSeq();
for(SessionSeq::iterator i = sessions.begin();
i != sessions.end(); ++i)
{
- if ((*i)->ice_getIdentity() == mSourceSession->ice_getIdentity())
+ if (!(*i) || (*i)->ice_getIdentity() == mSourceSession->ice_getIdentity())
{
continue;
}
SessionWrapperPtr destSessionWrapper = mBridge->getSessions()->getSession(*i);
- forward(destSessionWrapper, mRedirections);
+ if (destSessionWrapper)
+ {
+ forward(destSessionWrapper, mRedirections);
+ }
+ else
+ {
+ mLogger(Debug) << " Destination wrapper for session does not exist, continuing";
+ }
}
}
@@ -339,31 +351,37 @@ protected:
RedirectionsPtr destSpecificRedirections = redirections;
PartyIdHooksPtr partyIdHooks = mBridge->getPartyIdHooks();
-
- // Allow the ForwardingRedirectionsPartyId hooks to alter the Redirections record.
- for(vector<ForwardingRedirectionsPartyIdHookPrx>::const_iterator i = partyIdHooks->forwardingRedirectionsHooks.begin();
- i != partyIdHooks->forwardingRedirectionsHooks.end(); ++i)
+ if (partyIdHooks)
{
- try
+ // Allow the ForwardingRedirectionsPartyId hooks to alter the Redirections record.
+ for(vector<ForwardingRedirectionsPartyIdHookPrx>::const_iterator i = partyIdHooks->forwardingRedirectionsHooks.begin();
+ i != partyIdHooks->forwardingRedirectionsHooks.end(); ++i)
{
- // Apply this hook.
- AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyForwardingRedirections(mSourceSession,
- destinationSession->getSession(),
- currentRedirections, destSpecificRedirections);
-
- if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ try
{
- currentRedirections = destSpecificRedirections;
+ // Apply this hook.
+ AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyForwardingRedirections(mSourceSession,
+ destinationSession->getSession(),
+ currentRedirections, destSpecificRedirections);
+
+ if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ {
+ currentRedirections = destSpecificRedirections;
+ }
+ }
+ catch(const std::exception& e)
+ {
+ mLogger(Warning) << FUNLOG << " : " << e.what();
}
- }
- catch(const std::exception& e)
- {
- mLogger(Warning) << FUNLOG << " : " << e.what();
}
}
- // Forward the info via the SessionController for this session.
- destinationSession->getSessionController()->updateRedirections(currentRedirections);
+ SessionControllerPrx sessionController = destinationSession->getSessionController();
+ if (sessionController)
+ {
+ // Forward the info via the SessionController for this session.
+ destinationSession->getSessionController()->updateRedirections(currentRedirections);
+ }
}
private:
@@ -401,10 +419,14 @@ public:
protected:
bool executeImpl()
{
+ mLogger(Trace) << FUNLOG;
+ if (!mSourceSession)
+ {
+ mLogger(Trace) << FUNLOG << " Source session is nil, returning early with true";
+ return true;
+ }
try
{
- mLogger(Trace) << FUNLOG;
-
ConnectedLinePtr currentConnectedLine;
SessionWrapperPtr sourceWrapper = mBridge->getSessions()->getSession(mSourceSession);
bool isSet = sourceWrapper->getConnectedLine(currentConnectedLine);
@@ -420,13 +442,20 @@ protected:
for(SessionSeq::iterator i = sessions.begin();
i != sessions.end(); ++i)
{
- if ((*i)->ice_getIdentity() == mSourceSession->ice_getIdentity())
+ if (!(*i) || (*i)->ice_getIdentity() == mSourceSession->ice_getIdentity())
{
continue;
}
SessionWrapperPtr destSessionWrapper = mBridge->getSessions()->getSession(*i);
- forward(destSessionWrapper, currentConnectedLine);
+ if (destSessionWrapper)
+ {
+ forward(destSessionWrapper, currentConnectedLine);
+ }
+ else
+ {
+ mLogger(Debug) << " Destination wrapper for session does not exist, continuing";
+ }
}
}
@@ -447,30 +476,37 @@ protected:
PartyIdHooksPtr partyIdHooks = mBridge->getPartyIdHooks();
- // Allow the ForwardingConnectedLinePartyId hooks to alter the ConnectedLine record.
- for(vector<ForwardingConnectedLinePartyIdHookPrx>::const_iterator i = partyIdHooks->forwardingConnectedLineHooks.begin();
- i != partyIdHooks->forwardingConnectedLineHooks.end(); ++i)
+ if (partyIdHooks)
{
- try
+ // Allow the ForwardingConnectedLinePartyId hooks to alter the ConnectedLine record.
+ for(vector<ForwardingConnectedLinePartyIdHookPrx>::const_iterator i = partyIdHooks->forwardingConnectedLineHooks.begin();
+ i != partyIdHooks->forwardingConnectedLineHooks.end(); ++i)
{
- // Apply a hook
- AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyForwardingConnectedLine(mSourceSession,
- destinationSession->getSession(),
- currentConnectedLine, destSpecificConnectedLine);
-
- if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ try
{
- currentConnectedLine = destSpecificConnectedLine;
+ // Apply a hook
+ AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyForwardingConnectedLine(mSourceSession,
+ destinationSession->getSession(),
+ currentConnectedLine, destSpecificConnectedLine);
+
+ if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ {
+ currentConnectedLine = destSpecificConnectedLine;
+ }
+ }
+ catch (const std::exception& e)
+ {
+ mLogger(Debug) << FUNLOG << " : " << e.what();
}
- }
- catch (const std::exception& e)
- {
- mLogger(Debug) << FUNLOG << " : " << e.what();
}
}
- // Forward the info via the SessionController for this session.
- destinationSession->getSessionController()->updateConnectedLine(currentConnectedLine);
+ SessionControllerPrx sessionController = destinationSession->getSessionController();
+ if (sessionController)
+ {
+ // Forward the info via the SessionController for this session.
+ destinationSession->getSessionController()->updateConnectedLine(currentConnectedLine);
+ }
}
private:
@@ -497,22 +533,35 @@ public:
protected:
bool executeImpl()
{
+ mLogger(Trace) << FUNLOG;
+ if (!mSource)
+ {
+ mLogger(Trace) << FUNLOG << " Source session is nil, returning early with true";
+ return true;
+ }
+
try
{
- mLogger(Trace) << FUNLOG;
// Forward the Caller to each bridged session.
SessionSeq sessions = mBridge->getSessions()->getSessionSeq();
for(SessionSeq::iterator i = sessions.begin();
i != sessions.end(); ++i)
{
- if ((*i)->ice_getIdentity() == mSource->ice_getIdentity())
+ if (!(*i) || (*i)->ice_getIdentity() == mSource->ice_getIdentity())
{
continue;
}
SessionWrapperPtr destSessionWrapper = mBridge->getSessions()->getSession(*i);
- forward(destSessionWrapper, mCallerID);
+ if (destSessionWrapper)
+ {
+ forward(destSessionWrapper, mCallerID);
+ }
+ else
+ {
+ mLogger(Debug) << " Destination wrapper for session does not exist, continuing";
+ }
}
}
@@ -524,7 +573,7 @@ protected:
return true;
}
- void forward(const SessionWrapperPtr destinationSession, const CallerPtr& callerID)
+ void forward(const SessionWrapperPtr& destinationSession, const CallerPtr& callerID)
{
mLogger(Trace) << FUNLOG;
@@ -533,30 +582,38 @@ protected:
PartyIdHooksPtr partyIdHooks = mBridge->getPartyIdHooks();
- // Allow the ForwardingConnectedLinePartyId hooks to alter the ConnectedLine record.
- for(vector<ForwardingCallerPartyIdHookPrx>::const_iterator i = partyIdHooks->forwardingCallerHooks.begin();
- i != partyIdHooks->forwardingCallerHooks.end(); ++i)
+ if (partyIdHooks)
{
- try
+ // Allow the ForwardingConnectedLinePartyId hooks to alter the ConnectedLine record.
+ for(vector<ForwardingCallerPartyIdHookPrx>::const_iterator i = partyIdHooks->forwardingCallerHooks.begin();
+ i != partyIdHooks->forwardingCallerHooks.end(); ++i)
{
- // Apply a hook
- AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyForwardingCaller(mSource,
- destinationSession->getSession(),
- currentCallerID, destSpecificCallerID);
-
- if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ try
{
- currentCallerID = destSpecificCallerID;
+ // Apply a hook
+ AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyForwardingCaller(mSource,
+ destinationSession->getSession(),
+ currentCallerID, destSpecificCallerID);
+
+ if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ {
+ currentCallerID = destSpecificCallerID;
+ }
+ }
+ catch (const std::exception& e)
+ {
+ mLogger(Debug) << FUNLOG << " : " << e.what();
}
- }
- catch (const std::exception& e)
- {
- mLogger(Debug) << FUNLOG << " : " << e.what();
}
}
-
+ //
// Forward the info via the SessionController for this session.
- destinationSession->getSessionController()->updateCallerID(currentCallerID);
+ //
+ SessionControllerPrx sessionController = destinationSession->getSessionController();
+ if (sessionController)
+ {
+ sessionController->updateCallerID(currentCallerID);
+ }
}
private:
@@ -589,31 +646,43 @@ protected:
bool executeImpl()
{
mLogger(Trace) << FUNLOG;
+ if (!mSourceSession)
+ {
+ mLogger(Trace) << FUNLOG << "Source session is nil, returning early with true.";
+ return true;
+ }
ConnectedLinePtr currentConnectedLine = mConnectedLine;
ConnectedLinePtr updatedConnectedLine = mConnectedLine;
SessionWrapperPtr wrapper = mBridge->getSessions()->getSession(mSourceSession);
- PartyIdHooksPtr partyIdHooks = mBridge->getPartyIdHooks();
+ if (!wrapper)
+ {
+ mLogger(Debug) << "Unable to find matching session for, returning early with true.";
+ }
- // Allow the ReceivedConnectedLinePartyId hooks to alter the ConnectedLine record.
- for(vector<ReceivedConnectedLinePartyIdHookPrx>::const_iterator i = partyIdHooks->receivedConnectedLineHooks.begin();
- i != partyIdHooks->receivedConnectedLineHooks.end(); ++i)
+ PartyIdHooksPtr partyIdHooks = mBridge->getPartyIdHooks();
+ if (partyIdHooks)
{
- try
+ // Allow the ReceivedConnectedLinePartyId hooks to alter the ConnectedLine record.
+ for(vector<ReceivedConnectedLinePartyIdHookPrx>::const_iterator i = partyIdHooks->receivedConnectedLineHooks.begin();
+ i != partyIdHooks->receivedConnectedLineHooks.end(); ++i)
{
- // Apply this hook.
- AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyReceivedConnectedLine(mSourceSession,
- currentConnectedLine, updatedConnectedLine);
-
- if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ try
{
- currentConnectedLine = updatedConnectedLine;
+ // Apply this hook.
+ AsteriskSCF::System::Hook::V1::HookResult hookResult = (*i)->modifyReceivedConnectedLine(mSourceSession,
+ currentConnectedLine, updatedConnectedLine);
+
+ if (hookResult.status == AsteriskSCF::System::Hook::V1::Succeeded)
+ {
+ currentConnectedLine = updatedConnectedLine;
+ }
+ }
+ catch (const std::exception& e)
+ {
+ mLogger(Debug) << FUNLOG << " : " << e.what();
}
- }
- catch (const std::exception& e)
- {
- mLogger(Debug) << FUNLOG << " : " << e.what();
}
}
@@ -1497,7 +1566,6 @@ void BridgeImpl::shutdown(const Ice::Current& current)
update = createUpdate();
}
pushUpdate(update);
- update = 0;
{
//
// Currently the slice defines the response "Normal Clearing" as the default
@@ -1513,12 +1581,24 @@ void BridgeImpl::shutdown(const Ice::Current& current)
//
// Remove references to the session listener implementation.
//
- update = createUpdate();
mObjAdapter->remove(mSessionListenerPrx->ice_getIdentity());
mSessionListener = 0;
}
+
+ try
+ {
+ if (replicate())
+ {
+ Ice::StringSeq keys;
+ keys.push_back(mState->key);
+ mReplicator->removeState(keys);
+ }
+ }
+ catch (const Ice::Exception&)
+ {
+ }
+
mSessions = 0;
- pushUpdate(update);
}
void BridgeImpl::destroy(const Ice::Current& current)
@@ -1549,6 +1629,19 @@ void BridgeImpl::destroy(const Ice::Current& current)
// Remove references to the session listener implementation.
//
mObjAdapter->remove(mSessionListenerPrx->ice_getIdentity());
+
+ try
+ {
+ if (replicate())
+ {
+ Ice::StringSeq keys;
+ keys.push_back(mState->key);
+ mReplicator->removeState(keys);
+ }
+ }
+ catch (const Ice::Exception&)
+ {
+ }
}
void BridgeImpl::addListener(const BridgeListenerPrx& listener, const Ice::Current& current)
diff --git a/src/BridgeManagerImpl.cpp b/src/BridgeManagerImpl.cpp
index 18fd3da..a317d0c 100644
--- a/src/BridgeManagerImpl.cpp
+++ b/src/BridgeManagerImpl.cpp
@@ -124,7 +124,7 @@ public:
void createBridgeReplica(const BridgeStateItemPtr& bridgeState);
- void removeBridge(const BridgeStateItemPtr& bridgeState);
+ void removeBridge(const string& bridgeState);
private:
@@ -561,13 +561,13 @@ void BridgeManagerImpl::createBridgeReplica(const BridgeStateItemPtr& state)
mBridges.push_back(info);
}
-void BridgeManagerImpl::removeBridge(const BridgeStateItemPtr& state)
+void BridgeManagerImpl::removeBridge(const string& bridgeId)
{
mLogger(Trace) << FUNLOG;
boost::unique_lock<boost::shared_mutex> lock(mLock);
for (vector<BridgeInfo>::iterator i = mBridges.begin(); i != mBridges.end(); ++i)
{
- if (i->servant->id() == state->bridgeId)
+ if (i->servant->id() == bridgeId)
{
i->servant->destroyImpl();
mBridges.erase(i);
diff --git a/src/BridgeManagerImpl.h b/src/BridgeManagerImpl.h
index b1c4cc9..e9a8b7e 100644
--- a/src/BridgeManagerImpl.h
+++ b/src/BridgeManagerImpl.h
@@ -56,7 +56,7 @@ public:
virtual void createBridgeReplica(const AsteriskSCF::Replication::BridgeService::V1::BridgeStateItemPtr& bridgeState) = 0;
- virtual void removeBridge(const AsteriskSCF::Replication::BridgeService::V1::BridgeStateItemPtr& bridgeState) = 0;
+ virtual void removeBridge(const std::string& bridgeId) = 0;
};
typedef IceUtil::Handle<BridgeManagerServant> BridgeManagerServantPtr;
diff --git a/src/BridgeReplicatorStateListenerI.cpp b/src/BridgeReplicatorStateListenerI.cpp
index fcbd5bf..7dae181 100644
--- a/src/BridgeReplicatorStateListenerI.cpp
+++ b/src/BridgeReplicatorStateListenerI.cpp
@@ -41,18 +41,35 @@ public:
{
}
+ void removeBridgeItem(const string& bridgeKey, const string& item, const string& sliceType)
+ {
+ IceUtil::Mutex::Lock lock(mMutex);
+ mLogger(Trace) << " removing bridge item id " << item << " (" << sliceType << ") from bridge " << bridgeKey;
+ mBridgeItemMap[bridgeKey].erase(item);
+ if (mBridgeItemMap[bridgeKey].empty())
+ {
+ mManager->removeBridge(bridgeKey);
+ }
+ }
+
void stateRemoved(const Ice::StringSeq& itemKeys, const Ice::Current& current)
{
for (Ice::StringSeq::const_iterator k = itemKeys.begin(); k != itemKeys.end(); ++k)
{
-
- map<string, ReplicatedStateItemPtr>::iterator entry = mItems.find((*k));
- if (entry != mItems.end())
+ ReplicatedStateItemPtr item;
+ {
+ IceUtil::Mutex::Lock lock(mMutex);
+ map<string, ReplicatedStateItemPtr>::iterator entry = mItems.find((*k));
+ if (entry != mItems.end())
+ {
+ item = entry->second;
+ mItems.erase(entry);
+ }
+ }
+ if (item)
{
- ReplicatedStateItemPtr item = entry->second;
mLogger(Trace) << " received removal of " << (*k) << ": a " << item->ice_id();
- mItems.erase(entry);
BridgedSessionPtr bridgedSessionItem = BridgedSessionPtr::dynamicCast(item);
if (bridgedSessionItem)
{
@@ -68,6 +85,7 @@ public:
// Keep the session list clean.
//
found = true;
+ removeBridgeItem(bridgedSessionItem->bridgeId, bridgedSessionItem->key, item->ice_id());
}
//
// We could break here if we could be sure that there were no other updates.
@@ -85,6 +103,7 @@ public:
if ((*b) && (*b)->id() == bridgeListener->bridgeId)
{
(*b)->removeListener(bridgeListener);
+ removeBridgeItem(bridgeListener->bridgeId, bridgeListener->key, item->ice_id());
}
//
// We could break here if we could be sure that there were no other updates.
@@ -100,13 +119,23 @@ public:
{
dumpState(cerr, bridgeItem, current.adapter->getCommunicator());
}
- mManager->removeBridge(bridgeItem);
+ if (mBridgeItemMap[bridgeItem->bridgeId].empty())
+ {
+ mManager->removeBridge(bridgeItem->bridgeId);
+ }
continue;
}
//
// Session pairings are cleaned up by way of sessions going away.
//
+ SessionPairingPtr pairing = SessionPairingPtr::dynamicCast(item);
+ if (pairing)
+ {
+ removeBridgeItem(pairing->bridgeKey, pairing->key, item->ice_id());
+ continue;
+ }
+
///
// The bridge manager isn't really removable.
@@ -121,30 +150,33 @@ public:
{
mLogger(Trace) << " received update " << (*i)->serial << " for " << (*i)->key << " (a " <<
(*i)->ice_id() << ")";
- map<string, ReplicatedStateItemPtr>::iterator entry = mItems.find((*i)->key);
ReplicatedStateItemPtr existingItem;
- if (entry != mItems.end())
{
- //
- // Note: Another serial number situation to consider is two updates of the same key with
- // the same serial number.
- //
+ IceUtil::Mutex::Lock lock(mMutex);
+ map<string, ReplicatedStateItemPtr>::iterator entry = mItems.find((*i)->key);
+ if (entry != mItems.end())
+ {
+ //
+ // Note: Another serial number situation to consider is two updates of the same key with
+ // the same serial number.
+ //
- //
- // Look at serial numbers and indicate an out of sequence update. We should
- // ignore out of sequence updates.
- //
- if ((entry->second->serial > (*i)->serial) && (*i)->serial > SerialCounterStart)
+ //
+ // Look at serial numbers and indicate an out of sequence update. We should
+ // ignore out of sequence updates.
+ //
+ if ((entry->second->serial > (*i)->serial) && (*i)->serial > SerialCounterStart)
+ {
+ mLogger(Error) << "Update serial number for " << (*i)->key << " out of sequence! " <<
+ (*i)->serial << " updating " << entry->second->serial;
+ continue;
+ }
+ existingItem = entry->second;
+ }
+ else
{
- mLogger(Error) << "Update serial number for " << (*i)->key << " out of sequence! " <<
- (*i)->serial << " updating " << entry->second->serial;
- continue;
+ mItems[(*i)->key] = *i;
}
- existingItem = entry->second;
- }
- else
- {
- mItems[(*i)->key] = *i;
}
BridgeManagerStateItemPtr managerItem = BridgeManagerStateItemPtr::dynamicCast((*i));
@@ -187,7 +219,7 @@ public:
{
if (existingItem)
{
- mLogger(Error) << "Replica listener has a bridge object that the bridge manager "
+ mLogger(Warning) << "Replica listener has a bridge object that the bridge manager "
"does not know about. This likely indicates an error and should be investigated.";
}
mManager->createBridgeReplica(bridgeItem);
@@ -210,6 +242,7 @@ public:
// Keep the session list clean.
//
found = true;
+ mBridgeItemMap[bridgedSessionItem->bridgeId][bridgedSessionItem->key] = bridgedSessionItem;
}
//
// We could break here if we could be sure that there were no other updates.
@@ -217,7 +250,8 @@ public:
}
if (!found)
{
- mLogger(Error) << "received an update for a session on a bridge that does not exist!";
+ mLogger(Error) << "received a " << bridgedSessionItem->ice_id() << " (id: " << bridgedSessionItem->key << ")"
+ << " update for a session on a bridge that does not exist: " << bridgedSessionItem->bridgeId;
}
continue;
@@ -241,6 +275,7 @@ public:
// Keep the session list clean.
//
found = true;
+ mBridgeItemMap[sessionPairing->bridgeKey][sessionPairing->key] = sessionPairing;
}
//
// We could break here if we could be sure that there were no other updates.
@@ -248,7 +283,8 @@ public:
}
if (!found)
{
- mLogger(Error) << "received an update for a session on a bridge that does not exist!";
+ mLogger(Error) << "received a " << sessionPairing->ice_id() << " (id: " << sessionPairing->key << ")"
+ << " update for a session on a bridge that does not exist: " << sessionPairing->bridgeKey;
}
continue;
@@ -264,6 +300,7 @@ public:
if ((*b) && (*b)->id() == bridgeListener->bridgeId)
{
(*b)->addListener(bridgeListener);
+ mBridgeItemMap[bridgeListener->bridgeId][bridgeListener->key] = bridgeListener;
}
//
// We could break here if we could be sure that there were no other updates.
@@ -271,21 +308,23 @@ public:
}
if (!found)
{
- mLogger(Error) << "received an update for a session on a bridge that does not exist!";
+ mLogger(Error) << "received a " << bridgeListener->ice_id() << " (id: " << bridgeListener->key << ")"
+ << " update for a session on a bridge that does not exist: " << bridgeListener->bridgeId;
}
continue;
}
- mLogger(Info) << "Bridge replicator service received an unrecognized replication item.";
+ mLogger(Error) << "Bridge replicator service received an unrecognized replication item: " << (*i)->ice_id();
}
}
private:
-
+ IceUtil::Mutex mMutex;
map<string, ReplicatedStateItemPtr> mItems;
Logger mLogger;
BridgeManagerServantPtr mManager;
+ map<string, map<string, ReplicatedStateItemPtr> > mBridgeItemMap;
};
}
diff --git a/src/Component.cpp b/src/Component.cpp
index 331e6af..2557694 100644
--- a/src/Component.cpp
+++ b/src/Component.cpp
@@ -57,7 +57,9 @@ public:
Component() :
AsteriskSCF::Component::Component(lg,
"BridgeService"),
- mListeningToReplicator(false) {}
+ mListeningToReplicator(false)
+ {
+ }
private:
// Optional base Component notification overrides.
@@ -120,6 +122,16 @@ void Component::onActivated()
void Component::onPreInitialize()
{
lg(Debug) << "Launching AsteriskSCF Session-Oriented Bridging Service " << getName();
+ Ice::Int logLevel = getCommunicator()->getProperties()->getPropertyAsIntWithDefault(getName() + ".LogLevel", static_cast<Ice::Int>(Debug));
+ if (logLevel >= 0 && logLevel <= static_cast<Ice::Int>(AsteriskSCF::System::Logging::Off))
+ {
+ lg.setLevel(static_cast<AsteriskSCF::System::Logging::Level>(logLevel));
+ mLogger.setLevel(static_cast<AsteriskSCF::System::Logging::Level>(logLevel));
+ }
+ else
+ {
+ mLogger(Error) << "Configuration attempted to set log level to an invalid value.";
+ }
}
/**
diff --git a/src/MediaSplicer.cpp b/src/MediaSplicer.cpp
index f1a39aa..81dfbad 100755
--- a/src/MediaSplicer.cpp
+++ b/src/MediaSplicer.cpp
@@ -233,7 +233,6 @@ public:
//
void destroy()
{
- SessionPairingPtr newState;
vector<OutgoingPairing> outgoing;
vector<IncomingPairing> incoming;
{
@@ -243,9 +242,14 @@ public:
incoming = mIncoming;
mIncoming.clear();
mConnected = false;
- newState = createUpdate();
}
- pushUpdate(newState);
+ if (mReplicator)
+ {
+ vector<string> keys;
+ keys.push_back(mKey);
+ mReplicator->removeState(keys);
+ mReplicator = AsteriskSCF::BridgeService::ReplicatorSmartPrx();
+ }
mLogger(Trace) << FUNLOG << ": unplugging sinks and sources";
//
diff --git a/src/SessionCollection.cpp b/src/SessionCollection.cpp
index bd3f1ff..9b6c815 100644
--- a/src/SessionCollection.cpp
+++ b/src/SessionCollection.cpp
@@ -202,8 +202,11 @@ void SessionCollection::replicaUpdate(const BridgedSessionPtr& session)
void SessionCollection::removeSession(const BridgedSessionPtr& session)
{
+ SessionWrapperPtr removedSession;
+ {
boost::unique_lock<boost::shared_mutex> lock(mLock);
SessionMap::iterator i = mMap.find(session->key);
+ removedSession = i->second;
if (i != mMap.end())
{
mMap.erase(i);
@@ -214,4 +217,23 @@ void SessionCollection::removeSession(const BridgedSessionPtr& session)
{
mSplicer->disableMixing();
}
+ }
+ if (removedSession)
+ {
+ removedSession->destroy();
+ }
+}
+
+void SessionCollection::destroy()
+{
+ SessionMap copy;
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+ copy = mMap;
+ mMap.clear();
+ }
+ for (SessionMap::iterator i = copy.begin(); i != copy.end();)
+ {
+ i->second->destroy();
+ }
}
diff --git a/src/SessionCollection.h b/src/SessionCollection.h
index d758e85..92460a5 100644
--- a/src/SessionCollection.h
+++ b/src/SessionCollection.h
@@ -132,6 +132,8 @@ public:
void replicaUpdate(const AsteriskSCF::Replication::BridgeService::V1::BridgedSessionPtr& bridgedSession);
void removeSession(const AsteriskSCF::Replication::BridgeService::V1::BridgedSessionPtr& bridgedSession);
+
+ void destroy();
private:
diff --git a/src/SessionListener.cpp b/src/SessionListener.cpp
index c6571e3..21778b2 100644
--- a/src/SessionListener.cpp
+++ b/src/SessionListener.cpp
@@ -217,6 +217,12 @@ public:
mBridgePrx->shutdown();
}
}
+ catch (const Ice::ObjectNotExistException&)
+ {
+ //
+ // The bridge has already gone away... this is okay.
+ //
+ }
catch (const Ice::Exception& ex)
{
mLogger(Error) << "Unexpected exception when initiating auto shutdown: " << ex.what();
diff --git a/src/SessionWrapper.cpp b/src/SessionWrapper.cpp
index 11b59b8..f8dc5df 100644
--- a/src/SessionWrapper.cpp
+++ b/src/SessionWrapper.cpp
@@ -491,6 +491,8 @@ void SessionWrapper::setupMedia()
// basic method. If we need to prevent this from recurring, then it needs
// to be in the caller's logic, not here.
//
+ // TODO: State check
+ //
mConnector = 0;
mSplicer->connect(this, mSession->session);
}
@@ -500,6 +502,9 @@ void SessionWrapper::setConnector(const MediaConnectorPtr& connector)
mLogger(Trace) << FUNLOG << " for " << mId;
{
boost::unique_lock<boost::shared_mutex> lock(mLock);
+ //
+ // TODO: state check?
+ //
mConnector = connector;
}
}
@@ -509,6 +514,9 @@ void SessionWrapper::updateMedia(const SessionPairingPtr& pairings)
mLogger(Trace) << FUNLOG << " for " << mId;
boost::unique_lock<boost::shared_mutex> lock(mLock);
+ //
+ // TODO: shouldn't the state be checked here?
+ //
mSplicer->update(pairings);
if (mConnector)
@@ -567,10 +575,15 @@ void SessionWrapper::destroy()
//
unplugMedia();
mSession->currentState = Done;
- newState = createUpdate();
}
}
- pushUpdate(newState);
+ if (mReplicator)
+ {
+ vector<string> keys;
+ keys.push_back(mSession->key);
+ mReplicator->removeState(keys);
+ mReplicator = AsteriskSCF::BridgeService::ReplicatorSmartPrx();
+ }
}
bool SessionWrapper::isDestroyed()
@@ -644,7 +657,6 @@ bool SessionWrapper::getConnectedLine(ConnectedLinePtr& connectedLine)
return true;
}
-
AsteriskSCF::Replication::BridgeService::V1::BridgedSessionState SessionWrapper::setState(const AsteriskSCF::Replication::BridgeService::V1::BridgedSessionState newState)
{
mLogger(Trace) << FUNLOG << ": updating state " << mId;
diff --git a/src/SessionWrapper.h b/src/SessionWrapper.h
index a03d78b..8bec9f1 100644
--- a/src/SessionWrapper.h
+++ b/src/SessionWrapper.h
@@ -158,7 +158,7 @@ public:
* @return true if the connectedLine parameter was set.
*/
bool getConnectedLine(AsteriskSCF::SessionCommunications::PartyIdentification::V1::ConnectedLinePtr& connectedLine);
-
+
private:
mutable boost::shared_mutex mLock;
diff --git a/test/TestBridging.cpp b/test/TestBridging.cpp
index 6fe199a..68d6199 100644
--- a/test/TestBridging.cpp
+++ b/test/TestBridging.cpp
@@ -1474,6 +1474,10 @@ public:
BOOST_FAIL("Unexpected exception");
}
}
+ catch (const Ice::Exception& ex)
+ {
+ BOOST_FAIL(ex.what());
+ }
catch (...)
{
BOOST_FAIL("Unexpected exception");
commit e5178070c8f1620af13bf68f4f3939de20ed9ab1
Author: Brent Eagles <beagles at digium.com>
Date: Wed Dec 21 12:18:32 2011 -0330
* Moved some logging to Trace level to clean up test output.
* Fixed some catch() blocks in BridgeImpl.cpp that were catching
std::exception by value which slices the exception and removes the actual
exception info. Test output currently indicates that a null handle
exception is being thrown.. which is a little disconcerting.
diff --git a/src/BridgeCreationExtensionPointImpl.cpp b/src/BridgeCreationExtensionPointImpl.cpp
index 6a4ab31..172321f 100755
--- a/src/BridgeCreationExtensionPointImpl.cpp
+++ b/src/BridgeCreationExtensionPointImpl.cpp
@@ -41,7 +41,7 @@ namespace
find_if(mHooks.begin(), mHooks.end(), IdentityComparePredicate<BridgeCreationHookPrx>(newHook));
if (i != mHooks.end())
{
- mLogger(Debug) << "refreshing creation hook " << newHook << " on "
+ mLogger(Trace) << "refreshing creation hook " << newHook << " on "
<< objectIdFromCurrent(current);
//
// Refresh the proxy if it is already in the vector.
@@ -50,7 +50,7 @@ namespace
}
else
{
- mLogger(Debug) << "adding creation hook " << newHook << " on "
+ mLogger(Trace) << "adding creation hook " << newHook << " on "
<< objectIdFromCurrent(current);
mHooks.push_back(newHook);
}
@@ -59,7 +59,7 @@ namespace
void removeHook(const BridgeCreationHookPrx& hookToRemove, const Ice::Current& current)
{
UniqueLock lock(mLock);
- mLogger(Debug) << "removing creation hook " << hookToRemove << " on "
+ mLogger(Trace) << "removing creation hook " << hookToRemove << " on "
<< objectIdFromCurrent(current);
mHooks.erase(remove_if(mHooks.begin(), mHooks.end(),
IdentityComparePredicate<BridgeCreationHookPrx>(hookToRemove)), mHooks.end());
@@ -68,13 +68,13 @@ namespace
void clearHooks(const Ice::Current& current)
{
UniqueLock lock(mLock);
- mLogger(Debug) << "clearing hooks from " << objectIdFromCurrent(current);
+ mLogger(Trace) << "clearing hooks from " << objectIdFromCurrent(current);
mHooks.clear();
}
BridgeCreationHookDataPtr runHooks(const BridgeCreationHookDataPtr& originalData)
{
- mLogger(Debug) << "executing hooks";
+ mLogger(Trace) << "executing hooks";
BridgeCreationHookSeq hooks = getHooks();
BridgeCreationHookSeq deadHooks;
BridgeCreationHookDataPtr tokenData = BridgeCreationHookDataPtr::dynamicCast(originalData->ice_clone());
diff --git a/src/BridgeImpl.cpp b/src/BridgeImpl.cpp
index 3dd19e4..5f2c03a 100755
--- a/src/BridgeImpl.cpp
+++ b/src/BridgeImpl.cpp
@@ -306,7 +306,7 @@ protected:
{
try
{
- mLogger(Debug) << FUNLOG;
+ mLogger(Trace) << FUNLOG;
// Forward the ConnectedLine to each bridged session.
SessionSeq sessions = mBridge->getSessions()->getSessionSeq();
@@ -323,7 +323,7 @@ protected:
}
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -333,7 +333,7 @@ protected:
void forward(const SessionWrapperPtr destinationSession, const RedirectionsPtr& redirections)
{
- mLogger(Debug) << FUNLOG;
+ mLogger(Trace) << FUNLOG;
RedirectionsPtr currentRedirections = redirections;
RedirectionsPtr destSpecificRedirections = redirections;
@@ -403,7 +403,7 @@ protected:
{
try
{
- mLogger(Debug) << FUNLOG;
+ mLogger(Trace) << FUNLOG;
ConnectedLinePtr currentConnectedLine;
SessionWrapperPtr sourceWrapper = mBridge->getSessions()->getSession(mSourceSession);
@@ -430,7 +430,7 @@ protected:
}
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -440,7 +440,7 @@ protected:
void forward(const SessionWrapperPtr destinationSession, const ConnectedLinePtr& connectedLine)
{
- mLogger(Debug) << FUNLOG;
+ mLogger(Trace) << FUNLOG;
ConnectedLinePtr currentConnectedLine = connectedLine;
ConnectedLinePtr destSpecificConnectedLine = connectedLine;
@@ -463,7 +463,7 @@ protected:
currentConnectedLine = destSpecificConnectedLine;
}
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -499,7 +499,7 @@ protected:
{
try
{
- mLogger(Debug) << FUNLOG;
+ mLogger(Trace) << FUNLOG;
// Forward the Caller to each bridged session.
SessionSeq sessions = mBridge->getSessions()->getSessionSeq();
@@ -516,7 +516,7 @@ protected:
}
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -526,7 +526,7 @@ protected:
void forward(const SessionWrapperPtr destinationSession, const CallerPtr& callerID)
{
- mLogger(Debug) << FUNLOG;
+ mLogger(Trace) << FUNLOG;
CallerPtr currentCallerID = callerID;
CallerPtr destSpecificCallerID = callerID;
@@ -549,7 +549,7 @@ protected:
currentCallerID = destSpecificCallerID;
}
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -588,7 +588,7 @@ public:
protected:
bool executeImpl()
{
- mLogger(Debug) << FUNLOG;
+ mLogger(Trace) << FUNLOG;
ConnectedLinePtr currentConnectedLine = mConnectedLine;
ConnectedLinePtr updatedConnectedLine = mConnectedLine;
@@ -611,7 +611,7 @@ protected:
currentConnectedLine = updatedConnectedLine;
}
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -1300,7 +1300,7 @@ void BridgeImpl::updateConnectedLine(const SessionWrapperPtr& sourceSession, con
ExecutorPtr runner(new Executor(tasks, mLogger));
runner->start();
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -1321,7 +1321,7 @@ void BridgeImpl::updateRedirections(const SessionWrapperPtr& sourceSession, cons
ExecutorPtr runner(new Executor(tasks, mLogger));
runner->start();
}
- catch (const std::exception e)
+ catch (const std::exception& e)
{
mLogger(Debug) << FUNLOG << " : " << e.what();
}
@@ -1816,7 +1816,7 @@ void BridgeImpl::activate(const BridgePrx& proxy, const std::string& bridgeId)
mState->key = bridgeId;
mState->bridgeId = bridgeId;
string listenerId = bridgeId + ".sessionListener";
- mLogger(Debug) << FUNLOG << " : activating session listener with " << listenerId;
+ mLogger(Trace) << FUNLOG << " : activating session listener with " << listenerId;
mActivated = true;
mSessionListenerPrx =
SessionListenerPrx::uncheckedCast(
diff --git a/src/BridgeManagerImpl.cpp b/src/BridgeManagerImpl.cpp
index 4ec541d..18fd3da 100644
--- a/src/BridgeManagerImpl.cpp
+++ b/src/BridgeManagerImpl.cpp
@@ -267,7 +267,10 @@ void BridgeManagerImpl::createBridge_async(const AMD_BridgeManager_createBridgeP
listeners = hookData->listeners;
initialSessions = hookData->initialSessions;
info.decoratingPrx = hookData->bridge;
- dump(mLogger, hookData);
+ if (mLogger.isEnabledFor(Trace))
+ {
+ dump(mLogger, hookData);
+ }
}
//
diff --git a/src/BridgeReplicatorStateListenerI.cpp b/src/BridgeReplicatorStateListenerI.cpp
index ecdbf20..fcbd5bf 100644
--- a/src/BridgeReplicatorStateListenerI.cpp
+++ b/src/BridgeReplicatorStateListenerI.cpp
@@ -50,7 +50,7 @@ public:
if (entry != mItems.end())
{
ReplicatedStateItemPtr item = entry->second;
- mLogger(Debug) << " received removal of " << (*k) << ": a " << item->ice_id();
+ mLogger(Trace) << " received removal of " << (*k) << ": a " << item->ice_id();
mItems.erase(entry);
BridgedSessionPtr bridgedSessionItem = BridgedSessionPtr::dynamicCast(item);
@@ -96,7 +96,10 @@ public:
BridgeStateItemPtr bridgeItem = BridgeStateItemPtr::dynamicCast(item);
if (bridgeItem)
{
- dumpState(cerr, bridgeItem, current.adapter->getCommunicator());
+ if (mLogger.isEnabledFor(Trace))
+ {
+ dumpState(cerr, bridgeItem, current.adapter->getCommunicator());
+ }
mManager->removeBridge(bridgeItem);
continue;
}
@@ -161,7 +164,10 @@ public:
BridgeStateItemPtr bridgeItem = BridgeStateItemPtr::dynamicCast((*i));
if (bridgeItem)
{
- dumpState(cerr, bridgeItem, current.adapter->getCommunicator());
+ if (mLogger.isEnabledFor(Trace))
+ {
+ dumpState(cerr, bridgeItem, current.adapter->getCommunicator());
+ }
vector<BridgeServantPtr> bridges = mManager->getBridges();
bool found = false;
for (vector<BridgeServantPtr>::iterator b = bridges.begin(); b != bridges.end(); ++b)
diff --git a/src/MediaSplicer.cpp b/src/MediaSplicer.cpp
index 27411cc..f1a39aa 100755
--- a/src/MediaSplicer.cpp
+++ b/src/MediaSplicer.cpp
@@ -588,13 +588,13 @@ public:
if (existingMixer == mixers.end())
{
- mLogger(Debug) << FUNLOG << ": creating media mixer for format " << format;
+ mLogger(Trace) << FUNLOG << ": creating media mixer for format " << format;
mixer = new MediaMixer(mAdapter, stream->second->formats.front());
mMediaMixers.insert(make_pair(format, mixer));
}
else
{
- mLogger(Debug) << FUNLOG << ": using found media mixer for format " << format;
+ mLogger(Trace) << FUNLOG << ": using found media mixer for format " << format;
mixer = existingMixer->second;
mixers.erase(format);
}
diff --git a/src/SessionListener.cpp b/src/SessionListener.cpp
index 9062ca6..c6571e3 100644
--- a/src/SessionListener.cpp
+++ b/src/SessionListener.cpp
@@ -124,7 +124,7 @@ public:
if ((connected = AsteriskSCF::SessionCommunications::V1::ConnectedIndicationPtr::dynamicCast(indication)))
{
string proxyString = source->ice_toString();
- mLogger(Debug) << FUNLOG << ": session connected " << proxyString;
+ mLogger(Trace) << FUNLOG << ": session connected " << proxyString;
mSessions->reap();
try
{
@@ -150,7 +150,7 @@ public:
else if ((stopped = AsteriskSCF::SessionCommunications::V1::StoppedIndicationPtr::dynamicCast(indication)))
{
string proxyString = source->ice_toString();
- mLogger(Debug) << FUNLOG << ": session stopped " << proxyString;
+ mLogger(Trace) << FUNLOG << ": session stopped " << proxyString;
//
// IMPLNOTE: The AMI approach.
commit 33c6a09cc7f2605fb3512e8bc0fb236d42e7e72c
Merge: 1751d60 f3fc03f
Author: Mark Michelson <mmichelson at digium.com>
Date: Mon Dec 19 16:11:39 2011 -0600
Merge branch 'master' of git.asterisk.org:asterisk-scf/release/bridging
commit f3fc03f25cf923dd3e1f264bd2f9797f5fa08b4a
Author: Brent Eagles <beagles at digium.com>
Date: Tue Dec 6 16:40:52 2011 -0330
Convert all unit tests to use loopback address instead of all adapters.
diff --git a/config/test_bridging.conf b/config/test_bridging.conf
index a2c794d..5ec14f2 100644
--- a/config/test_bridging.conf
+++ b/config/test_bridging.conf
@@ -25,19 +25,19 @@ Ice.ThreadPool.Client.Size=4
ServiceDiscovery.IceStorm.TopicManager.Endpoints=default -p 4421
ServiceDiscovery.IceStorm.TopicManager.ThreadPool.Size=4
ServiceDiscovery.IceStorm.InstanceName=ServiceDiscovery
-ServiceDiscovery.IceStorm.Publish.Endpoints=tcp -p 10001:udp -p 10001
+ServiceDiscovery.IceStorm.Publish.Endpoints=tcp -h 127.0.0.1 -p 10001:udp -h 127.0.0.1 -p 10001
ServiceDiscovery.IceStorm.Publish.ThreadPool.Size=4
ServiceDiscovery.IceStorm.Transient=1
ServiceDiscovery.IceStorm.Trace.TopicManager=0
ServiceDiscovery.IceStorm.Flush.Timeout=2000
-TopicManager.Proxy=ServiceDiscovery/TopicManager:default -p 4421
+TopicManager.Proxy=ServiceDiscovery/TopicManager:default -h 127.0.0.1 -p 4421
-ServiceDiscovery.BackplaneAdapter.Endpoints=tcp -p 4410
-ServiceDiscovery.Locator.ServiceAdapter.Endpoints=tcp -p 4411
-ServiceDiscovery.Management.ServiceAdapter.Endpoints=tcp -p 4422
+ServiceDiscovery.BackplaneAdapter.Endpoints=tcp -h 127.0.0.1 -p 4410
+ServiceDiscovery.Locator.ServiceAdapter.Endpoints=tcp -h 127.0.0.1 -p 4411
+ServiceDiscovery.Management.ServiceAdapter.Endpoints=tcp -h 127.0.0.1 -p 4422
-LocatorService.Proxy=LocatorService:tcp -p 4411
-LocatorServiceManagement.Proxy=LocatorServiceManagement:tcp -p 4422
+LocatorService.Proxy=LocatorService:tcp -h 127.0.0.1 -p 4411
+LocatorServiceManagement.Proxy=LocatorServiceManagement:tcp -h 127.0.0.1 -p 4422
LoggerAdapter.Endpoints=default
Logger.ServiceAdapter.Endpoints=default
@@ -66,6 +66,7 @@ IceBox.Service.TestDriver=bridge_component_test:create
# service discovery in a later branch.
#
TestChannel.InstanceName=BridgeTest
+TestChannel.TestChannel.Endpoints=default -h 127.0.0.1
#
# Configuration for the test bridge instances. There are are two: one master
@@ -76,18 +77,18 @@ TestChannel.InstanceName=BridgeTest
#
TestBridge.InstanceName=TestBridge
TestBridge.BridgeManagerObjectId=TestBridgeManager
-TestBridge.ServiceAdapter.Endpoints=default -p 57000
-TestBridge.ServiceAdapter.ThreadPool.Size=4
-TestBridge.BackplaneAdapter.Endpoints=default -p 57001
-TestBridge.BackplaneAdapter.ThreadPool.Size=4
+TestBridge.ServiceAdapter.Endpoints=default -h 127.0.0.1 -p 57000
+TestBridge.ServiceAdapter.ThreadPool.Size=2
+TestBridge.BackplaneAdapter.Endpoints=default -h 127.0.0.1 -p 57001
+TestBridge.BackplaneAdapter.ThreadPool.Size=2
TestBridge.Standby=false
TestBridge2.InstanceName=TestBridge2
TestBridge2.BridgeManagerObjectId=TestBridgeManager2
TestBridge2.Standby=true
-TestBridge2.ServiceAdapter.Endpoints=default -p 57010
+TestBridge2.ServiceAdapter.Endpoints=default -h 127.0.0.1 -p 57010
TestBridge2.ServiceAdapter.ThreadPool.Size=4
-TestBridge2.BackplaneAdapter.Endpoints=default -p 57011
+TestBridge2.BackplaneAdapter.Endpoints=default -h 127.0.0.1 -p 57011
TestBridge2.BackplaneAdapter.ThreadPool.Size=4
#
@@ -111,6 +112,6 @@ IceBox.LoadOrder=ServiceDiscovery Logger Replicator TestBridge TestBridge2 TestC
# Configuring a manager endpoint for the IceBox service allows services to be managed and
# monitored.
#
-IceBox.ServiceManager.Endpoints=default -p 56000
+IceBox.ServiceManager.Endpoints=default -h 127.0.0.1 -p 56000
IceBox.ServiceManager.ThreadPool.Size=4
-IceBoxMgr.Proxy=IceBox/ServiceManager:default -p 56000
+IceBoxMgr.Proxy=IceBox/ServiceManager:default -h 127.0.0.1 -p 56000
diff --git a/test/TestBridging.cpp b/test/TestBridging.cpp
index 0476241..7ae58f4 100644
--- a/test/TestBridging.cpp
+++ b/test/TestBridging.cpp
@@ -221,7 +221,7 @@ public:
Ice::PropertiesPtr props = mCommunicator->getProperties();
propGetSet(props, "TestUtilAdapter.ThreadPool.Size", "4");
propGetSet(props, "TestUtilAdapter.ThreadPool.Size", "4");
- propGetSet(props, "TestUtilAdapter.Endpoints", "default");
+ propGetSet(props, "TestUtilAdapter.Endpoints", "default -h 127.0.0.1");
lookupProxies();
}
commit 7a865ca109654e97c88c15ed4e6bebf0f3b6a1b9
Author: Joshua Colp <jcolp at digium.com>
Date: Sun Dec 4 18:52:08 2011 -0400
Add support for mixing streams from multiple sessions within a bridge. (issue ASTSCF-2)
diff --git a/src/BridgeImpl.cpp b/src/BridgeImpl.cpp
index 0c94a05..515d710 100755
--- a/src/BridgeImpl.cpp
+++ b/src/BridgeImpl.cpp
@@ -1152,7 +1152,7 @@ BridgeImpl::BridgeImpl(const string& name, const Ice::ObjectAdapterPtr& adapter,
const Logger& logger) :
mActivated(false),
mState(state),
- mSessions(new SessionCollection(adapter->getCommunicator(), name, replicator, logger)),
+ mSessions(new SessionCollection(adapter->getCommunicator(), name, replicator, logger, adapter)),
mName(name),
mObjAdapter(adapter),
mListeners(listenerMgr),
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 4e20b5c..1b718a9 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -30,6 +30,8 @@ astscf_component_add_files(bridgeservice ServiceUtil.h)
astscf_component_add_files(bridgeservice DebugUtil.h)
astscf_component_add_files(bridgeservice MediaSplicer.h)
astscf_component_add_files(bridgeservice MediaSplicer.cpp)
+astscf_component_add_files(bridgeservice MediaMixer.h)
+astscf_component_add_files(bridgeservice MediaMixer.cpp)
astscf_component_add_files(bridgeservice Tasks.h)
astscf_component_add_files(bridgeservice InternalExceptions.h)
astscf_component_add_files(bridgeservice BridgeServiceConfig.h)
diff --git a/src/MediaMixer.cpp b/src/MediaMixer.cpp
new file mode 100755
index 0000000..4237041
--- /dev/null
+++ b/src/MediaMixer.cpp
@@ -0,0 +1,566 @@
+/*
+ * Asterisk SCF -- An open-source communications framework.
+ *
+ * Copyright (C) 2010-2011, Digium, Inc.
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk SCF project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE.txt file
+ * at the top of the source tree.
+ */
+#include "MediaMixer.h"
+
+#include <AsteriskSCF/Media/Formats/AudioFormats.h>
+
+#include <Ice/Ice.h>
+#include <IceUtil/Shared.h>
+#include <IceUtil/Handle.h>
+
+using namespace AsteriskSCF::System::Logging;
+using namespace AsteriskSCF::Media::V1;
+using namespace AsteriskSCF::Media::Formats::Audio::V1;
+using namespace std;
+
+namespace AsteriskSCF
+{
+namespace BridgeService
+{
+
+/**
+ * Interval (in milliseconds) that media is mixed in.
+ */
+#define MIXING_INTERVAL 20
+
+/**
+ * Implementation of a StreamSource which provides mixed audio.
+ */
+class MediaMixerSource : public StreamSource
+{
+public:
+ MediaMixerSource(const MediaMixerPtr& mixer) : mMixer(mixer) { }
+
+ void addSink(const StreamSinkPrx& sink, const Ice::Current&)
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+
+ if (std::find(mSinks.begin(), mSinks.end(), sink) != mSinks.end())
+ {
+ return;
+ }
+
+ mSinks.push_back(sink);
+ }
+
+ void removeSink(const StreamSinkPrx& sink, const Ice::Current&)
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+
+ mSinks.erase(std::remove(mSinks.begin(), mSinks.end(), sink), mSinks.end());
+
+ // If no more sinks exist then this should terminate
+ if (mSinks.empty())
+ {
+ mMixer->removeSource(this, mProxy);
+ }
+ }
+
+ StreamSinkSeq getSinks(const Ice::Current&)
+ {
+ boost::shared_lock<boost::shared_mutex> lock(mLock);
+
+ return mSinks;
+ }
+
+ FormatSeq getFormats(const Ice::Current&)
+ {
+ // This is not protected by the lock because once created this will never be altered
+ return mFormats;
+ }
+
+ std::string getId(const Ice::Current&)
+ {
+ // Same goes for the id
+ return mProxy->ice_getIdentity().name;
+ }
+
+ void requestFormat(const AsteriskSCF::Media::V1::FormatPtr&, const Ice::Current&)
+ {
+ throw MediaFormatSwitchException();
+ }
+
+ /**
+ * Internal function which sets our proxy.
+ */
+ void setProxy(const StreamSourcePrx& proxy)
+ {
+ mProxy = proxy;
+ }
+
+ /**
+ * Internal function which gets our proxy.
+ */
+ StreamSourcePrx getProxy()
+ {
+ return mProxy;
+ }
+
+private:
+ /**
+ * Lock which protects this source.
+ */
+ boost::shared_mutex mLock;
+
+ /**
+ * Media mixer we are associated with.
+ */
+ MediaMixerPtr mMixer;
+
+ /**
+ * Proxy to ourselves.
+ */
+ StreamSourcePrx mProxy;
+
+ /**
+ * Sinks that we are writing media to.
+ */
+ StreamSinkSeq mSinks;
+
+ /**
+ * Formats supported by this source.
+ */
+ FormatSeq mFormats;
+};
+
+/**
+ * Implementation of a StreamSink which provides audio to be mixed into the bridge.
+ */
+class MediaMixerSink : public StreamSink
+{
+public:
+ MediaMixerSink(const MediaMixerPtr& mixer, const MediaMixerSourcePtr& source) :
+ mMixer(mixer), mLocalSource(source) { }
+
+ void write(const FrameSeq& frames, const Ice::Current&)
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+
+ mFrames.insert(mFrames.begin(), frames.begin(), frames.end());
+ }
+
+ void setSource(const StreamSourcePrx& source, const Ice::Current&)
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+
+ if (source == 0)
+ {
+ mMixer->removeSink(this, mProxy);
+ }
+ else
+ {
+ mSource = source;
+ }
+ }
+
+ StreamSourcePrx getSource(const Ice::Current&)
+ {
+ boost::shared_lock<boost::shared_mutex> lock(mLock);
+
+ return mSource;
+ }
+
+ FormatSeq getFormats(const Ice::Current&)
+ {
+ return mFormats;
+ }
+
+ std::string getId(const Ice::Current&)
+ {
+ return mProxy->ice_getIdentity().name;
+ }
+
+ /**
+ * Internal function which returns the latest frame to use in the mixer.
+ */
+ const FramePtr& getFrame()
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+
+ if (!mFrames.empty())
+ {
+ mProvidedFrame = mFrames.back();
+ mFrames.pop_back();
+ }
+
+ return mProvidedFrame;
+ }
+
+ /**
+ * Internal function which removes the latest frame, presumably having already been mixed.
+ */
+ const FramePtr popFrame()
+ {
+ boost::unique_lock<boost::shared_mutex> lock(mLock);
+
+ FramePtr frame = mProvidedFrame;
+ mProvidedFrame = 0;
+
+ return frame;
+ }
+
+ /**
+ * Internal function which sets our proxy.
+ */
+ void setProxy(const StreamSinkPrx& proxy)
+ {
+ mProxy = proxy;
+ }
+
+ /**
+ * Internal function which returns our proxy.
+ */
+ StreamSinkPrx getProxy()
+ {
+ return mProxy;
+ }
+
+ /**
+ * Internal function which returns the local source.
+ */
+ MediaMixerSourcePtr getLocalSource()
+ {
+ return mLocalSource;
+ }
+
+private:
+ /**
+ * Lock which protects this sink.
+ */
+ boost::shared_mutex mLock;
+
+ /**
+ * Media mixer we are associated with.
+ */
+ MediaMixerPtr mMixer;
+
+ /**
+ * Pointer to our source of media, this exists within this class because the two
+ * are inherently associated.
+ */
+ MediaMixerSourcePtr mLocalSource;
+
+ /**
+ * Proxy to ourselves.
+ */
+ StreamSinkPrx mProxy;
+
+ /**
+ * Sequence of received frames pending mixing.
+ */
+ FrameSeq mFrames;
+
+ /**
+ * Source of media for this sink.
+ */
+ StreamSourcePrx mSource;
+
+ /**
+ * Formats supported by this sink.
+ */
+ FormatSeq mFormats;
+
+ /**
+ * Frame we provided when getFrame was called.
+ */
+ FramePtr mProvidedFrame;
+};
+
+/**
+ * Simple callback object used when writing to sinks.
+ */
+class WriteCallback : public IceUtil::Shared
+{
+public:
+ void writtenCB()
+ {
+ }
+
+ void failureCB(const Ice::Exception&)
+ {
+ }
+};
+
+/**
+ * Smart pointer for the above WriteCallback class.
+ */
+typedef IceUtil::Handle<WriteCallback> WriteCallbackPtr;
+
+class MediaMixerI
+{
+public:
+ MediaMixerI(const Ice::ObjectAdapterPtr& adapter, const FormatPtr& format) : mAdapter(adapter), mFormat(format),
+ mTimer(new IceUtil::Timer()),
+ mCallback(new WriteCallback()) { }
+
+ /**
+ * Lock which protects the mixer.
+ */
+ boost::shared_mutex mLock;
+
+ /**
+ * Object adapter that sinks/sources are on.
+ */
+ Ice::ObjectAdapterPtr mAdapter;
+
+ /**
+ * Format for produced frames.
+ */
+ FormatPtr mFormat;
+
+ /**
+ * Vector of sinks for media.
+ */
+ std::vector<MediaMixerSinkPtr> mSinks;
+
+ /**
+ * Vector of sources for media.
+ */
+ std::vector<MediaMixerSourcePtr> mSources;
+
+ /**
+ * Timer thread used to mix the media.
+ */
+ IceUtil::TimerPtr mTimer;
+
+ /**
+ * Callback object for when writing to sinks.
+ */
+ WriteCallbackPtr mCallback;
+};
+
+/**
+ * Transform function which performs a signed linear saturated add.
+ */
+static short saturatedSignedLinearAdd(short input, short value)
+{
+ int res = input + value;
+
+ if (res > 32767)
+ {
+ return 32767;
+ }
+ else if (res < -32767)
+ {
+ return -32767;
+ }
+ else
+ {
+ return static_cast<short>(res);
+ }
+}
+
+/**
+ * Transform function which performs a signed linear saturated subtract.
+ */
+static short saturatedSignedLinearSubtract(short input, short value)
+{
+ int res = input - value;
+
+ if (res > 32767)
+ {
+ return 32767;
+ }
+ else if (res < -32767)
+ {
+ return -32767;
+ }
+ else
+ {
+ return static_cast<short>(res);
+ }
+}
+
+MediaMixer::MediaMixer(const Ice::ObjectAdapterPtr& adapter, const FormatPtr& format) : mImpl(new MediaMixerI(adapter, format))
+{
+ mImpl->mTimer->schedule(this, IceUtil::Time::milliSeconds(MIXING_INTERVAL));
+}
+
+void MediaMixer::createMixing(const Ice::Identity& sinkId, AsteriskSCF::Media::V1::StreamSinkPrx& mixerSink,
+ const Ice::Identity& sourceId, AsteriskSCF::Media::V1::StreamSourcePrx& mixerSource)
+{
+ boost::unique_lock<boost::shared_mutex> lock(mImpl->mLock);
+
+ // The source is where mixed audio will originate from, within this implementation it essentially acts
+ // as a store of sinks we need to write to for this source
+ Ice::Identity id = sourceId;
+ id.name += ".mixer";
+ MediaMixerSourcePtr source;
+ if (mImpl->mAdapter->find(id))
+ {
+ mixerSource = StreamSourcePrx::uncheckedCast(mImpl->mAdapter->createProxy(id));
+ }
+ else
+ {
+ source = new MediaMixerSource(this);
+ mImpl->mSources.push_back(source);
+ mixerSource = StreamSourcePrx::uncheckedCast(mImpl->mAdapter->add(source, id));
+ source->setProxy(mixerSource);
+ }
+
+ // The sink is where the source will send audio to to be mixed into the bridge, the reason the local source
+ // is associated is because we need to know what media to remove from the mixed frame when writing out
+ id = sinkId;
+ id.name += ".mixer";
+ if (mImpl->mAdapter->find(id))
+ {
+ mixerSink = StreamSinkPrx::uncheckedCast(mImpl->mAdapter->createProxy(id));
+ }
+ else
+ {
+ MediaMixerSinkPtr sink = new MediaMixerSink(this, source);
+ mImpl->mSinks.push_back(sink);
+ mixerSink = StreamSinkPrx::uncheckedCast(mImpl->mAdapter->add(sink, id));
+ sink->setProxy(mixerSink);
+ }
+}
+
+void MediaMixer::runTimerTask()
+{
+ IceUtil::Time start = IceUtil::Time::now();
+
+ std::vector<MediaMixerSinkPtr> sinks;
+
+ {
+ boost::shared_lock<boost::shared_mutex> lock(mImpl->mLock);
+ sinks = mImpl->mSinks;
+ }
+
+ FramePtr frame = new Frame();
+ frame->mediaFormat = mImpl->mFormat;
+
+ ShortSeqPayloadPtr mixedPayload = new ShortSeqPayload();
+ frame->payload = mixedPayload;
+
+ for (std::vector<MediaMixerSinkPtr>::const_iterator sink = sinks.begin();
+ sink != sinks.end();
+ ++sink)
+ {
+ FramePtr sinkFrame = (*sink)->getFrame();
+
+ // If no frame exists for us to mix in skip doing so
+ if (!sinkFrame)
+ {
+ continue;
+ }
+
+ ShortSeqPayloadPtr payload = ShortSeqPayloadPtr::dynamicCast(sinkFrame->payload);
+
+ // We only currently support short sequence payloads
+ if (!payload)
+ {
+ continue;
+ }
+
+ // If no audio exists within the mixedPayload yet just copy this frame in, otherwise mix it in
+ if (!mixedPayload->payload.size())
+ {
+ mixedPayload->payload.resize(payload->payload.size());
+ std::copy(payload->payload.begin(), payload->payload.end(), mixedPayload->payload.begin());
+ }
+ else
+ {
+ std::transform(mixedPayload->payload.begin(), mixedPayload->payload.end(),
+ payload->payload.begin(), mixedPayload->payload.begin(),
+ saturatedSignedLinearAdd);
+ }
+ }
+
+ for (std::vector<MediaMixerSinkPtr>::const_iterator sink = sinks.begin();
+ sink != sinks.end();
+ ++sink)
+ {
+ FramePtr sinkFrame = (*sink)->popFrame();
+ FramePtr theirFrame(frame);
+
+ if (sinkFrame)
+ {
+ theirFrame = FramePtr::dynamicCast(frame->ice_clone());
+
+ ShortSeqPayloadPtr sinkPayload = ShortSeqPayloadPtr::dynamicCast(sinkFrame->payload);
+
+ // Remove the participants own media so they do not hear themselves
+ ShortSeqPayloadPtr payload = new ShortSeqPayload();
+ payload->payload.resize(mixedPayload->payload.size());
+ std::transform(mixedPayload->payload.begin(), mixedPayload->payload.end(),
+ sinkPayload->payload.begin(), payload->payload.begin(),
+ saturatedSignedLinearSubtract);
+ theirFrame->payload = payload;
+ }
+
+ // Send this frame to all the sinks we should
+ StreamSinkSeq sourceSinks = (*sink)->getLocalSource()->getSinks(Ice::Current());
+
+ FrameSeq frames;
+ frames.push_back(theirFrame);
+
+ for (StreamSinkSeq::const_iterator sourceSink = sourceSinks.begin();
+ sourceSink != sourceSinks.end();
+ ++sourceSink)
+ {
+ (*sourceSink)->begin_write(frames, newCallback_StreamSink_write(mImpl->mCallback,
+ &WriteCallback::writtenCB,
+ &WriteCallback::failureCB));
+ }
+ }
+
+ IceUtil::Time elapsed = IceUtil::Time::now() - start;
+ mImpl->mTimer->schedule(this, IceUtil::Time::milliSeconds(MIXING_INTERVAL - elapsed.toMilliSeconds()));
+}
+
+void MediaMixer::removeSource(const MediaMixerSourcePtr& source, const StreamSourcePrx& proxy)
+{
+ boost::unique_lock<boost::shared_mutex> lock(mImpl->mLock);
+
+ mImpl->mSources.erase(std::remove(mImpl->mSources.begin(), mImpl->mSources.end(), source), mImpl->mSources.end());
+ mImpl->mAdapter->remove(proxy->ice_getIdentity());
+}
+
+void MediaMixer::removeSink(const MediaMixerSinkPtr& sink, const StreamSinkPrx& proxy)
+{
+ boost::unique_lock<boost::shared_mutex> lock(mImpl->mLock);
+
+ mImpl->mSinks.erase(std::remove(mImpl->mSinks.begin(), mImpl->mSinks.end(), sink), mImpl->mSinks.end());
+ mImpl->mAdapter->remove(proxy->ice_getIdentity());
+}
+
+void MediaMixer::stop()
+{
+ boost::unique_lock<boost::shared_mutex> lock(mImpl->mLock);
+
+ mImpl->mTimer->destroy();
+
+ for (std::vector<MediaMixerSinkPtr>::const_iterator sink = mImpl->mSinks.begin();
+ sink != mImpl->mSinks.end();
+ ++sink)
+ {
+ mImpl->mAdapter->remove((*sink)->getProxy()->ice_getIdentity());
+ }
+
+ mImpl->mSinks.clear();
+
+ for (std::vector<MediaMixerSourcePtr>::const_iterator source = mImpl->mSources.begin();
+ source != mImpl->mSources.end();
+ ++source)
+ {
+ mImpl->mAdapter->remove((*source)->getProxy()->ice_getIdentity());
+ }
+
+ mImpl->mSources.clear();
+}
+
+}
+}
diff --git a/src/MediaMixer.h b/src/MediaMixer.h
new file mode 100644
index 0000000..d167e25
--- /dev/null
+++ b/src/MediaMixer.h
@@ -0,0 +1,76 @@
+/*
+ * Asterisk SCF -- An open-source communications framework.
+ *
+ * Copyright (C) 2011, Digium, Inc.
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk SCF project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE.txt file
+ * at the top of the source tree.
+ */
+#pragma once
+
+#include <IceUtil/Shared.h>
+#include <IceUtil/Timer.h>
+#include <AsteriskSCF/Media/MediaIf.h>
+#include <AsteriskSCF/logger.h>
+#include <boost/shared_ptr.hpp>
+
+namespace AsteriskSCF
+{
+namespace BridgeService
+{
+
+/**
+ * Forward declarations.
+ */
+class MediaMixerSource;
+class MediaMixerSink;
+
+/**
+ * Smart pointer type for MediaMixerSource class.
+ */
+typedef IceUtil::Handle<MediaMixerSource> MediaMixerSourcePtr;
+
+/**
+ * Smart pointer type for MediaMixerSink class.
+ */
+typedef IceUtil::Handle<MediaMixerSink> MediaMixerSinkPtr;
+
+/**
+ * Implementation details for MediaMixer class.
+ */
+class MediaMixerI;
+class MediaMixer : public IceUtil::TimerTask
+{
+public:
+ MediaMixer(const Ice::ObjectAdapterPtr&, const AsteriskSCF::Media::V1::FormatPtr&);
+
... 2031 lines suppressed ...
--
asterisk-scf/integration/bridging.git
More information about the asterisk-scf-commits
mailing list