[asterisk-scf-commits] asterisk-scf/integration/media_rtp_pjmedia.git branch "resampling" created.
Commits to the Asterisk SCF project code repositories
asterisk-scf-commits at lists.digium.com
Mon Aug 22 16:52:00 CDT 2011
branch "resampling" has been created
at 9a520629a795d2d952fdd209f9662e92df784a9e (commit)
- Log -----------------------------------------------------------------
commit 9a520629a795d2d952fdd209f9662e92df784a9e
Author: Mark Michelson <mmichelson at digium.com>
Date: Mon Aug 22 16:49:45 2011 -0500
Adjust the RTP test to accommodate the need for FormatOperationsService.
diff --git a/test/TestRTPpjmedia.cpp b/test/TestRTPpjmedia.cpp
index 24a6500..1f75514 100644
--- a/test/TestRTPpjmedia.cpp
+++ b/test/TestRTPpjmedia.cpp
@@ -179,6 +179,11 @@ public:
* Condition used to signal test thread that state was set.
*/
boost::condition mCondition;
+
+ /**
+ * Operations for our test format. Necessary for encoding/decoding frames
+ */
+ FormatOperationsServicePrx mFormatOperationsService;
};
static SharedTestData Testbed;
@@ -288,6 +293,40 @@ private:
StreamSourcePrx mSource;
};
+class TestFormatOperations : public FormatOperationsService
+{
+ bool checkCompatible(
+ const FormatPtr& format1,
+ const FormatPtr& format2,
+ const Ice::Current&)
+ {
+ return format1 == format2;
+ }
+
+ FramePayloadPtr decodePayload(
+ const Ice::ByteSeq& toDecode,
+ const Ice::Current&)
+ {
+ Ice::ByteSeq bytePayload;
+ std::copy(toDecode.begin(), toDecode.end(), std::back_inserter(bytePayload));
+ return new ByteSeqPayload(bytePayload);
+ }
+
+ Ice::ByteSeq encodePayload(
+ const FramePayloadPtr& toEncode,
+ const Ice::Current&)
+ {
+ ByteSeqPayloadPtr bytePayload = ByteSeqPayloadPtr::dynamicCast(toEncode);
+
+ if (!bytePayload)
+ {
+ return Ice::ByteSeq();
+ }
+
+ return bytePayload->payload;
+ }
+};
+
/**
* A global fixture for Ice initialization.
* Provides setup/teardown for the entire set of tests.
@@ -321,6 +360,8 @@ struct GlobalIceFixture
Testbed.mInformationListenerProxy = RTCP::V1::InformationListenerPrx::uncheckedCast(Testbed.adapter->addWithUUID(
Testbed.mInformationListener));
+ Testbed.mFormatOperationsService = FormatOperationsServicePrx::uncheckedCast(Testbed.adapter->addWithUUID(new TestFormatOperations));
+
Testbed.adapter->activate();
Testbed.locator = ServiceLocatorPrx::checkedCast(Testbed.communicator->stringToProxy("LocatorService:tcp -p 4411"));
@@ -477,6 +518,7 @@ BOOST_AUTO_TEST_CASE(AllocateRTPSession)
AudioFormatPtr format = new AudioFormat();
format->name = "test_format";
format->frameSize = 20;
+ format->operations = Testbed.mFormatOperationsService;
FormatSeq formats;
formats.push_back(format);
@@ -749,6 +791,7 @@ BOOST_AUTO_TEST_CASE(TransmitFrametoEmptySink)
AudioFormatPtr format = new AudioFormat();
format->name = "test_format";
format->frameSize = 20;
+ format->operations = Testbed.mFormatOperationsService;
AudioFramePtr frame = new AudioFrame();
frame->mediaFormat = format;
@@ -879,6 +922,7 @@ BOOST_AUTO_TEST_CASE(PushPayloadMappings)
AudioFormatPtr format = new AudioFormat();
format->name = "test_format";
format->frameSize = 20;
+ format->operations = Testbed.mFormatOperationsService;
/* I'm just going to go ahead and use payload 98 for it since
* that is dynamic.
@@ -1069,6 +1113,7 @@ BOOST_AUTO_TEST_CASE(TransmitandReceiveFrame)
AudioFormatPtr format = new AudioFormat();
format->name = "test_format";
format->frameSize = 20;
+ format->operations = Testbed.mFormatOperationsService;
AudioFramePtr frame = new AudioFrame();
frame->mediaFormat = format;
@@ -1109,9 +1154,13 @@ BOOST_AUTO_TEST_CASE(TransmitandReceiveFrame)
(received_frame->mediaFormat->name == format->name))
{
AudioFormatPtr received_format;
+ ByteSeqPayloadPtr originalPayload;
+ ByteSeqPayloadPtr receivedPayload;
if ((received_format = AudioFormatPtr::dynamicCast(received_frame->mediaFormat)) &&
(received_format->frameSize == format->frameSize) &&
- (received_frame->payload == frame->payload))
+ (originalPayload = ByteSeqPayloadPtr::dynamicCast(frame->payload)) &&
+ (receivedPayload = ByteSeqPayloadPtr::dynamicCast(received_frame->payload)) &&
+ (receivedPayload->payload == originalPayload->payload))
{
received = true;
}
@@ -1141,6 +1190,7 @@ BOOST_AUTO_TEST_CASE(TransmitFrameWithUnsupportedMediaFormat)
AudioFormatPtr format = new AudioFormat();
format->name = "tacos";
format->frameSize = 20;
+ format->operations = Testbed.mFormatOperationsService;
AudioFramePtr frame = new AudioFrame();
frame->mediaFormat = format;
@@ -1203,6 +1253,7 @@ BOOST_AUTO_TEST_CASE(ReceiveUnknownRTPPacket)
AudioFormatPtr format = new AudioFormat();
format->name = "zombies";
format->frameSize = 20;
+ format->operations = Testbed.mFormatOperationsService;
FormatSeq formats;
formats.push_back(format);
commit a543a39849ddd8588c4e5d42e53e6607ba802f24
Author: Mark Michelson <mmichelson at digium.com>
Date: Mon Aug 22 15:24:43 2011 -0500
Use format operation services to encode and decode frame payloads.
diff --git a/src/RTPSink.cpp b/src/RTPSink.cpp
index 687ded0..48d2bc6 100644
--- a/src/RTPSink.cpp
+++ b/src/RTPSink.cpp
@@ -125,10 +125,12 @@ void StreamSinkRTPImpl::write(const AsteriskSCF::Media::V1::FrameSeq& frames, co
throw UnsupportedMediaFormatException();
}
+ Ice::ByteSeq framePayload = (*frame)->mediaFormat->operations->encodePayload((*frame)->payload);
+
/* Using the available information construct an RTP header that we can place at the front of our packet */
pj_status_t status = pjmedia_rtp_encode_rtp(&mImpl->mOutgoingSession,
- payload, 0, (int) (*frame)->payload.size(),
- (int) (*frame)->payload.size(), &header, &header_len);
+ payload, 0, (int) framePayload.size(),
+ (int) framePayload.size(), &header, &header_len);
if (status != PJ_SUCCESS)
{
@@ -143,11 +145,11 @@ void StreamSinkRTPImpl::write(const AsteriskSCF::Media::V1::FrameSeq& frames, co
pj_memcpy(packet, (const pjmedia_rtp_hdr*)header, header_len);
/* Now the actual payload */
- pj_memcpy(packet + header_len, &(*frame)->payload[0], (*frame)->payload.size());
+ pj_memcpy(packet + header_len, &framePayload[0], framePayload.size());
/* All done, transmission can now occur */
status = pjmedia_transport_send_rtp(mImpl->mTransport->getTransport(), packet,
- (*frame)->payload.size() + header_len);
+ framePayload.size() + header_len);
if (status != PJ_SUCCESS)
{
@@ -156,7 +158,7 @@ void StreamSinkRTPImpl::write(const AsteriskSCF::Media::V1::FrameSeq& frames, co
}
// Update RTCP information
- pjmedia_rtcp_tx_rtp(mImpl->mSessionAdapter->getRtcpSession(), static_cast<unsigned int>((*frame)->payload.size()));
+ pjmedia_rtcp_tx_rtp(mImpl->mSessionAdapter->getRtcpSession(), static_cast<unsigned int>(framePayload.size()));
}
}
diff --git a/src/RTPSource.cpp b/src/RTPSource.cpp
index 23dae4f..969ce7b 100644
--- a/src/RTPSource.cpp
+++ b/src/RTPSource.cpp
@@ -416,25 +416,29 @@ static void receiveRTP(void *userdata, void *packet, pj_ssize_t size)
if ((audioformat = AudioFormatPtr::dynamicCast(mediaformat)))
{
AudioFramePtr frame = new AudioFrame();
- frame->mediaFormat = mediaformat;
+
+ Ice::ByteSeq bytePayload(payload, payload + payload_size);
+ frame->payload = mediaformat->operations->decodePayload(bytePayload);
// Populate the common data
+ frame->mediaFormat = mediaformat;
frame->timestamp = header->ts;
frame->seqno = header->seq;
- // Copy the payload from the RTP packet into the frame
- copy(payload, payload + payload_size, std::back_inserter(frame->payload));
-
// Into the frames sequence it goes
frames.push_back(frame);
}
else if ((videoformat = VideoFormatPtr::dynamicCast(mediaformat)))
{
VideoFramePtr frame = new VideoFrame();
+
+ Ice::ByteSeq bytePayload(payload, payload + payload_size);
+ frame->payload = mediaformat->operations->decodePayload(bytePayload);
+
frame->mediaFormat = mediaformat;
frame->timestamp = header->ts;
frame->seqno = header->seq;
- copy(payload, payload + payload_size, std::back_inserter(frame->payload));
+
frames.push_back(frame);
}
diff --git a/test/TestRTPpjmedia.cpp b/test/TestRTPpjmedia.cpp
index 3c4e6d8..24a6500 100644
--- a/test/TestRTPpjmedia.cpp
+++ b/test/TestRTPpjmedia.cpp
@@ -753,17 +753,20 @@ BOOST_AUTO_TEST_CASE(TransmitFrametoEmptySink)
AudioFramePtr frame = new AudioFrame();
frame->mediaFormat = format;
+ Ice::ByteSeq payload;
/* Populate the payload with some useless data, but enough to confirm the payload passes unaltered. */
- frame->payload.push_back('a');
- frame->payload.push_back('b');
- frame->payload.push_back('c');
- frame->payload.push_back('d');
- frame->payload.push_back('e');
- frame->payload.push_back('f');
- frame->payload.push_back('g');
- frame->payload.push_back('h');
- frame->payload.push_back('i');
- frame->payload.push_back('j');
+ payload.push_back('a');
+ payload.push_back('b');
+ payload.push_back('c');
+ payload.push_back('d');
+ payload.push_back('e');
+ payload.push_back('f');
+ payload.push_back('g');
+ payload.push_back('h');
+ payload.push_back('i');
+ payload.push_back('j');
+
+ frame->payload = new ByteSeqPayload(payload);
FrameSeq frames;
frames.push_back(frame);
@@ -1070,17 +1073,20 @@ BOOST_AUTO_TEST_CASE(TransmitandReceiveFrame)
AudioFramePtr frame = new AudioFrame();
frame->mediaFormat = format;
+ Ice::ByteSeq payload;
/* Populate the payload with some useless data, but enough to confirm the payload passes unaltered. */
- frame->payload.push_back('a');
- frame->payload.push_back('b');
- frame->payload.push_back('c');
- frame->payload.push_back('d');
- frame->payload.push_back('e');
- frame->payload.push_back('f');
- frame->payload.push_back('g');
- frame->payload.push_back('h');
- frame->payload.push_back('i');
- frame->payload.push_back('j');
+ payload.push_back('a');
+ payload.push_back('b');
+ payload.push_back('c');
+ payload.push_back('d');
+ payload.push_back('e');
+ payload.push_back('f');
+ payload.push_back('g');
+ payload.push_back('h');
+ payload.push_back('i');
+ payload.push_back('j');
+
+ frame->payload = new ByteSeqPayload(payload);
FrameSeq frames;
frames.push_back(frame);
@@ -1139,16 +1145,19 @@ BOOST_AUTO_TEST_CASE(TransmitFrameWithUnsupportedMediaFormat)
AudioFramePtr frame = new AudioFrame();
frame->mediaFormat = format;
- frame->payload.push_back('a');
- frame->payload.push_back('b');
- frame->payload.push_back('c');
- frame->payload.push_back('d');
- frame->payload.push_back('e');
- frame->payload.push_back('f');
- frame->payload.push_back('g');
- frame->payload.push_back('h');
- frame->payload.push_back('i');
- frame->payload.push_back('j');
+ Ice::ByteSeq payload;
+ payload.push_back('a');
+ payload.push_back('b');
+ payload.push_back('c');
+ payload.push_back('d');
+ payload.push_back('e');
+ payload.push_back('f');
+ payload.push_back('g');
+ payload.push_back('h');
+ payload.push_back('i');
+ payload.push_back('j');
+
+ frame->payload = new ByteSeqPayload(payload);
FrameSeq frames;
frames.push_back(frame);
@@ -1218,16 +1227,20 @@ BOOST_AUTO_TEST_CASE(ReceiveUnknownRTPPacket)
frame->mediaFormat = format;
/* Populate the payload with some useless data, but enough to confirm the payload passes unaltered. */
- frame->payload.push_back('a');
- frame->payload.push_back('b');
- frame->payload.push_back('c');
- frame->payload.push_back('d');
- frame->payload.push_back('e');
- frame->payload.push_back('f');
- frame->payload.push_back('g');
- frame->payload.push_back('h');
- frame->payload.push_back('i');
- frame->payload.push_back('j');
+ Ice::ByteSeq payload;
+
+ payload.push_back('a');
+ payload.push_back('b');
+ payload.push_back('c');
+ payload.push_back('d');
+ payload.push_back('e');
+ payload.push_back('f');
+ payload.push_back('g');
+ payload.push_back('h');
+ payload.push_back('i');
+ payload.push_back('j');
+
+ frame->payload = new ByteSeqPayload(payload);
FrameSeq frames;
frames.push_back(frame);
commit 67faf7d901a446943516e0d76583f666d2a4a199
Author: Ken Hunt <ken.hunt at digium.com>
Date: Tue Aug 9 14:50:20 2011 -0500
Register the timer thread with pjlib (if needed). Fixes the test execution for Windows.
diff --git a/src/RTPSource.cpp b/src/RTPSource.cpp
index c9592b1..23dae4f 100644
--- a/src/RTPSource.cpp
+++ b/src/RTPSource.cpp
@@ -47,23 +47,56 @@ Logger lg = getLoggerFactory().getLogger("AsteriskSCF.MediaRTP");
}
/**
+ * Wrapper class around pj_thread_desc.
+ */
+class ThreadDescWrapper
+{
+public:
+ /**
+ * pjthread thread description information, must persist for the life of the thread
+ */
+ pj_thread_desc mDesc;
+};
+
+/**
+ * Type definition used to create a smart pointer for the above.
+ */
+typedef boost::shared_ptr<ThreadDescWrapper> ThreadDescWrapperPtr;
+
+
+/**
* TimerTask implementation which sends RTCP at a defined interval.
*/
class RtcpTransmission : public IceUtil::TimerTask
{
public:
- RtcpTransmission(const SessionAdapterPtr& sessionAdapter, const StreamSinkRTPPrx& sink, const PJMediaTransportPtr& transport) :
- mSessionAdapter(sessionAdapter), mSink(sink), mTransport(transport) { }
+ RtcpTransmission(const SessionAdapterPtr& sessionAdapter,
+ const StreamSinkRTPPrx& sink,
+ const PJMediaTransportPtr& transport,
+ const ThreadDescWrapperPtr& threadDesciptor)
+ : mSessionAdapter(sessionAdapter),
+ mSink(sink),
+ mTransport(transport),
+ mThreadDescriptor(threadDesciptor)
+ {
+ }
void runTimerTask()
{
+ if (pj_thread_is_registered() == PJ_FALSE)
+ {
+ pj_thread_t *thread;
+ pj_status_t status = pj_thread_register("ICE Thread", mThreadDescriptor->mDesc, &thread);
+ assert(status == PJ_SUCCESS);
+ }
+
void *packet;
int packet_size;
pjmedia_rtcp_build_rtcp(mSessionAdapter->getRtcpSession(), &packet, &packet_size);
pjmedia_transport_send_rtcp(mTransport->getTransport(), packet, packet_size);
- std::vector<RTCP::V1::InformationListenerPrx> listeners = mSessionAdapter->getSenderReportListeners();
+ std::vector<RTCP::V1::InformationListenerPrx> listeners = mSessionAdapter->getSenderReportListeners();
// If no listeners exist don't bother getting the statistics
if (listeners.empty())
@@ -71,7 +104,7 @@ public:
return;
}
- RTCP::V1::StatisticsPtr statistics = mSessionAdapter->getSenderReportStatistics();
+ RTCP::V1::StatisticsPtr statistics = mSessionAdapter->getSenderReportStatistics();
for (std::vector<RTCP::V1::InformationListenerPrx>::const_iterator listener = listeners.begin();
listener != listeners.end();
@@ -96,6 +129,8 @@ private:
* Pointer to the transport used for communication.
*/
PJMediaTransportPtr mTransport;
+
+ ThreadDescWrapperPtr mThreadDescriptor;
};
/**
@@ -164,6 +199,8 @@ public:
* Lock that protects information contained.
*/
boost::shared_mutex mLock;
+
+ ThreadDescWrapperPtr mThreadDescriptor;
};
/**
@@ -178,7 +215,8 @@ StreamSourceRTPImplPriv::StreamSourceRTPImplPriv(const SessionAdapterPtr& sessio
mSourceStateItem(new RtpStreamSourceStateItem),
mSessionId(sessionId),
mSource(source),
- mSink(sink)
+ mSink(sink),
+ mThreadDescriptor(new ThreadDescWrapper)
{
pjmedia_rtp_session_init(&mIncomingSession, 0, 0);
mSourceStateItem->sessionId = sessionId;
@@ -552,7 +590,7 @@ void StreamSourceRTPImpl::setRemoteRtcpDetails(const std::string& address, Ice::
{
RtcpTransmissionPtr transmission;
- if ((transmission = new RtcpTransmission(mImpl->mSessionAdapter, mImpl->mSink, mImpl->mTransport)))
+ if ((transmission = new RtcpTransmission(mImpl->mSessionAdapter, mImpl->mSink, mImpl->mTransport, mImpl->mThreadDescriptor)))
{
mImpl->mTimer->scheduleRepeated(transmission, IceUtil::Time::milliSeconds(PJMEDIA_RTCP_INTERVAL));
}
commit 8dde821ebd1f512fe493da504e5c93d7c71403fb
Author: Brent Eagles <beagles at digium.com>
Date: Tue Aug 9 14:08:20 2011 -0230
Correct test initialization to be compatible with recent changes to discovery,
registration, etc.
diff --git a/test/TestRTPICE.cpp b/test/TestRTPICE.cpp
index 36928a9..9c4ad80 100644
--- a/test/TestRTPICE.cpp
+++ b/test/TestRTPICE.cpp
@@ -137,7 +137,6 @@ ConfigurationServicePrx locateConfigurationService(const string& name, const Ser
{
ServiceLocatorParamsPtr query = new ServiceLocatorParams;
query->category = ConfigurationDiscoveryCategory;
- query->service = "default";
query->id = name;
cout << "using name " << name << endl;
return ConfigurationServicePrx::checkedCast(locator->locate(query));
commit 8676ab8ee29c419eb601e2fdcf7dd3b22687f058
Author: Ken Hunt <ken.hunt at digium.com>
Date: Tue Aug 9 10:57:36 2011 -0500
Added scope qualifier to static_pointer_cast. Not caught by Visual Studio.
diff --git a/src/Component.cpp b/src/Component.cpp
index efee798..217e4be 100644
--- a/src/Component.cpp
+++ b/src/Component.cpp
@@ -363,7 +363,7 @@ void Component::createPrimaryServices()
try
{
RtpReplicationContextPtr rtpReplicationContext =
- static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+ boost::static_pointer_cast<RtpReplicationContext>(getReplicationContext());
mConfigurationService = ConfigurationServiceImpl::create();
mConfigurationServicePrx = mConfigurationService->activate(getBackplaneAdapter(), IceUtil::generateUUID());
@@ -450,7 +450,7 @@ void Component::findRemoteServices()
try
{
RtpReplicationContextPtr rtpReplicationContext =
- static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+ boost::static_pointer_cast<RtpReplicationContext>(getReplicationContext());
AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> pw(getServiceLocator(), replicatorParams, lg);
rtpReplicationContext->setReplicator(pw);
@@ -473,7 +473,7 @@ void Component::createReplicationStateListeners()
try
{
RtpReplicationContextPtr rtpReplicationContext =
- static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+ boost::static_pointer_cast<RtpReplicationContext>(getReplicationContext());
// Create and publish our state replicator listener interface.
mReplicatorListener = new RtpStateReplicatorListenerI(getServiceAdapter(), mRtpMediaServicePtr->getEnvironment(),
@@ -494,7 +494,7 @@ void Component::createReplicationStateListeners()
void Component::listenToStateReplicators()
{
RtpReplicationContextPtr rtpReplicationContext =
- static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+ boost::static_pointer_cast<RtpReplicationContext>(getReplicationContext());
if (mListeningToReplicator == true)
{
@@ -531,7 +531,7 @@ void Component::listenToStateReplicators()
void Component::stopListeningToStateReplicators()
{
RtpReplicationContextPtr rtpReplicationContext =
- static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+ boost::static_pointer_cast<RtpReplicationContext>(getReplicationContext());
if ((!rtpReplicationContext->getReplicator().isInitialized()) || (mListeningToReplicator == false))
{
@@ -590,7 +590,7 @@ void Component::onStart()
if (getReplicationContext()->isReplicating() == true)
{
RtpReplicationContextPtr rtpReplicationContext =
- static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+ boost::static_pointer_cast<RtpReplicationContext>(getReplicationContext());
RtpStateItemSeq items;
items.push_back(mGeneralState);
commit ba68890ab24ec2f71cac3236b1cbacb40ee400da
Author: Ken Hunt <ken.hunt at digium.com>
Date: Mon Aug 8 03:08:11 2011 -0500
Refactored to use base Component class and expanded ServiceLocatorParams.
diff --git a/config/test_component.config b/config/test_component.config
index fb6e341..d507786 100644
--- a/config/test_component.config
+++ b/config/test_component.config
@@ -7,6 +7,8 @@
IceBox.InheritProperties=1
IceBox.LoadOrder=ServiceDiscovery,RtpStateReplicator,MediaRTPpjmedia,MediaRTPpjmediaTest
+Ice.Override.Timeout=5000
+
# RtpStateReplicator Configuration
IceBox.Service.RtpStateReplicator=RtpStateReplicator:create
diff --git a/config/test_component_v6.config b/config/test_component_v6.config
index 36021d4..7349902 100644
--- a/config/test_component_v6.config
+++ b/config/test_component_v6.config
@@ -7,6 +7,8 @@
IceBox.InheritProperties=1
IceBox.LoadOrder=ServiceDiscovery,RtpStateReplicator,MediaRTPpjmedia,MediaRTPpjmediaTest
+Ice.Override.Timeout=5000
+
# RtpStateReplicator Configuration
IceBox.Service.RtpStateReplicator=RtpStateReplicator:create
diff --git a/slice/AsteriskSCF/Configuration/MediaRTPPJMedia/RtpConfigurationIf.ice b/slice/AsteriskSCF/Configuration/MediaRTPPJMedia/RtpConfigurationIf.ice
index 1198f65..7853123 100644
--- a/slice/AsteriskSCF/Configuration/MediaRTPPJMedia/RtpConfigurationIf.ice
+++ b/slice/AsteriskSCF/Configuration/MediaRTPPJMedia/RtpConfigurationIf.ice
@@ -37,17 +37,6 @@ module V1
const string ConfigurationDiscoveryCategory = "RtpConfiguration";
/**
- * Service locator parameters class for discovering the configuration service
- */
- unsliceable class RtpConfigurationParams extends AsteriskSCF::Core::Discovery::V1::ServiceLocatorParams
- {
- /**
- * Unique name for the configuration service
- */
- string name;
- };
-
- /**
* Local visitor class for visiting RTP configuration groups
*/
local class RtpConfigurationGroupVisitor extends AsteriskSCF::System::Configuration::V1::ConfigurationGroupVisitor
diff --git a/slice/AsteriskSCF/Replication/MediaRTPPJMedia/RtpStateReplicationIf.ice b/slice/AsteriskSCF/Replication/MediaRTPPJMedia/RtpStateReplicationIf.ice
index 4c44683..b6f49ad 100644
--- a/slice/AsteriskSCF/Replication/MediaRTPPJMedia/RtpStateReplicationIf.ice
+++ b/slice/AsteriskSCF/Replication/MediaRTPPJMedia/RtpStateReplicationIf.ice
@@ -39,11 +39,6 @@ module V1
const string StateReplicatorComponentCategory = "RtpStateReplicatorComponent";
const string StateReplicatorDiscoveryCategory = "RtpStateReplicator";
- unsliceable class RtpStateReplicatorParams extends AsteriskSCF::Core::Discovery::V1::ServiceLocatorParams
- {
- string name;
- };
-
["visitor"] local class RtpStateItemVisitor
{
};
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 13185bd..28e729d 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -2,13 +2,14 @@ include_directories(${astscf-ice-util-cpp_dir}/include)
include_directories(${logger_dir}/include)
astscf_component_init(media_rtp_pjmedia)
-astscf_component_add_files(media_rtp_pjmedia MediaRTPpjmedia.cpp)
+astscf_component_add_files(media_rtp_pjmedia Component.cpp)
astscf_component_add_files(media_rtp_pjmedia RTPSession.cpp)
astscf_component_add_files(media_rtp_pjmedia RTPSource.cpp)
astscf_component_add_files(media_rtp_pjmedia RTPSink.cpp)
astscf_component_add_files(media_rtp_pjmedia RTPSession.h)
astscf_component_add_files(media_rtp_pjmedia RTPSource.h)
astscf_component_add_files(media_rtp_pjmedia RTPSink.h)
+astscf_component_add_files(media_rtp_pjmedia RtpReplicationContext.h)
astscf_component_add_files(media_rtp_pjmedia RtpStateReplicatorListener.cpp)
astscf_component_add_files(media_rtp_pjmedia RtpStateReplicator.h)
astscf_component_add_files(media_rtp_pjmedia RTPConfiguration.cpp)
diff --git a/src/Component.cpp b/src/Component.cpp
new file mode 100644
index 0000000..efee798
--- /dev/null
+++ b/src/Component.cpp
@@ -0,0 +1,621 @@
+/*
+ * Asterisk SCF -- An open-source communications framework.
+ *
+ * Copyright (C) 2010, Digium, Inc.
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk SCF project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE.txt file
+ * at the top of the source tree.
+ */
+
+#include <pjlib.h>
+#include <pjmedia.h>
+
+#include <Ice/Ice.h>
+#include <IceBox/IceBox.h>
+#include <IceUtil/UUID.h>
+
+#include <boost/shared_ptr.hpp>
+
+#include <AsteriskSCF/Core/Discovery/ServiceLocatorIf.h>
+#include <AsteriskSCF/Media/MediaIf.h>
+#include <AsteriskSCF/Media/RTP/MediaRTPIf.h>
+#include <AsteriskSCF/System/Component/ConfigurationIf.h>
+#include <AsteriskSCF/Logger/IceLogger.h>
+#include <AsteriskSCF/logger.h>
+#include <AsteriskSCF/Discovery/SmartProxy.h>
+#include <AsteriskSCF/Component/Component.h>
+
+#include "RtpReplicationContext.h"
+#include "RTPSession.h"
+#include "RtpStateReplicator.h"
+#include "RTPConfiguration.h"
+#include "RtpConfigurationIf.h"
+#include "PJMediaEnvironment.h"
+
+using namespace std;
+using namespace AsteriskSCF::Core::Discovery::V1;
+using namespace AsteriskSCF::Media::V1;
+using namespace AsteriskSCF::Media::RTP::V1;
+using namespace AsteriskSCF::Replication::MediaRTPPJMedia::V1;
+using namespace AsteriskSCF::Configuration::MediaRTPPJMedia::V1;
+using namespace AsteriskSCF::System::Configuration::V1;
+using namespace AsteriskSCF::System::Component::V1;
+using namespace AsteriskSCF::System::Logging;
+using namespace AsteriskSCF::Discovery;
+using namespace AsteriskSCF::Replication;
+using namespace AsteriskSCF::PJMediaRTP;
+
+namespace
+{
+Logger lg = getLoggerFactory().getLogger("AsteriskSCF.MediaRTP");
+}
+
+static const string ReplicaServiceId("MediaRtpReplica");
+static const string MediaServiceId("RTPMediaService");
+static const string MediaComparatorServiceId("RTPMediaServiceComparator");
+
+/**
+ * Implementation of the RTPMediaService interface as defined in MediaRTPIf.ice
+ */
+class RTPMediaServiceImpl : public RTPMediaService
+{
+public:
+ RTPMediaServiceImpl(const Ice::ObjectAdapterPtr&,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr&);
+
+ RTPSessionPrx allocate(const RTPServiceLocatorParamsPtr&, const Ice::Current&);
+ pj_pool_factory *getPoolFactory() { return mEnvironment->poolFactory(); };
+
+ PJMediaEnvironmentPtr getEnvironment()
+ {
+ return mEnvironment;
+ }
+
+private:
+ Ice::ObjectAdapterPtr mAdapter;
+ PJMediaEnvironmentPtr mEnvironment;
+ RtpReplicationContextPtr mReplicationContext;
+ ConfigurationServiceImplPtr mConfigurationService;
+
+#if CONTROL_POINTS_ENABLED
+ AsteriskSCF::PJMediaRTPTesting mMediaServiceSwitchBoard;
+#endif
+};
+
+/**
+ * Typedef which gives us a smart pointer type for RTPMediaServiceImpl class.
+ */
+typedef IceUtil::Handle<RTPMediaServiceImpl> RTPMediaServiceImplPtr;
+
+/**
+ * Implementation of the ServiceLocatorParamsCompare class
+ */
+class RTPMediaServiceCompareServiceImpl : public ServiceLocatorParamsCompare
+{
+public:
+ RTPMediaServiceCompareServiceImpl(const ConfigurationServiceImplPtr& config) :
+ mConfig(config)
+ {
+ }
+
+ bool isSupported(const ServiceLocatorParamsPtr& locatorParams, const Ice::Current&)
+ {
+ RTPServiceLocatorParamsPtr params;
+
+ if (!(params = RTPServiceLocatorParamsPtr::dynamicCast(locatorParams)))
+ {
+ return false;
+ }
+
+ bool result = true;
+
+ // This is done on purpose for additional checks in the future
+ if (params->ipv6 == true)
+ {
+#if defined(PJ_HAS_IPV6) && PJ_HAS_IPV6!=0
+ result = true;
+#else
+ result = false;
+#endif
+ }
+ //
+ // We can ignore the SRTP criteria since we support it one way or the other.
+ //
+ if (!result)
+ {
+ return false;
+ }
+
+ RTPOverICEServiceLocatorParamsPtr iceParams = RTPOverICEServiceLocatorParamsPtr::dynamicCast(locatorParams);
+ if (iceParams)
+ {
+ if (iceParams->enableRTPOverICE)
+ {
+ NATConfigPtr natConfig = mConfig->natConfig();
+
+ if (natConfig && natConfig->isSTUNEnabled())
+ {
+ if (iceParams->enableTURN)
+ {
+ if (!natConfig->isTURNEnabled())
+ {
+ result = false;
+ }
+ }
+ }
+ else
+ {
+ result = false;
+ }
+ }
+ //
+ // We ignore the else case because we can definitely do non-ICE related stuff... its not clear
+ // that negative matches in this case should be exclusionary. Actual ICE usage will be specified
+ // when the RTP session is allocated.
+ //
+ }
+
+ return result;
+ }
+
+private:
+ ConfigurationServiceImplPtr mConfig;
+};
+
+/**
+ * Implementation of the Component class.
+ */
+class Component : public AsteriskSCF::Component::Component
+{
+public:
+ Component() :
+ AsteriskSCF::Component::Component(lg, AsteriskSCF::Media::RTP::V1::ComponentServiceDiscoveryCategory),
+ mListeningToReplicator(false), mGeneralState(new RtpGeneralStateItem()) { mGeneralState->key = IceUtil::generateUUID(); };
+
+private:
+ // Required base Component overrides
+ virtual void createPrimaryServices();
+ virtual void preparePrimaryServicesForDiscovery();
+ virtual void createReplicationStateListeners();
+ virtual void stopListeningToStateReplicators();
+ virtual void listenToStateReplicators();
+ virtual void findRemoteServices();
+ virtual void onRegisterPrimaryServices();
+
+ // Optional base Component notifcation overrides
+ virtual void onSuspend();
+ virtual void onResume();
+ virtual void onPreInitialize();
+ virtual void onStop();
+ virtual void onStart();
+
+ // Other base Component overrides
+ virtual void prepareBackplaneServicesForDiscovery();
+ ReplicationContextPtr createReplicationContext(ReplicationStateType state);
+
+ // A proxy to the service locator manager for the component service.
+ ServiceManagementPrx mComponentServiceManagement;
+
+ // State replicator listener.
+ RtpStateReplicatorListenerPtr mReplicatorListener;
+ RtpStateReplicatorListenerPrx mReplicatorListenerProxy;
+ bool mListeningToReplicator;
+
+ // An instance of the general state information class.
+ RtpGeneralStateItemPtr mGeneralState;
+
+ // Media service
+ RTPMediaServiceImplPtr mRtpMediaServicePtr;
+ RTPMediaServicePrx mRtpMediaServicePrx;
+ LocatorRegistrationWrapperPtr mRtpMediaServiceRegistration;
+
+ // Media comparator service
+ ServiceLocatorParamsComparePtr mRtpMediaComparatorService;
+ ServiceLocatorParamsComparePrx mRtpMediaComparatorServicePrx;
+ RTPOverICEServiceLocatorParamsPtr mRtpOverIceLocatorParams;
+
+ // Configuration state
+ ConfigurationServiceImplPtr mConfigurationService;
+ ConfigurationServicePrx mConfigurationServicePrx;
+ LocatorRegistrationWrapperPtr mConfigurationRegistration;
+};
+
+void Component::onSuspend()
+{
+ mGeneralState->serviceManagement->suspend();
+}
+
+void Component::onResume()
+{
+ mGeneralState->serviceManagement->unsuspend();
+}
+
+/**
+ * Wrapper class around pj_thread_desc.
+ */
+class ThreadDescWrapper
+{
+public:
+ /**
+ * pjthread thread description information, must persist for the life of the thread
+ */
+ pj_thread_desc mDesc;
+};
+
+/**
+ * Type definition used to create a smart pointer for the above.
+ */
+typedef boost::shared_ptr<ThreadDescWrapper> ThreadDescWrapperPtr;
+
+/**
+ * Implementation of the Ice::ThreadNotification class.
+ */
+class pjlibHook : public Ice::ThreadNotification
+{
+public:
+ /**
+ * Implementation of the start function which is called when a thread starts.
+ */
+ void start()
+ {
+ ThreadDescWrapperPtr wrapper = ThreadDescWrapperPtr(new ThreadDescWrapper());
+ pj_thread_t *thread;
+ pj_thread_register("ICE Thread", wrapper->mDesc, &thread);
+ boost::lock_guard<boost::mutex> lock(mLock);
+ pjThreads.insert(make_pair(thread, wrapper));
+ }
+
+ /**
+ * Implementation of the stop function which is called when a thread stops.
+ */
+ void stop()
+ {
+ if (pj_thread_is_registered())
+ {
+ boost::lock_guard<boost::mutex> lock(mLock);
+ pjThreads.erase(pj_thread_this());
+ }
+ }
+private:
+ /**
+ * A map containing thread lifetime persistent data.
+ */
+ map<pj_thread_t*, ThreadDescWrapperPtr> pjThreads;
+
+ /**
+ * Mutex to protect the map
+ */
+ boost::mutex mLock;
+};
+
+/**
+ * Constructor for the RTPMediaServiceImpl class.
+ */
+RTPMediaServiceImpl::RTPMediaServiceImpl(const Ice::ObjectAdapterPtr& adapter,
+ const RtpReplicationContextPtr& replicationContext,
+ const ConfigurationServiceImplPtr& configurationService) :
+ mAdapter(adapter),
+ mEnvironment(PJMediaEnvironment::create(adapter->getCommunicator()->getProperties(), configurationService)),
+ mReplicationContext(replicationContext),
+ mConfigurationService(configurationService)
+{
+}
+
+/**
+ * Implementation of the allocate method as defined in MediaRTPIf.ice
+ */
+RTPSessionPrx RTPMediaServiceImpl::allocate(const RTPServiceLocatorParamsPtr& params, const Ice::Current&)
+{
+ return AsteriskSCF::PJMediaRTP::RTPSession::create(mAdapter, IceUtil::generateUUID(), params, mEnvironment,
+ mReplicationContext, mConfigurationService);
+}
+
+void Component::onPreInitialize()
+{
+ /* Initialize pjlib as pjmedia will be using it */
+ pj_status_t status = pj_init();
+ if (status != PJ_SUCCESS)
+ {
+ lg(Error) << "PJ library initialization failed.";
+ return;
+ }
+
+ if ((status = pjlib_util_init()) != PJ_SUCCESS)
+ {
+ lg(Error) << "PJ Utility library initialization failed.";
+ return;
+ }
+
+ lg(Info) << "Initializing pjmedia rtp component" << endl;
+
+ Ice::InitializationData id;
+ id.threadHook = new pjlibHook();
+ id.properties = getCommunicator()->getProperties();
+
+ // To use our thread-hook, we need to set an alternate
+ // communicator in our Component base.
+ setCommunicator(Ice::initialize(id));
+}
+
+/**
+ * Override of factory method to create our custom replication context.
+ */
+ReplicationContextPtr Component::createReplicationContext(ReplicationStateType state)
+{
+ RtpReplicationContextPtr context(new RtpReplicationContext(state));
+ return context;
+}
+
+/**
+ * Create the objects that implement the main services this component provides
+ * the system.
+ */
+void Component::createPrimaryServices()
+{
+ try
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ mConfigurationService = ConfigurationServiceImpl::create();
+ mConfigurationServicePrx = mConfigurationService->activate(getBackplaneAdapter(), IceUtil::generateUUID());
+
+ mRtpMediaServicePtr =
+ new RTPMediaServiceImpl(getServiceAdapter(), rtpReplicationContext, mConfigurationService);
+ mRtpMediaServicePrx = RTPMediaServicePrx::uncheckedCast(getServiceAdapter()->add(mRtpMediaServicePtr,
+ getCommunicator()->stringToIdentity(MediaServiceId)));
+
+ mRtpMediaComparatorService = new RTPMediaServiceCompareServiceImpl(mConfigurationService);
+ mRtpMediaComparatorServicePrx = ServiceLocatorParamsComparePrx::uncheckedCast(
+ getServiceAdapter()->add(mRtpMediaComparatorService,
+ getCommunicator()->stringToIdentity(MediaComparatorServiceId)));
+
+
+ mRtpOverIceLocatorParams = new RTPOverICEServiceLocatorParams;
+ mRtpOverIceLocatorParams->category = "rtp";
+ PJMediaEnvironmentPtr mediaEnvironment = mRtpMediaServicePtr->getEnvironment();
+
+ //
+ // Service wide configuration is done through properties allowing certain features
+ // to be completely disabled.
+ //
+ NATConfigPtr natConfig = mediaEnvironment->natConfig();
+ if (natConfig && natConfig->isSTUNEnabled())
+ {
+ mRtpOverIceLocatorParams->enableRTPOverICE = true;
+ mRtpOverIceLocatorParams->enableTURN = natConfig->isTURNEnabled();
+ }
+ else
+ {
+ mRtpOverIceLocatorParams->enableRTPOverICE = false;
+ mRtpOverIceLocatorParams->enableTURN = false;
+ }
+
+ if (rtpReplicationContext->isActive() == true)
+ {
+ mGeneralState->comparatorId = IceUtil::generateUUID();
+ getServiceLocatorManagement()->addCompare(mGeneralState->comparatorId, mRtpMediaComparatorServicePrx);
+ }
+
+ }
+ catch(const Ice::Exception& e)
+ {
+ lg(Critical) << getName() << " : " << BOOST_CURRENT_FUNCTION << " : " << e.what();
+ }
+}
+
+/**
+ * Prepare this component's backplane interfaces for the Service Locator.
+ * This enables other Asterisk SCF components to locate our interfaces.
+ */
+void Component::prepareBackplaneServicesForDiscovery()
+{
+ // Insure the default Component services are prepped.
+ AsteriskSCF::Component::Component::prepareBackplaneServicesForDiscovery();
+
+ try
+ {
+ // Register our configuration interface with the Service Locator.
+ mConfigurationRegistration = wrapServiceForRegistration(mConfigurationServicePrx,
+ ConfigurationDiscoveryCategory);
+ manageBackplaneService(mConfigurationRegistration);
+ }
+ catch(const std::exception& e)
+ {
+ lg(Error) << "Exception in " << getName() << ", " << BOOST_CURRENT_FUNCTION << " : " << e.what();
+ }
+}
+
+void Component::findRemoteServices()
+{
+ if (getReplicationContext()->getState() == ACTIVE_STANDALONE)
+ {
+ return;
+ }
+
+ // Look for the configured state replicator or default one
+ ServiceLocatorParamsPtr replicatorParams = new ServiceLocatorParams();
+ replicatorParams->category = StateReplicatorDiscoveryCategory;
+ replicatorParams->service =
+ getCommunicator()->getProperties()->getPropertyWithDefault("Rtp.StateReplicatorService", "default");
+
+ try
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> pw(getServiceLocator(), replicatorParams, lg);
+ rtpReplicationContext->setReplicator(pw);
+
+ // Since we're not in standalone mode, we'll get our configuration updates routed via the
+ // replicator service.
+ ConfigurationReplicatorPrx configurationReplicator = ConfigurationReplicatorPrx::checkedCast(
+ rtpReplicationContext->getReplicator().initialize(), ReplicatorFacet);
+ configurationReplicator->registerConfigurationService(mConfigurationServicePrx);
+
+ }
+ catch (const std::exception& e)
+ {
+ lg(Error) << getName() << ": " << BOOST_CURRENT_FUNCTION << " State replicator could not be found, operating without. " << e.what();
+ }
+}
+
+void Component::createReplicationStateListeners()
+{
+ try
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ // Create and publish our state replicator listener interface.
+ mReplicatorListener = new RtpStateReplicatorListenerI(getServiceAdapter(), mRtpMediaServicePtr->getEnvironment(),
+ mGeneralState, rtpReplicationContext, mConfigurationService);
+ RtpStateReplicatorListenerPrx replicatorListener = RtpStateReplicatorListenerPrx::uncheckedCast(
+ getBackplaneAdapter()->addWithUUID(mReplicatorListener));
+ mReplicatorListenerProxy = RtpStateReplicatorListenerPrx::uncheckedCast(replicatorListener->ice_oneway());
+
+ lg(Debug) << "Got proxy to RTP state replicator";
+ }
+ catch(const Ice::Exception &e)
+ {
+ lg(Error) << getName() << " in " << BOOST_CURRENT_FUNCTION << " : " << e.what();
+ throw;
+ }
+}
+
+void Component::listenToStateReplicators()
+{
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ if (mListeningToReplicator == true)
+ {
+ return;
+ }
+
+ if (!rtpReplicationContext->getReplicator().isInitialized())
+ {
+ lg(Error) << getName() << " : State replicator could not be found. Unable to listen for state updates!";
+ return;
+ }
+
+ try
+ {
+ // Are we in standby mode?
+ if (rtpReplicationContext->getState() == STANDBY_IN_REPLICA_GROUP)
+ {
+ rtpReplicationContext->getReplicator()->addListener(mReplicatorListenerProxy);
+ mListeningToReplicator = true;
+ }
+ }
+ catch (const Ice::Exception& e)
+ {
+ lg(Error) << e.what();
+ throw;
+ }
+}
+
+/**
+ * Unregister as a listener to our state replicator.
+ * A component in active mode doesn't neeed to listen to
+ * state replication data.
+ */
+void Component::stopListeningToStateReplicators()
+{
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ if ((!rtpReplicationContext->getReplicator().isInitialized()) || (mListeningToReplicator == false))
+ {
+ return;
+ }
+
+ try
+ {
+ rtpReplicationContext->getReplicator()->removeListener(mReplicatorListenerProxy);
+ mListeningToReplicator = false;
+ }
+ catch (const Ice::Exception& e)
+ {
+ lg(Error) << e.what();
+ throw;
+ }
+}
+
+/**
+ * Prepares this component's primary public interfaces for discovery via the Service Locator.
+ * This enables other Asterisk SCF components to locate the interfaces we are publishing.
+ */
+void Component::preparePrimaryServicesForDiscovery()
+{
+ try
+ {
+ mRtpMediaServiceRegistration = wrapServiceForRegistration(mRtpMediaServicePrx,
+ "rtp");
+ managePrimaryService(mRtpMediaServiceRegistration);
+ }
+ catch(const std::exception& e)
+ {
+ lg(Error) << "Unable to publish component interfaces in " << getName() << BOOST_CURRENT_FUNCTION <<
+ ". Exception: " << e.what();
+ throw; // rethrow
+ }
+}
+
+void Component::onRegisterPrimaryServices()
+{
+ if (getReplicationContext()->isActive() == false)
+ {
+ return;
+ }
+
+ mGeneralState->serviceManagement = mRtpMediaServiceRegistration->getServiceMangement();
+ mGeneralState->serviceManagement->addLocatorParams(mRtpOverIceLocatorParams, mGeneralState->comparatorId);
+}
+
+void Component::onStart()
+{
+ // Note: I don't think this is necessary. If we make the
+ // comparator computed from a "service" identifier (which could default
+ // to "default"), there's nothing replicated here that the standby component
+ // couldn't already determine itself.
+ if (getReplicationContext()->isReplicating() == true)
+ {
+ RtpReplicationContextPtr rtpReplicationContext =
+ static_pointer_cast<RtpReplicationContext>(getReplicationContext());
+
+ RtpStateItemSeq items;
+ items.push_back(mGeneralState);
+ RtpStateReplicatorPrx oneway = RtpStateReplicatorPrx::uncheckedCast(rtpReplicationContext->getReplicator()->ice_oneway());
+ oneway->setState(items);
+ }
+}
+
+void Component::onStop()
+{
+ if (getReplicationContext()->isActive() == true)
+ {
+ mGeneralState->serviceManagement->unregister();
+ }
+
+ if (!mGeneralState->comparatorId.empty())
+ {
+ getServiceLocatorManagement()->removeCompare(mGeneralState->comparatorId);
+ }
+}
+
+extern "C"
+{
+ASTSCF_DLL_EXPORT IceBox::Service* create(Ice::CommunicatorPtr)
+{
+ return new Component;
+}
+}
diff --git a/src/MediaRTPpjmedia.cpp b/src/MediaRTPpjmedia.cpp
deleted file mode 100644
index 9b2933f..0000000
--- a/src/MediaRTPpjmedia.cpp
+++ /dev/null
@@ -1,716 +0,0 @@
-/*
- * Asterisk SCF -- An open-source communications framework.
- *
- * Copyright (C) 2010, Digium, Inc.
- *
- * See http://www.asterisk.org for more information about
- * the Asterisk SCF project. Please do not directly contact
- * any of the maintainers of this project for assistance;
- * the project provides a web site, mailing lists and IRC
- * channels for your use.
- *
- * This program is free software, distributed under the terms of
- * the GNU General Public License Version 2. See the LICENSE.txt file
- * at the top of the source tree.
- */
-
-#include <pjlib.h>
-#include <pjmedia.h>
-
-#include <Ice/Ice.h>
-#include <IceBox/IceBox.h>
-#include <IceUtil/UUID.h>
-
-#include <boost/shared_ptr.hpp>
-
-#include <AsteriskSCF/Core/Discovery/ServiceLocatorIf.h>
-#include <AsteriskSCF/Media/MediaIf.h>
-#include <AsteriskSCF/Media/RTP/MediaRTPIf.h>
-#include <AsteriskSCF/System/Component/ConfigurationIf.h>
-#include <AsteriskSCF/System/Component/ComponentServiceIf.h>
-#include <AsteriskSCF/System/Component/ReplicaIf.h>
-#include <AsteriskSCF/Logger/IceLogger.h>
-#include <AsteriskSCF/logger.h>
-#include <AsteriskSCF/Discovery/SmartProxy.h>
-
-#include "RtpStateReplicationIf.h"
-
-#include "RTPSession.h"
-#include "RtpStateReplicator.h"
-#include "RTPConfiguration.h"
-#include "RtpConfigurationIf.h"
-
-#include "PJMediaEnvironment.h"
-
-using namespace std;
-using namespace AsteriskSCF::Core::Discovery::V1;
-using namespace AsteriskSCF::Media::V1;
-using namespace AsteriskSCF::Media::RTP::V1;
-using namespace AsteriskSCF::Replication::MediaRTPPJMedia::V1;
-using namespace AsteriskSCF::Configuration::MediaRTPPJMedia::V1;
-using namespace AsteriskSCF::System::Configuration::V1;
-using namespace AsteriskSCF::System::Component::V1;
-using namespace AsteriskSCF::System::Logging;
-using namespace AsteriskSCF::Discovery;
-using namespace AsteriskSCF::PJMediaRTP;
-
-namespace
-{
-Logger lg = getLoggerFactory().getLogger("AsteriskSCF.MediaRTP");
-}
-
-static const string ReplicaServiceId("MediaRtpReplica");
-static const string MediaServiceId("RTPMediaService");
-static const string MediaComparatorServiceId("RTPMediaServiceComparator");
-
-/**
- * Implementation of the RTPMediaService interface as defined in MediaRTPIf.ice
- */
-class RTPMediaServiceImpl : public RTPMediaService
-{
-public:
- RTPMediaServiceImpl(const Ice::ObjectAdapterPtr&, const ReplicaPrx&,
- const AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx>&,
- const ConfigurationServiceImplPtr&);
- RTPSessionPrx allocate(const RTPServiceLocatorParamsPtr&, const Ice::Current&);
- pj_pool_factory *getPoolFactory() { return mEnvironment->poolFactory(); };
-
- PJMediaEnvironmentPtr getEnvironment()
- {
- return mEnvironment;
- }
-
-private:
- /**
- * A pointer to the object adapter that objects should be added to.
- */
- Ice::ObjectAdapterPtr mAdapter;
-
- /**
- * The media environment object.
- */
- PJMediaEnvironmentPtr mEnvironment;
-
- /**
- * A proxy for the replica service
- */
- ReplicaPrx mReplicaServicePrx;
-
- /**
- * A pointer to the configuration service.
- */
- ConfigurationServiceImplPtr mConfigurationService;
-
- /**
- * A proxy to the state replicator.
- */
- AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> mStateReplicator;
-
-#if CONTROL_POINTS_ENABLED
- AsteriskSCF::PJMediaRTPTesting mMediaServiceSwitchBoard;
-#endif
-};
-
-/**
- * Typedef which gives us a smart pointer type for RTPMediaServiceImpl class.
- */
-typedef IceUtil::Handle<RTPMediaServiceImpl> RTPMediaServiceImplPtr;
-
-/**
- * This class provides implementation for the Replica interface.
- */
-class ReplicaImpl : public Replica
-{
-public:
- ReplicaImpl(const Ice::ObjectAdapterPtr& adapter) : mAdapter(adapter), mPaused(false), mActive(true) { }
-
- bool isActive(const Ice::Current&)
- {
- return mActive;
- }
-
- bool activate(const Ice::Current&)
- {
- mActive = true;
-
- for (vector<AsteriskSCF::System::Component::V1::ReplicaListenerPrx>::const_iterator listener =
- mListeners.begin(); listener != mListeners.end(); ++listener)
- {
- (*listener)->activated(ReplicaPrx::uncheckedCast(
- mAdapter->createDirectProxy(mAdapter->getCommunicator()->stringToIdentity(ReplicaServiceId))));
- }
-
- return true;
- }
-
- void standby(const Ice::Current&)
- {
- mActive = false;
-
- for (vector<AsteriskSCF::System::Component::V1::ReplicaListenerPrx>::const_iterator listener =
- mListeners.begin(); listener != mListeners.end(); ++listener)
- {
- (*listener)->onStandby(ReplicaPrx::uncheckedCast(
- mAdapter->createDirectProxy(mAdapter->getCommunicator()->stringToIdentity(ReplicaServiceId))));
- }
- }
-
- void addListener(const AsteriskSCF::System::Component::V1::ReplicaListenerPrx& listener, const Ice::Current&)
- {
- mListeners.push_back(listener);
- }
-
- void removeListener(const AsteriskSCF::System::Component::V1::ReplicaListenerPrx& listener, const Ice::Current&)
- {
- mListeners.erase(std::remove(mListeners.begin(), mListeners.end(), listener), mListeners.end());
- }
-
-private:
- /**
- * Pointer to the object adapter we exist on.
- */
- Ice::ObjectAdapterPtr mAdapter;
-
- /**
- * Listeners that we need to push state change notifications out to.
- */
- vector<AsteriskSCF::System::Component::V1::ReplicaListenerPrx> mListeners;
-
- bool mPaused;
-
- bool mActive;
-};
-
-/**
- * Implementation of the ServiceLocatorParamsCompare class
- */
-class RTPMediaServiceCompareServiceImpl : public ServiceLocatorParamsCompare
-{
-public:
- RTPMediaServiceCompareServiceImpl(const ConfigurationServiceImplPtr& config) :
- mConfig(config)
- {
- }
-
- bool isSupported(const ServiceLocatorParamsPtr& locatorParams, const Ice::Current&)
- {
- RTPServiceLocatorParamsPtr params;
-
- if (!(params = RTPServiceLocatorParamsPtr::dynamicCast(locatorParams)))
- {
- return false;
- }
-
- bool result = true;
-
- // This is done on purpose for additional checks in the future
- if (params->ipv6 == true)
- {
-#if defined(PJ_HAS_IPV6) && PJ_HAS_IPV6!=0
- result = true;
-#else
- result = false;
-#endif
- }
- //
- // We can ignore the SRTP criteria since we support it one way or the other.
- //
- if (!result)
- {
- return false;
- }
-
- RTPOverICEServiceLocatorParamsPtr iceParams = RTPOverICEServiceLocatorParamsPtr::dynamicCast(locatorParams);
- if (iceParams)
- {
- if (iceParams->enableRTPOverICE)
- {
- NATConfigPtr natConfig = mConfig->natConfig();
-
- if (natConfig && natConfig->isSTUNEnabled())
- {
- if (iceParams->enableTURN)
- {
- if (!natConfig->isTURNEnabled())
- {
- result = false;
- }
- }
- }
- else
- {
- result = false;
- }
- }
- //
- // We ignore the else case because we can definitely do non-ICE related stuff... its not clear
- // that negative matches in this case should be exclusionary. Actual ICE usage will be specified
- // when the RTP session is allocated.
- //
- }
-
- return result;
- };
-
-private:
- ConfigurationServiceImplPtr mConfig;
-};
-
-/**
- * Implementation of the IceBox::Service class
- */
-class MediaRTPpjmediaApp : public IceBox::Service
-{
-public:
- MediaRTPpjmediaApp() : mGeneralState(new RtpGeneralStateItem()) { mGeneralState->key = IceUtil::generateUUID(); };
- void start(const std::string&, const Ice::CommunicatorPtr&, const Ice::StringSeq&);
- void stop();
-
-private:
- /**
- * Ice Communicator used for this service.
- */
- Ice::CommunicatorPtr mCommunicator;
-
- /**
- * Object adapter that global stuff is associated with.
- */
- Ice::ObjectAdapterPtr mGlobalAdapter;
-
- /**
- * Object adapter that local stuff is associated with.
- */
- Ice::ObjectAdapterPtr mLocalAdapter;
-
- /**
- * The object adapter for the Logger.
- */
- Ice::ObjectAdapterPtr mLoggerAdapter;
-
- /**
- * A proxy to the service locator manager for the component service.
- */
- ServiceManagementPrx mComponentServiceManagement;
-
- /**
- * Instance of our replica implementation.
- */
- ReplicaPtr mReplicaService;
-
- /**
- * A proxy to the replica control object.
- */
- ReplicaPrx mReplicaServicePrx;
-
- /**
- * Instance of our configuration service implementation.
- */
- ConfigurationServiceImplPtr mConfigurationService;
-
- /**
- * Instance of our state replicator listener.
- */
- RtpStateReplicatorListenerPtr mReplicatorListener;
-
- /**
- * A proxy to our state replicator listener.
- */
- RtpStateReplicatorListenerPrx mReplicatorListenerProxy;
-
- /**
- * A proxy to the state replicator.
- */
- AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> mStateReplicator;
-
- /**
- * An instance of the general state information class.
- */
- RtpGeneralStateItemPtr mGeneralState;
-
- /**
- * A proxy to the service locator management service.
- */
- ServiceLocatorManagementPrx mManagement;
-
- /**
- * A proxy to the service locator manager for the configuration service.
- */
- ServiceManagementPrx mConfigurationManagement;
-
- /**
- * Unique guid for configuration service name comparator.
- */
- std::string mConfigCompareGuid;
-};
-
-/**
- * Implementation of the ComponentService interface as defined in ComponentServiceIf.ice
- */
-class ComponentServicepjmediaImpl : public ComponentService
-{
-public:
- /**
- * A constructor for this implementation which just sets a few variables, nothing extreme.
- */
- ComponentServicepjmediaImpl(MediaRTPpjmediaApp& app, const RtpGeneralStateItemPtr& generalState) :
- mApplication(app), mGeneralState(generalState) { };
-
- /**
- * An implementation of the suspend method which actually suspends ourselves
- * from the service locator.
- */
- virtual void suspend(const ::Ice::Current&)
- {
- mGeneralState->serviceManagement->suspend();
- }
-
- /**
- * An implementation of the resume method which actually unsuspends ourselves
- * from the service locator.
- */
- virtual void resume(const ::Ice::Current&)
- {
- mGeneralState->serviceManagement->unsuspend();
- }
-
- /**
- * An implementation of the shutdown method which really does shut us down.
- * Goodbye cruel world.
- */
- virtual void shutdown(const ::Ice::Current&)
- {
- // TODO - Actually support this
- }
-
-private:
- /**
- * Our application instance, used for shutting the component down.
- */
- MediaRTPpjmediaApp& mApplication;
-
- /**
- * Pointer to general state information.
- */
- RtpGeneralStateItemPtr mGeneralState;
-};
-
-/**
- * Wrapper class around pj_thread_desc.
- */
-class ThreadDescWrapper
-{
-public:
- /**
- * pjthread thread description information, must persist for the life of the thread
- */
- pj_thread_desc mDesc;
-};
-
-/**
- * Type definition used to create a smart pointer for the above.
- */
-typedef boost::shared_ptr<ThreadDescWrapper> ThreadDescWrapperPtr;
-
-/**
- * Implementation of the Ice::ThreadNotification class.
- */
-class pjlibHook : public Ice::ThreadNotification
-{
-public:
- /**
- * Implementation of the start function which is called when a thread starts.
- */
- void start()
- {
- ThreadDescWrapperPtr wrapper = ThreadDescWrapperPtr(new ThreadDescWrapper());
- pj_thread_t *thread;
- pj_thread_register("ICE Thread", wrapper->mDesc, &thread);
- boost::lock_guard<boost::mutex> lock(mLock);
- pjThreads.insert(make_pair(thread, wrapper));
- }
-
- /**
- * Implementation of the stop function which is called when a thread stops.
- */
- void stop()
- {
- if (pj_thread_is_registered())
- {
- boost::lock_guard<boost::mutex> lock(mLock);
- pjThreads.erase(pj_thread_this());
- }
- }
-private:
- /**
- * A map containing thread lifetime persistent data.
- */
- map<pj_thread_t*, ThreadDescWrapperPtr> pjThreads;
-
- /**
- * Mutex to protect the map
- */
- boost::mutex mLock;
-};
-
-/**
- * Comparator implementation for name based configuration service locating
- */
-class RtpConfigurationCompare : public ServiceLocatorParamsCompare
-{
-public:
- RtpConfigurationCompare(const string& name) : mName(name) {}
- bool isSupported(const ServiceLocatorParamsPtr ¶ms, const Ice::Current &)
- {
- RtpConfigurationParamsPtr configParams = RtpConfigurationParamsPtr::dynamicCast(params);
- if (configParams->name == mName)
- {
- return true;
- }
- return false;
- }
-private:
- string mName;
-};
-
-typedef IceUtil::Handle<RtpConfigurationCompare> RtpConfigurationComparePtr;
-
-/**
- * Constructor for the RTPMediaServiceImpl class.
- */
-RTPMediaServiceImpl::RTPMediaServiceImpl(const Ice::ObjectAdapterPtr& adapter, const ReplicaPrx& replicaService,
- const AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx>& stateReplicator,
- const ConfigurationServiceImplPtr& configurationService) :
- mAdapter(adapter),
- mEnvironment(PJMediaEnvironment::create(adapter->getCommunicator()->getProperties(), configurationService)),
- mReplicaServicePrx(replicaService),
- mConfigurationService(configurationService),
- mStateReplicator(stateReplicator)
-{
-}
-
-/**
- * Implementation of the allocate method as defined in MediaRTPIf.ice
- */
-RTPSessionPrx RTPMediaServiceImpl::allocate(const RTPServiceLocatorParamsPtr& params, const Ice::Current&)
-{
- return AsteriskSCF::PJMediaRTP::RTPSession::create(mAdapter, IceUtil::generateUUID(), params, mEnvironment,
- mReplicaServicePrx, mStateReplicator, mConfigurationService);
-}
-
-/**
- * Implementation of the IceBox::Service::start method.
- */
-void MediaRTPpjmediaApp::start(const std::string&, const Ice::CommunicatorPtr& communicator,
- const Ice::StringSeq&)
-{
- // we need a logger before we're ready to build the real communicator.
- // use the one we're provided to create the IceLogger.
- mLoggerAdapter = communicator->createObjectAdapter("MediaRTPpjmediaAdapterLogger");
- ConfiguredIceLoggerPtr iceLogger = createIceLogger(mLoggerAdapter);
- getLoggerFactory().setLogOutput(iceLogger->getLogger());
- mLoggerAdapter->activate();
-
- /* Initialize pjlib as pjmedia will be using it */
- pj_status_t status = pj_init();
- if (status != PJ_SUCCESS)
- {
- lg(Error) << "PJ library initialization failed.";
- return;
- }
-
- if ((status = pjlib_util_init()) != PJ_SUCCESS)
- {
- lg(Error) << "PJ Utility library initialization failed.";
- return;
- }
-
- lg(Info) << "Initializing pjmedia rtp component" << endl;
-
- Ice::InitializationData id;
- id.threadHook = new pjlibHook();
- id.properties = communicator->getProperties();
-
- mCommunicator = Ice::initialize(id);
-
- mLocalAdapter = mCommunicator->createObjectAdapter("MediaRTPpjmediaAdapterLocal");
-
- mReplicaService = new ReplicaImpl(mLocalAdapter);
- mReplicaServicePrx = ReplicaPrx::uncheckedCast(mLocalAdapter->add(mReplicaService, mCommunicator->stringToIdentity(ReplicaServiceId)));
-
- mConfigurationService = ConfigurationServiceImpl::create();
- ConfigurationServicePrx mConfigurationServiceProxy = mConfigurationService->activate(mLocalAdapter, IceUtil::generateUUID());
- mLocalAdapter->activate();
-
- mGlobalAdapter = mCommunicator->createObjectAdapter("MediaRTPpjmediaAdapter");
-
- mGlobalAdapter->activate();
-
- lg(Info) << "Activated pjmedia rtp component media service." << endl;
-
- mManagement = ServiceLocatorManagementPrx::checkedCast(mCommunicator->propertyToProxy("ServiceLocatorManagementProxy"));
-
- // The service locator is required for state replicator operation, so go ahead and find it
- ServiceLocatorPrx locator = ServiceLocatorPrx::checkedCast(mCommunicator->propertyToProxy("LocatorService.Proxy"));
-
- // Look for the configured state replicator or default one
- RtpStateReplicatorParamsPtr replicatorParams = new RtpStateReplicatorParams();
- replicatorParams->category = StateReplicatorDiscoveryCategory;
- replicatorParams->name =
- mCommunicator->getProperties()->getPropertyWithDefault("Rtp.StateReplicatorName", "default");
-
- try
- {
- AsteriskSCF::Discovery::SmartProxy<RtpStateReplicatorPrx> pw(locator, replicatorParams, lg);
- mStateReplicator = pw;
- }
- catch (...)
- {
- lg(Error) << "State replicator could not be found, operating without.";
- }
-
- RTPMediaServiceImplPtr rtpmediaservice =
- new RTPMediaServiceImpl(mGlobalAdapter, mReplicaServicePrx, mStateReplicator, mConfigurationService);
-
- if (mCommunicator->getProperties()->getPropertyWithDefault("Rtp.Standalone", "false") == "true")
- {
- // Publish the configuration service IceStorm topic so everybody gets configuration
- mConfigurationManagement = ServiceManagementPrx::uncheckedCast(
- mManagement->addService(mConfigurationServiceProxy, ""));
-
- // Populate the configuration parameters with details so we can be found
- RtpConfigurationParamsPtr configurationParams = new RtpConfigurationParams();
- configurationParams->category = ConfigurationDiscoveryCategory;
- configurationParams->name = mCommunicator->getProperties()->getPropertyWithDefault("RtpConfiguration.Name", "");
-
- // Add our custom comparator so we can support multiple simultaneous configuration sinks
- RtpConfigurationComparePtr configNameCompare = new RtpConfigurationCompare(configurationParams->name);
- ServiceLocatorParamsComparePrx configCompareProxy = ServiceLocatorParamsComparePrx::uncheckedCast(
- mLocalAdapter->addWithUUID(configNameCompare));
-
- mConfigCompareGuid = IceUtil::generateUUID();
- mManagement->addCompare(mConfigCompareGuid, configCompareProxy);
- mConfigurationManagement->addLocatorParams(configurationParams, mConfigCompareGuid);
- }
- else if (mStateReplicator)
- {
- ConfigurationReplicatorPrx configurationReplicator = ConfigurationReplicatorPrx::checkedCast(
- mStateReplicator.initialize(), ReplicatorFacet);
- configurationReplicator->registerConfigurationService(mConfigurationServiceProxy);
- }
-
- if (mStateReplicator)
- {
- mReplicatorListener =
- new RtpStateReplicatorListenerI(mGlobalAdapter, rtpmediaservice->getEnvironment(), mGeneralState,
- mConfigurationService);
- mReplicatorListenerProxy =
- RtpStateReplicatorListenerPrx::uncheckedCast(mLocalAdapter->addWithUUID(mReplicatorListener));
-
- if (mCommunicator->getProperties()->getPropertyWithDefault("Rtp.StateReplicatorListener", "no") == "yes")
- {
- mStateReplicator->addListener(mReplicatorListenerProxy);
- mReplicaService->standby();
- lg(Info) << "Operating as a standby replica." << endl;
- }
- else
- {
- lg(Info) << "Operating in an active state." << endl;
- }
- }
-
- ServiceLocatorParamsComparePtr rtpmediacomparatorservice = new RTPMediaServiceCompareServiceImpl(mConfigurationService);
- ServiceLocatorParamsComparePrx RTPMediaComparatorServiceProxy = ServiceLocatorParamsComparePrx::uncheckedCast(
- mGlobalAdapter->add(rtpmediacomparatorservice, mCommunicator->stringToIdentity(MediaComparatorServiceId)));
-
- if (mReplicaService->isActive() == true)
- {
- mGeneralState->comparatorId = IceUtil::generateUUID();
- mManagement->addCompare(mGeneralState->comparatorId, RTPMediaComparatorServiceProxy);
- }
-
-
- RTPMediaServicePrx RTPMediaServiceProxy = RTPMediaServicePrx::uncheckedCast(mGlobalAdapter->add(rtpmediaservice,
- mCommunicator->stringToIdentity(MediaServiceId)));
-
- RTPOverICEServiceLocatorParamsPtr rtpparams = new RTPOverICEServiceLocatorParams;
- rtpparams->category = "rtp";
- PJMediaEnvironmentPtr mediaEnvironment = rtpmediaservice->getEnvironment();
-
- //
- // Service wide configuration is done through properties allowing certain features
- // to be completely disabled.
- //
- NATConfigPtr natConfig = mediaEnvironment->natConfig();
- if (natConfig && natConfig->isSTUNEnabled())
- {
- rtpparams->enableRTPOverICE = true;
- rtpparams->enableTURN = natConfig->isTURNEnabled();
- }
- else
- {
- rtpparams->enableRTPOverICE = false;
- rtpparams->enableTURN = false;
- }
-
- if (mReplicaService->isActive() == true)
- {
- mGeneralState->serviceManagement = ServiceManagementPrx::uncheckedCast(
- mManagement->addService(RTPMediaServiceProxy, "media_rtp_pjmedia"));
- /* Now we can add some parameters to help find us. */
- mGeneralState->serviceManagement->addLocatorParams(rtpparams, mGeneralState->comparatorId);
- }
-
- ServiceLocatorParamsPtr genericparams = new ServiceLocatorParams();
-
- /* One must provide a component service to manage us, if someone wants to */
- ComponentServicePtr ComponentService = new ComponentServicepjmediaImpl(*this, mGeneralState);
- ComponentServicePrx ComponentServiceProxy =
- ComponentServicePrx::uncheckedCast(mLocalAdapter->addWithUUID(ComponentService));
-
- /* Let's add the component service to the service locator first */
- mComponentServiceManagement =
- ServiceManagementPrx::uncheckedCast(mManagement->addService(ComponentServiceProxy, "media_rtp_pjmedia.component"));
- genericparams->category = "Component/media_rtp_pjmedia";
- mComponentServiceManagement->addLocatorParams(genericparams, "");
-
- // Replicate general state information so the backup is ready
- if (mStateReplicator && mReplicaService->isActive() == true)
- {
- RtpStateItemSeq items;
- items.push_back(mGeneralState);
- RtpStateReplicatorPrx oneway = RtpStateReplicatorPrx::uncheckedCast(mStateReplicator->ice_oneway());
- oneway->setState(items);
- }
-}
-
-/**
- * Implementation of the IceBox::Service::stop method.
- */
-void MediaRTPpjmediaApp::stop()
-{
- mComponentServiceManagement->unregister();
- if (mReplicaService->isActive() == true)
- {
- mGeneralState->serviceManagement->unregister();
- }
- if (mConfigurationManagement)
- {
- mConfigurationManagement->unregister();
- }
- if (!mConfigCompareGuid.empty())
- {
- mManagement->removeCompare(mConfigCompareGuid);
- ServiceLocatorManagementPrx management =
- ServiceLocatorManagementPrx::checkedCast(mCommunicator->propertyToProxy("ServiceLocatorManagementProxy"));
- management->removeCompare(mGeneralState->comparatorId);
- }
- mCommunicator->destroy();
-}
-
-extern "C"
-{
-ASTSCF_DLL_EXPORT IceBox::Service* create(Ice::CommunicatorPtr)
-{
- return new MediaRTPpjmediaApp;
-}
-}
diff --git a/src/RTPConfiguration.h b/src/RTPConfiguration.h
index 4ae017c..fcc6e68 100644
--- a/src/RTPConfiguration.h
+++ b/src/RTPConfiguration.h
@@ -42,3 +42,4 @@ protected:
... 1381 lines suppressed ...
--
asterisk-scf/integration/media_rtp_pjmedia.git
More information about the asterisk-scf-commits
mailing list