diff options
Diffstat (limited to 'media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp')
| -rw-r--r-- | media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp | 2377 |
1 files changed, 2377 insertions, 0 deletions
diff --git a/media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp b/media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp new file mode 100644 index 000000000..586876406 --- /dev/null +++ b/media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp @@ -0,0 +1,2377 @@ +/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this file, + * You can obtain one at http://mozilla.org/MPL/2.0/. */ + +// Original author: ekr@rtfm.com + +#include "MediaPipeline.h" + +#ifndef USE_FAKE_MEDIA_STREAMS +#include "MediaStreamGraphImpl.h" +#endif + +#include <math.h> + +#include "nspr.h" +#include "srtp.h" + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) +#include "VideoSegment.h" +#include "Layers.h" +#include "LayersLogging.h" +#include "ImageTypes.h" +#include "ImageContainer.h" +#include "DOMMediaStream.h" +#include "MediaStreamTrack.h" +#include "MediaStreamListener.h" +#include "MediaStreamVideoSink.h" +#include "VideoUtils.h" +#include "VideoStreamTrack.h" +#ifdef WEBRTC_GONK +#include "GrallocImages.h" +#include "mozilla/layers/GrallocTextureClient.h" +#endif +#endif + +#include "nsError.h" +#include "AudioSegment.h" +#include "MediaSegment.h" +#include "MediaPipelineFilter.h" +#include "databuffer.h" +#include "transportflow.h" +#include "transportlayer.h" +#include "transportlayerdtls.h" +#include "transportlayerice.h" +#include "runnable_utils.h" +#include "libyuv/convert.h" +#include "mozilla/SharedThreadPool.h" +#if !defined(MOZILLA_EXTERNAL_LINKAGE) +#include "mozilla/PeerIdentity.h" +#include "mozilla/TaskQueue.h" +#endif +#include "mozilla/gfx/Point.h" +#include "mozilla/gfx/Types.h" +#include "mozilla/UniquePtr.h" +#include "mozilla/UniquePtrExtensions.h" +#include "mozilla/Sprintf.h" + +#include "webrtc/common_types.h" +#include "webrtc/common_video/interface/native_handle.h" +#include "webrtc/common_video/libyuv/include/webrtc_libyuv.h" +#include "webrtc/video_engine/include/vie_errors.h" + +#include "logging.h" + +// Max size given stereo is 480*2*2 = 1920 (10ms of 16-bits stereo audio at +// 48KHz) +#define AUDIO_SAMPLE_BUFFER_MAX 480*2*2 +static_assert((WEBRTC_DEFAULT_SAMPLE_RATE/100)*sizeof(uint16_t) * 2 + <= AUDIO_SAMPLE_BUFFER_MAX, + "AUDIO_SAMPLE_BUFFER_MAX is not large enough"); + +using namespace mozilla; +using namespace mozilla::dom; +using namespace mozilla::gfx; +using namespace mozilla::layers; + +// Logging context +MOZ_MTLOG_MODULE("mediapipeline") + +namespace mozilla { + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) +class VideoConverterListener +{ +public: + NS_INLINE_DECL_THREADSAFE_REFCOUNTING(VideoConverterListener) + + virtual void OnVideoFrameConverted(unsigned char* aVideoFrame, + unsigned int aVideoFrameLength, + unsigned short aWidth, + unsigned short aHeight, + VideoType aVideoType, + uint64_t aCaptureTime) = 0; + + virtual void OnVideoFrameConverted(webrtc::I420VideoFrame& aVideoFrame) = 0; + +protected: + virtual ~VideoConverterListener() {} +}; + +// I420 buffer size macros +#define YSIZE(x,y) ((x)*(y)) +#define CRSIZE(x,y) ((((x)+1) >> 1) * (((y)+1) >> 1)) +#define I420SIZE(x,y) (YSIZE((x),(y)) + 2 * CRSIZE((x),(y))) + +// An async video frame format converter. +// +// Input is typically a MediaStream(Track)Listener driven by MediaStreamGraph. +// +// We keep track of the size of the TaskQueue so we can drop frames if +// conversion is taking too long. +// +// Output is passed through to all added VideoConverterListeners on a TaskQueue +// thread whenever a frame is converted. +class VideoFrameConverter +{ +public: + NS_INLINE_DECL_THREADSAFE_REFCOUNTING(VideoFrameConverter) + + VideoFrameConverter() + : mLength(0) + , last_img_(-1) // -1 is not a guaranteed invalid serial. See bug 1262134. + , disabled_frame_sent_(false) +#ifdef DEBUG + , mThrottleCount(0) + , mThrottleRecord(0) +#endif + , mMutex("VideoFrameConverter") + { + MOZ_COUNT_CTOR(VideoFrameConverter); + + RefPtr<SharedThreadPool> pool = + SharedThreadPool::Get(NS_LITERAL_CSTRING("VideoFrameConverter")); + + mTaskQueue = MakeAndAddRef<TaskQueue>(pool.forget()); + } + + void QueueVideoChunk(VideoChunk& aChunk, bool aForceBlack) + { + if (aChunk.IsNull()) { + return; + } + + // We get passed duplicate frames every ~10ms even with no frame change. + int32_t serial = aChunk.mFrame.GetImage()->GetSerial(); + if (serial == last_img_) { + return; + } + last_img_ = serial; + + // A throttling limit of 1 allows us to convert 2 frames concurrently. + // It's short enough to not build up too significant a delay, while + // giving us a margin to not cause some machines to drop every other frame. + const int32_t queueThrottlingLimit = 1; + if (mLength > queueThrottlingLimit) { + MOZ_MTLOG(ML_DEBUG, "VideoFrameConverter " << this << " queue is full." << + " Throttling by throwing away a frame."); +#ifdef DEBUG + ++mThrottleCount; + mThrottleRecord = std::max(mThrottleCount, mThrottleRecord); +#endif + return; + } + +#ifdef DEBUG + if (mThrottleCount > 0) { + auto level = ML_DEBUG; + if (mThrottleCount > 5) { + // Log at a higher level when we have large drops. + level = ML_INFO; + } + MOZ_MTLOG(level, "VideoFrameConverter " << this << " stopped" << + " throttling after throwing away " << mThrottleCount << + " frames. Longest throttle so far was " << + mThrottleRecord << " frames."); + mThrottleCount = 0; + } +#endif + + bool forceBlack = aForceBlack || aChunk.mFrame.GetForceBlack(); + + if (forceBlack) { + // Reset the last-img check. + // -1 is not a guaranteed invalid serial. See bug 1262134. + last_img_ = -1; + + if (disabled_frame_sent_) { + // After disabling we just pass one black frame to the encoder. + // Allocating and setting it to black steals some performance + // that can be avoided. We don't handle resolution changes while + // disabled for now. + return; + } + + disabled_frame_sent_ = true; + } else { + disabled_frame_sent_ = false; + } + + ++mLength; // Atomic + + nsCOMPtr<nsIRunnable> runnable = + NewRunnableMethod<StorensRefPtrPassByPtr<Image>, bool>( + this, &VideoFrameConverter::ProcessVideoFrame, + aChunk.mFrame.GetImage(), forceBlack); + mTaskQueue->Dispatch(runnable.forget()); + } + + void AddListener(VideoConverterListener* aListener) + { + MutexAutoLock lock(mMutex); + + MOZ_ASSERT(!mListeners.Contains(aListener)); + mListeners.AppendElement(aListener); + } + + bool RemoveListener(VideoConverterListener* aListener) + { + MutexAutoLock lock(mMutex); + + return mListeners.RemoveElement(aListener); + } + + void Shutdown() + { + mTaskQueue->BeginShutdown(); + mTaskQueue->AwaitShutdownAndIdle(); + } + +protected: + virtual ~VideoFrameConverter() + { + MOZ_COUNT_DTOR(VideoFrameConverter); + } + + void VideoFrameConverted(unsigned char* aVideoFrame, + unsigned int aVideoFrameLength, + unsigned short aWidth, + unsigned short aHeight, + VideoType aVideoType, + uint64_t aCaptureTime) + { + MutexAutoLock lock(mMutex); + + for (RefPtr<VideoConverterListener>& listener : mListeners) { + listener->OnVideoFrameConverted(aVideoFrame, aVideoFrameLength, + aWidth, aHeight, aVideoType, aCaptureTime); + } + } + + void VideoFrameConverted(webrtc::I420VideoFrame& aVideoFrame) + { + MutexAutoLock lock(mMutex); + + for (RefPtr<VideoConverterListener>& listener : mListeners) { + listener->OnVideoFrameConverted(aVideoFrame); + } + } + + void ProcessVideoFrame(Image* aImage, bool aForceBlack) + { + --mLength; // Atomic + MOZ_ASSERT(mLength >= 0); + + if (aForceBlack) { + IntSize size = aImage->GetSize(); + uint32_t yPlaneLen = YSIZE(size.width, size.height); + uint32_t cbcrPlaneLen = 2 * CRSIZE(size.width, size.height); + uint32_t length = yPlaneLen + cbcrPlaneLen; + + // Send a black image. + auto pixelData = MakeUniqueFallible<uint8_t[]>(length); + if (pixelData) { + // YCrCb black = 0x10 0x80 0x80 + memset(pixelData.get(), 0x10, yPlaneLen); + // Fill Cb/Cr planes + memset(pixelData.get() + yPlaneLen, 0x80, cbcrPlaneLen); + + MOZ_MTLOG(ML_DEBUG, "Sending a black video frame"); + VideoFrameConverted(pixelData.get(), length, size.width, size.height, + mozilla::kVideoI420, 0); + } + return; + } + + ImageFormat format = aImage->GetFormat(); +#ifdef WEBRTC_GONK + GrallocImage* nativeImage = aImage->AsGrallocImage(); + if (nativeImage) { + android::sp<android::GraphicBuffer> graphicBuffer = nativeImage->GetGraphicBuffer(); + int pixelFormat = graphicBuffer->getPixelFormat(); /* PixelFormat is an enum == int */ + mozilla::VideoType destFormat; + switch (pixelFormat) { + case HAL_PIXEL_FORMAT_YV12: + // all android must support this + destFormat = mozilla::kVideoYV12; + break; + case GrallocImage::HAL_PIXEL_FORMAT_YCbCr_420_SP: + destFormat = mozilla::kVideoNV21; + break; + case GrallocImage::HAL_PIXEL_FORMAT_YCbCr_420_P: + destFormat = mozilla::kVideoI420; + break; + default: + // XXX Bug NNNNNNN + // use http://dxr.mozilla.org/mozilla-central/source/content/media/omx/I420ColorConverterHelper.cpp + // to convert unknown types (OEM-specific) to I420 + MOZ_MTLOG(ML_ERROR, "Un-handled GRALLOC buffer type:" << pixelFormat); + MOZ_CRASH(); + } + void *basePtr; + graphicBuffer->lock(android::GraphicBuffer::USAGE_SW_READ_MASK, &basePtr); + uint32_t width = graphicBuffer->getWidth(); + uint32_t height = graphicBuffer->getHeight(); + // XXX gralloc buffer's width and stride could be different depends on implementations. + + if (destFormat != mozilla::kVideoI420) { + unsigned char *video_frame = static_cast<unsigned char*>(basePtr); + webrtc::I420VideoFrame i420_frame; + int stride_y = width; + int stride_uv = (width + 1) / 2; + int target_width = width; + int target_height = height; + if (i420_frame.CreateEmptyFrame(target_width, + abs(target_height), + stride_y, + stride_uv, stride_uv) < 0) { + MOZ_ASSERT(false, "Can't allocate empty i420frame"); + return; + } + webrtc::VideoType commonVideoType = + webrtc::RawVideoTypeToCommonVideoVideoType( + static_cast<webrtc::RawVideoType>((int)destFormat)); + if (ConvertToI420(commonVideoType, video_frame, 0, 0, width, height, + I420SIZE(width, height), webrtc::kVideoRotation_0, + &i420_frame)) { + MOZ_ASSERT(false, "Can't convert video type for sending to I420"); + return; + } + i420_frame.set_ntp_time_ms(0); + VideoFrameConverted(i420_frame); + } else { + VideoFrameConverted(static_cast<unsigned char*>(basePtr), + I420SIZE(width, height), + width, + height, + destFormat, 0); + } + graphicBuffer->unlock(); + return; + } else +#endif + if (format == ImageFormat::PLANAR_YCBCR) { + // Cast away constness b/c some of the accessors are non-const + PlanarYCbCrImage* yuv = const_cast<PlanarYCbCrImage *>( + static_cast<const PlanarYCbCrImage *>(aImage)); + + const PlanarYCbCrData *data = yuv->GetData(); + if (data) { + uint8_t *y = data->mYChannel; + uint8_t *cb = data->mCbChannel; + uint8_t *cr = data->mCrChannel; + int32_t yStride = data->mYStride; + int32_t cbCrStride = data->mCbCrStride; + uint32_t width = yuv->GetSize().width; + uint32_t height = yuv->GetSize().height; + + webrtc::I420VideoFrame i420_frame; + int rv = i420_frame.CreateFrame(y, cb, cr, width, height, + yStride, cbCrStride, cbCrStride, + webrtc::kVideoRotation_0); + if (rv != 0) { + NS_ERROR("Creating an I420 frame failed"); + return; + } + + MOZ_MTLOG(ML_DEBUG, "Sending an I420 video frame"); + VideoFrameConverted(i420_frame); + return; + } + } + + RefPtr<SourceSurface> surf = aImage->GetAsSourceSurface(); + if (!surf) { + MOZ_MTLOG(ML_ERROR, "Getting surface from " << Stringify(format) << " image failed"); + return; + } + + RefPtr<DataSourceSurface> data = surf->GetDataSurface(); + if (!data) { + MOZ_MTLOG(ML_ERROR, "Getting data surface from " << Stringify(format) + << " image with " << Stringify(surf->GetType()) << "(" + << Stringify(surf->GetFormat()) << ") surface failed"); + return; + } + + IntSize size = aImage->GetSize(); + int half_width = (size.width + 1) >> 1; + int half_height = (size.height + 1) >> 1; + int c_size = half_width * half_height; + int buffer_size = YSIZE(size.width, size.height) + 2 * c_size; + auto yuv_scoped = MakeUniqueFallible<uint8[]>(buffer_size); + if (!yuv_scoped) { + return; + } + uint8* yuv = yuv_scoped.get(); + + DataSourceSurface::ScopedMap map(data, DataSourceSurface::READ); + if (!map.IsMapped()) { + MOZ_MTLOG(ML_ERROR, "Reading DataSourceSurface from " << Stringify(format) + << " image with " << Stringify(surf->GetType()) << "(" + << Stringify(surf->GetFormat()) << ") surface failed"); + return; + } + + int rv; + int cb_offset = YSIZE(size.width, size.height); + int cr_offset = cb_offset + c_size; + switch (surf->GetFormat()) { + case SurfaceFormat::B8G8R8A8: + case SurfaceFormat::B8G8R8X8: + rv = libyuv::ARGBToI420(static_cast<uint8*>(map.GetData()), + map.GetStride(), + yuv, size.width, + yuv + cb_offset, half_width, + yuv + cr_offset, half_width, + size.width, size.height); + break; + case SurfaceFormat::R5G6B5_UINT16: + rv = libyuv::RGB565ToI420(static_cast<uint8*>(map.GetData()), + map.GetStride(), + yuv, size.width, + yuv + cb_offset, half_width, + yuv + cr_offset, half_width, + size.width, size.height); + break; + default: + MOZ_MTLOG(ML_ERROR, "Unsupported RGB video format" << Stringify(surf->GetFormat())); + MOZ_ASSERT(PR_FALSE); + return; + } + if (rv != 0) { + MOZ_MTLOG(ML_ERROR, Stringify(surf->GetFormat()) << " to I420 conversion failed"); + return; + } + MOZ_MTLOG(ML_DEBUG, "Sending an I420 video frame converted from " << + Stringify(surf->GetFormat())); + VideoFrameConverted(yuv, buffer_size, size.width, size.height, mozilla::kVideoI420, 0); + } + + Atomic<int32_t, Relaxed> mLength; + RefPtr<TaskQueue> mTaskQueue; + + // Written and read from the queueing thread (normally MSG). + int32_t last_img_; // serial number of last Image + bool disabled_frame_sent_; // If a black frame has been sent after disabling. +#ifdef DEBUG + uint32_t mThrottleCount; + uint32_t mThrottleRecord; +#endif + + // mMutex guards the below variables. + Mutex mMutex; + nsTArray<RefPtr<VideoConverterListener>> mListeners; +}; +#endif + +// An async inserter for audio data, to avoid running audio codec encoders +// on the MSG/input audio thread. Basically just bounces all the audio +// data to a single audio processing/input queue. We could if we wanted to +// use multiple threads and a TaskQueue. +class AudioProxyThread +{ +public: + NS_INLINE_DECL_THREADSAFE_REFCOUNTING(AudioProxyThread) + + explicit AudioProxyThread(AudioSessionConduit *aConduit) + : mConduit(aConduit) + { + MOZ_ASSERT(mConduit); + MOZ_COUNT_CTOR(AudioProxyThread); + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + // Use only 1 thread; also forces FIFO operation + // We could use multiple threads, but that may be dicier with the webrtc.org + // code. If so we'd need to use TaskQueues like the videoframe converter + RefPtr<SharedThreadPool> pool = + SharedThreadPool::Get(NS_LITERAL_CSTRING("AudioProxy"), 1); + + mThread = pool.get(); +#else + nsCOMPtr<nsIThread> thread; + if (!NS_WARN_IF(NS_FAILED(NS_NewNamedThread("AudioProxy", getter_AddRefs(thread))))) { + mThread = thread; + } +#endif + } + + // called on mThread + void InternalProcessAudioChunk( + TrackRate rate, + AudioChunk& chunk, + bool enabled) { + + // Convert to interleaved, 16-bits integer audio, with a maximum of two + // channels (since the WebRTC.org code below makes the assumption that the + // input audio is either mono or stereo). + uint32_t outputChannels = chunk.ChannelCount() == 1 ? 1 : 2; + const int16_t* samples = nullptr; + UniquePtr<int16_t[]> convertedSamples; + + // We take advantage of the fact that the common case (microphone directly to + // PeerConnection, that is, a normal call), the samples are already 16-bits + // mono, so the representation in interleaved and planar is the same, and we + // can just use that. + if (enabled && outputChannels == 1 && chunk.mBufferFormat == AUDIO_FORMAT_S16) { + samples = chunk.ChannelData<int16_t>().Elements()[0]; + } else { + convertedSamples = MakeUnique<int16_t[]>(chunk.mDuration * outputChannels); + + if (!enabled || chunk.mBufferFormat == AUDIO_FORMAT_SILENCE) { + PodZero(convertedSamples.get(), chunk.mDuration * outputChannels); + } else if (chunk.mBufferFormat == AUDIO_FORMAT_FLOAT32) { + DownmixAndInterleave(chunk.ChannelData<float>(), + chunk.mDuration, chunk.mVolume, outputChannels, + convertedSamples.get()); + } else if (chunk.mBufferFormat == AUDIO_FORMAT_S16) { + DownmixAndInterleave(chunk.ChannelData<int16_t>(), + chunk.mDuration, chunk.mVolume, outputChannels, + convertedSamples.get()); + } + samples = convertedSamples.get(); + } + + MOZ_ASSERT(!(rate%100)); // rate should be a multiple of 100 + + // Check if the rate or the number of channels has changed since the last time + // we came through. I realize it may be overkill to check if the rate has + // changed, but I believe it is possible (e.g. if we change sources) and it + // costs us very little to handle this case. + + uint32_t audio_10ms = rate / 100; + + if (!packetizer_ || + packetizer_->PacketSize() != audio_10ms || + packetizer_->Channels() != outputChannels) { + // It's ok to drop the audio still in the packetizer here. + packetizer_ = new AudioPacketizer<int16_t, int16_t>(audio_10ms, outputChannels); + } + + packetizer_->Input(samples, chunk.mDuration); + + while (packetizer_->PacketsAvailable()) { + uint32_t samplesPerPacket = packetizer_->PacketSize() * + packetizer_->Channels(); + // We know that webrtc.org's code going to copy the samples down the line, + // so we can just use a stack buffer here instead of malloc-ing. + int16_t packet[AUDIO_SAMPLE_BUFFER_MAX]; + + packetizer_->Output(packet); + mConduit->SendAudioFrame(packet, samplesPerPacket, rate, 0); + } + } + + void QueueAudioChunk(TrackRate rate, AudioChunk& chunk, bool enabled) + { + RUN_ON_THREAD(mThread, + WrapRunnable(RefPtr<AudioProxyThread>(this), + &AudioProxyThread::InternalProcessAudioChunk, + rate, chunk, enabled), + NS_DISPATCH_NORMAL); + } + +protected: + virtual ~AudioProxyThread() + { + // Conduits must be released on MainThread, and we might have the last reference + // We don't need to worry about runnables still trying to access the conduit, since + // the runnables hold a ref to AudioProxyThread. + NS_ReleaseOnMainThread(mConduit.forget()); + MOZ_COUNT_DTOR(AudioProxyThread); + } + + RefPtr<AudioSessionConduit> mConduit; + nsCOMPtr<nsIEventTarget> mThread; + // Only accessed on mThread + nsAutoPtr<AudioPacketizer<int16_t, int16_t>> packetizer_; +}; + +static char kDTLSExporterLabel[] = "EXTRACTOR-dtls_srtp"; + +MediaPipeline::MediaPipeline(const std::string& pc, + Direction direction, + nsCOMPtr<nsIEventTarget> main_thread, + nsCOMPtr<nsIEventTarget> sts_thread, + const std::string& track_id, + int level, + RefPtr<MediaSessionConduit> conduit, + RefPtr<TransportFlow> rtp_transport, + RefPtr<TransportFlow> rtcp_transport, + nsAutoPtr<MediaPipelineFilter> filter) + : direction_(direction), + track_id_(track_id), + level_(level), + conduit_(conduit), + rtp_(rtp_transport, rtcp_transport ? RTP : MUX), + rtcp_(rtcp_transport ? rtcp_transport : rtp_transport, + rtcp_transport ? RTCP : MUX), + main_thread_(main_thread), + sts_thread_(sts_thread), + rtp_packets_sent_(0), + rtcp_packets_sent_(0), + rtp_packets_received_(0), + rtcp_packets_received_(0), + rtp_bytes_sent_(0), + rtp_bytes_received_(0), + pc_(pc), + description_(), + filter_(filter), + rtp_parser_(webrtc::RtpHeaderParser::Create()) { + // To indicate rtcp-mux rtcp_transport should be nullptr. + // Therefore it's an error to send in the same flow for + // both rtp and rtcp. + MOZ_ASSERT(rtp_transport != rtcp_transport); + + // PipelineTransport() will access this->sts_thread_; moved here for safety + transport_ = new PipelineTransport(this); +} + +MediaPipeline::~MediaPipeline() { + ASSERT_ON_THREAD(main_thread_); + MOZ_MTLOG(ML_INFO, "Destroying MediaPipeline: " << description_); +} + +nsresult MediaPipeline::Init() { + ASSERT_ON_THREAD(main_thread_); + + if (direction_ == RECEIVE) { + conduit_->SetReceiverTransport(transport_); + } else { + conduit_->SetTransmitterTransport(transport_); + } + + RUN_ON_THREAD(sts_thread_, + WrapRunnable( + RefPtr<MediaPipeline>(this), + &MediaPipeline::Init_s), + NS_DISPATCH_NORMAL); + + return NS_OK; +} + +nsresult MediaPipeline::Init_s() { + ASSERT_ON_THREAD(sts_thread_); + + return AttachTransport_s(); +} + + +// Disconnect us from the transport so that we can cleanly destruct the +// pipeline on the main thread. ShutdownMedia_m() must have already been +// called +void +MediaPipeline::DetachTransport_s() +{ + ASSERT_ON_THREAD(sts_thread_); + + disconnect_all(); + transport_->Detach(); + rtp_.Detach(); + rtcp_.Detach(); +} + +nsresult +MediaPipeline::AttachTransport_s() +{ + ASSERT_ON_THREAD(sts_thread_); + nsresult res; + MOZ_ASSERT(rtp_.transport_); + MOZ_ASSERT(rtcp_.transport_); + res = ConnectTransport_s(rtp_); + if (NS_FAILED(res)) { + return res; + } + + if (rtcp_.transport_ != rtp_.transport_) { + res = ConnectTransport_s(rtcp_); + if (NS_FAILED(res)) { + return res; + } + } + + transport_->Attach(this); + + return NS_OK; +} + +void +MediaPipeline::UpdateTransport_m(int level, + RefPtr<TransportFlow> rtp_transport, + RefPtr<TransportFlow> rtcp_transport, + nsAutoPtr<MediaPipelineFilter> filter) +{ + RUN_ON_THREAD(sts_thread_, + WrapRunnable( + this, + &MediaPipeline::UpdateTransport_s, + level, + rtp_transport, + rtcp_transport, + filter), + NS_DISPATCH_NORMAL); +} + +void +MediaPipeline::UpdateTransport_s(int level, + RefPtr<TransportFlow> rtp_transport, + RefPtr<TransportFlow> rtcp_transport, + nsAutoPtr<MediaPipelineFilter> filter) +{ + bool rtcp_mux = false; + if (!rtcp_transport) { + rtcp_transport = rtp_transport; + rtcp_mux = true; + } + + if ((rtp_transport != rtp_.transport_) || + (rtcp_transport != rtcp_.transport_)) { + DetachTransport_s(); + rtp_ = TransportInfo(rtp_transport, rtcp_mux ? MUX : RTP); + rtcp_ = TransportInfo(rtcp_transport, rtcp_mux ? MUX : RTCP); + AttachTransport_s(); + } + + level_ = level; + + if (filter_ && filter) { + // Use the new filter, but don't forget any remote SSRCs that we've learned + // by receiving traffic. + filter_->Update(*filter); + } else { + filter_ = filter; + } +} + +void +MediaPipeline::SelectSsrc_m(size_t ssrc_index) +{ + RUN_ON_THREAD(sts_thread_, + WrapRunnable( + this, + &MediaPipeline::SelectSsrc_s, + ssrc_index), + NS_DISPATCH_NORMAL); +} + +void +MediaPipeline::SelectSsrc_s(size_t ssrc_index) +{ + filter_ = new MediaPipelineFilter; + if (ssrc_index < ssrcs_received_.size()) { + filter_->AddRemoteSSRC(ssrcs_received_[ssrc_index]); + } else { + MOZ_MTLOG(ML_WARNING, "SelectSsrc called with " << ssrc_index << " but we " + << "have only seen " << ssrcs_received_.size() + << " ssrcs"); + } +} + +void MediaPipeline::StateChange(TransportFlow *flow, TransportLayer::State state) { + TransportInfo* info = GetTransportInfo_s(flow); + MOZ_ASSERT(info); + + if (state == TransportLayer::TS_OPEN) { + MOZ_MTLOG(ML_INFO, "Flow is ready"); + TransportReady_s(*info); + } else if (state == TransportLayer::TS_CLOSED || + state == TransportLayer::TS_ERROR) { + TransportFailed_s(*info); + } +} + +static bool MakeRtpTypeToStringArray(const char** array) { + static const char* RTP_str = "RTP"; + static const char* RTCP_str = "RTCP"; + static const char* MUX_str = "RTP/RTCP mux"; + array[MediaPipeline::RTP] = RTP_str; + array[MediaPipeline::RTCP] = RTCP_str; + array[MediaPipeline::MUX] = MUX_str; + return true; +} + +static const char* ToString(MediaPipeline::RtpType type) { + static const char* array[(int)MediaPipeline::MAX_RTP_TYPE] = {nullptr}; + // Dummy variable to cause init to happen only on first call + static bool dummy = MakeRtpTypeToStringArray(array); + (void)dummy; + return array[type]; +} + +nsresult MediaPipeline::TransportReady_s(TransportInfo &info) { + MOZ_ASSERT(!description_.empty()); + + // TODO(ekr@rtfm.com): implement some kind of notification on + // failure. bug 852665. + if (info.state_ != MP_CONNECTING) { + MOZ_MTLOG(ML_ERROR, "Transport ready for flow in wrong state:" << + description_ << ": " << ToString(info.type_)); + return NS_ERROR_FAILURE; + } + + MOZ_MTLOG(ML_INFO, "Transport ready for pipeline " << + static_cast<void *>(this) << " flow " << description_ << ": " << + ToString(info.type_)); + + // TODO(bcampen@mozilla.com): Should we disconnect from the flow on failure? + nsresult res; + + // Now instantiate the SRTP objects + TransportLayerDtls *dtls = static_cast<TransportLayerDtls *>( + info.transport_->GetLayer(TransportLayerDtls::ID())); + MOZ_ASSERT(dtls); // DTLS is mandatory + + uint16_t cipher_suite; + res = dtls->GetSrtpCipher(&cipher_suite); + if (NS_FAILED(res)) { + MOZ_MTLOG(ML_ERROR, "Failed to negotiate DTLS-SRTP. This is an error"); + info.state_ = MP_CLOSED; + UpdateRtcpMuxState(info); + return res; + } + + // SRTP Key Exporter as per RFC 5764 S 4.2 + unsigned char srtp_block[SRTP_TOTAL_KEY_LENGTH * 2]; + res = dtls->ExportKeyingMaterial(kDTLSExporterLabel, false, "", + srtp_block, sizeof(srtp_block)); + if (NS_FAILED(res)) { + MOZ_MTLOG(ML_ERROR, "Failed to compute DTLS-SRTP keys. This is an error"); + info.state_ = MP_CLOSED; + UpdateRtcpMuxState(info); + MOZ_CRASH(); // TODO: Remove once we have enough field experience to + // know it doesn't happen. bug 798797. Note that the + // code after this never executes. + return res; + } + + // Slice and dice as per RFC 5764 S 4.2 + unsigned char client_write_key[SRTP_TOTAL_KEY_LENGTH]; + unsigned char server_write_key[SRTP_TOTAL_KEY_LENGTH]; + int offset = 0; + memcpy(client_write_key, srtp_block + offset, SRTP_MASTER_KEY_LENGTH); + offset += SRTP_MASTER_KEY_LENGTH; + memcpy(server_write_key, srtp_block + offset, SRTP_MASTER_KEY_LENGTH); + offset += SRTP_MASTER_KEY_LENGTH; + memcpy(client_write_key + SRTP_MASTER_KEY_LENGTH, + srtp_block + offset, SRTP_MASTER_SALT_LENGTH); + offset += SRTP_MASTER_SALT_LENGTH; + memcpy(server_write_key + SRTP_MASTER_KEY_LENGTH, + srtp_block + offset, SRTP_MASTER_SALT_LENGTH); + offset += SRTP_MASTER_SALT_LENGTH; + MOZ_ASSERT(offset == sizeof(srtp_block)); + + unsigned char *write_key; + unsigned char *read_key; + + if (dtls->role() == TransportLayerDtls::CLIENT) { + write_key = client_write_key; + read_key = server_write_key; + } else { + write_key = server_write_key; + read_key = client_write_key; + } + + MOZ_ASSERT(!info.send_srtp_ && !info.recv_srtp_); + info.send_srtp_ = SrtpFlow::Create(cipher_suite, false, write_key, + SRTP_TOTAL_KEY_LENGTH); + info.recv_srtp_ = SrtpFlow::Create(cipher_suite, true, read_key, + SRTP_TOTAL_KEY_LENGTH); + if (!info.send_srtp_ || !info.recv_srtp_) { + MOZ_MTLOG(ML_ERROR, "Couldn't create SRTP flow for " + << ToString(info.type_)); + info.state_ = MP_CLOSED; + UpdateRtcpMuxState(info); + return NS_ERROR_FAILURE; + } + + MOZ_MTLOG(ML_INFO, "Listening for " << ToString(info.type_) + << " packets received on " << + static_cast<void *>(dtls->downward())); + + switch (info.type_) { + case RTP: + dtls->downward()->SignalPacketReceived.connect( + this, + &MediaPipeline::RtpPacketReceived); + break; + case RTCP: + dtls->downward()->SignalPacketReceived.connect( + this, + &MediaPipeline::RtcpPacketReceived); + break; + case MUX: + dtls->downward()->SignalPacketReceived.connect( + this, + &MediaPipeline::PacketReceived); + break; + default: + MOZ_CRASH(); + } + + info.state_ = MP_OPEN; + UpdateRtcpMuxState(info); + return NS_OK; +} + +nsresult MediaPipeline::TransportFailed_s(TransportInfo &info) { + ASSERT_ON_THREAD(sts_thread_); + + info.state_ = MP_CLOSED; + UpdateRtcpMuxState(info); + + MOZ_MTLOG(ML_INFO, "Transport closed for flow " << ToString(info.type_)); + + NS_WARNING( + "MediaPipeline Transport failed. This is not properly cleaned up yet"); + + // TODO(ekr@rtfm.com): SECURITY: Figure out how to clean up if the + // connection was good and now it is bad. + // TODO(ekr@rtfm.com): Report up so that the PC knows we + // have experienced an error. + + return NS_OK; +} + +void MediaPipeline::UpdateRtcpMuxState(TransportInfo &info) { + if (info.type_ == MUX) { + if (info.transport_ == rtcp_.transport_) { + rtcp_.state_ = info.state_; + if (!rtcp_.send_srtp_) { + rtcp_.send_srtp_ = info.send_srtp_; + rtcp_.recv_srtp_ = info.recv_srtp_; + } + } + } +} + +nsresult MediaPipeline::SendPacket(TransportFlow *flow, const void *data, + int len) { + ASSERT_ON_THREAD(sts_thread_); + + // Note that we bypass the DTLS layer here + TransportLayerDtls *dtls = static_cast<TransportLayerDtls *>( + flow->GetLayer(TransportLayerDtls::ID())); + MOZ_ASSERT(dtls); + + TransportResult res = dtls->downward()-> + SendPacket(static_cast<const unsigned char *>(data), len); + + if (res != len) { + // Ignore blocking indications + if (res == TE_WOULDBLOCK) + return NS_OK; + + MOZ_MTLOG(ML_ERROR, "Failed write on stream " << description_); + return NS_BASE_STREAM_CLOSED; + } + + return NS_OK; +} + +void MediaPipeline::increment_rtp_packets_sent(int32_t bytes) { + ++rtp_packets_sent_; + rtp_bytes_sent_ += bytes; + + if (!(rtp_packets_sent_ % 100)) { + MOZ_MTLOG(ML_INFO, "RTP sent packet count for " << description_ + << " Pipeline " << static_cast<void *>(this) + << " Flow : " << static_cast<void *>(rtp_.transport_) + << ": " << rtp_packets_sent_ + << " (" << rtp_bytes_sent_ << " bytes)"); + } +} + +void MediaPipeline::increment_rtcp_packets_sent() { + ++rtcp_packets_sent_; + if (!(rtcp_packets_sent_ % 100)) { + MOZ_MTLOG(ML_INFO, "RTCP sent packet count for " << description_ + << " Pipeline " << static_cast<void *>(this) + << " Flow : " << static_cast<void *>(rtcp_.transport_) + << ": " << rtcp_packets_sent_); + } +} + +void MediaPipeline::increment_rtp_packets_received(int32_t bytes) { + ++rtp_packets_received_; + rtp_bytes_received_ += bytes; + if (!(rtp_packets_received_ % 100)) { + MOZ_MTLOG(ML_INFO, "RTP received packet count for " << description_ + << " Pipeline " << static_cast<void *>(this) + << " Flow : " << static_cast<void *>(rtp_.transport_) + << ": " << rtp_packets_received_ + << " (" << rtp_bytes_received_ << " bytes)"); + } +} + +void MediaPipeline::increment_rtcp_packets_received() { + ++rtcp_packets_received_; + if (!(rtcp_packets_received_ % 100)) { + MOZ_MTLOG(ML_INFO, "RTCP received packet count for " << description_ + << " Pipeline " << static_cast<void *>(this) + << " Flow : " << static_cast<void *>(rtcp_.transport_) + << ": " << rtcp_packets_received_); + } +} + +void MediaPipeline::RtpPacketReceived(TransportLayer *layer, + const unsigned char *data, + size_t len) { + if (!transport_->pipeline()) { + MOZ_MTLOG(ML_ERROR, "Discarding incoming packet; transport disconnected"); + return; + } + + if (!conduit_) { + MOZ_MTLOG(ML_DEBUG, "Discarding incoming packet; media disconnected"); + return; + } + + if (rtp_.state_ != MP_OPEN) { + MOZ_MTLOG(ML_ERROR, "Discarding incoming packet; pipeline not open"); + return; + } + + if (rtp_.transport_->state() != TransportLayer::TS_OPEN) { + MOZ_MTLOG(ML_ERROR, "Discarding incoming packet; transport not open"); + return; + } + + // This should never happen. + MOZ_ASSERT(rtp_.recv_srtp_); + + if (direction_ == TRANSMIT) { + return; + } + + if (!len) { + return; + } + + // Filter out everything but RTP/RTCP + if (data[0] < 128 || data[0] > 191) { + return; + } + + webrtc::RTPHeader header; + if (!rtp_parser_->Parse(data, len, &header)) { + return; + } + + if (std::find(ssrcs_received_.begin(), ssrcs_received_.end(), header.ssrc) == + ssrcs_received_.end()) { + ssrcs_received_.push_back(header.ssrc); + } + + if (filter_ && !filter_->Filter(header)) { + return; + } + + // Make a copy rather than cast away constness + auto inner_data = MakeUnique<unsigned char[]>(len); + memcpy(inner_data.get(), data, len); + int out_len = 0; + nsresult res = rtp_.recv_srtp_->UnprotectRtp(inner_data.get(), + len, len, &out_len); + if (!NS_SUCCEEDED(res)) { + char tmp[16]; + + SprintfLiteral(tmp, "%.2x %.2x %.2x %.2x", + inner_data[0], + inner_data[1], + inner_data[2], + inner_data[3]); + + MOZ_MTLOG(ML_NOTICE, "Error unprotecting RTP in " << description_ + << "len= " << len << "[" << tmp << "...]"); + + return; + } + MOZ_MTLOG(ML_DEBUG, description_ << " received RTP packet."); + increment_rtp_packets_received(out_len); + + (void)conduit_->ReceivedRTPPacket(inner_data.get(), out_len); // Ignore error codes +} + +void MediaPipeline::RtcpPacketReceived(TransportLayer *layer, + const unsigned char *data, + size_t len) { + if (!transport_->pipeline()) { + MOZ_MTLOG(ML_DEBUG, "Discarding incoming packet; transport disconnected"); + return; + } + + if (!conduit_) { + MOZ_MTLOG(ML_DEBUG, "Discarding incoming packet; media disconnected"); + return; + } + + if (rtcp_.state_ != MP_OPEN) { + MOZ_MTLOG(ML_DEBUG, "Discarding incoming packet; pipeline not open"); + return; + } + + if (rtcp_.transport_->state() != TransportLayer::TS_OPEN) { + MOZ_MTLOG(ML_ERROR, "Discarding incoming packet; transport not open"); + return; + } + + if (!len) { + return; + } + + // Filter out everything but RTP/RTCP + if (data[0] < 128 || data[0] > 191) { + return; + } + + // We do not filter RTCP for send pipelines, since the webrtc.org code for + // senders already has logic to ignore RRs that do not apply. + // TODO bug 1279153: remove SR check for reduced size RTCP + if (filter_ && direction_ == RECEIVE) { + if (!filter_->FilterSenderReport(data, len)) { + MOZ_MTLOG(ML_NOTICE, "Dropping incoming RTCP packet; filtered out"); + return; + } + } + + // Make a copy rather than cast away constness + auto inner_data = MakeUnique<unsigned char[]>(len); + memcpy(inner_data.get(), data, len); + int out_len; + + nsresult res = rtcp_.recv_srtp_->UnprotectRtcp(inner_data.get(), + len, + len, + &out_len); + + if (!NS_SUCCEEDED(res)) + return; + + MOZ_MTLOG(ML_DEBUG, description_ << " received RTCP packet."); + increment_rtcp_packets_received(); + + MOZ_ASSERT(rtcp_.recv_srtp_); // This should never happen + + (void)conduit_->ReceivedRTCPPacket(inner_data.get(), out_len); // Ignore error codes +} + +bool MediaPipeline::IsRtp(const unsigned char *data, size_t len) { + if (len < 2) + return false; + + // Check if this is a RTCP packet. Logic based on the types listed in + // media/webrtc/trunk/src/modules/rtp_rtcp/source/rtp_utility.cc + + // Anything outside this range is RTP. + if ((data[1] < 192) || (data[1] > 207)) + return true; + + if (data[1] == 192) // FIR + return false; + + if (data[1] == 193) // NACK, but could also be RTP. This makes us sad + return true; // but it's how webrtc.org behaves. + + if (data[1] == 194) + return true; + + if (data[1] == 195) // IJ. + return false; + + if ((data[1] > 195) && (data[1] < 200)) // the > 195 is redundant + return true; + + if ((data[1] >= 200) && (data[1] <= 207)) // SR, RR, SDES, BYE, + return false; // APP, RTPFB, PSFB, XR + + MOZ_ASSERT(false); // Not reached, belt and suspenders. + return true; +} + +void MediaPipeline::PacketReceived(TransportLayer *layer, + const unsigned char *data, + size_t len) { + if (!transport_->pipeline()) { + MOZ_MTLOG(ML_DEBUG, "Discarding incoming packet; transport disconnected"); + return; + } + + if (IsRtp(data, len)) { + RtpPacketReceived(layer, data, len); + } else { + RtcpPacketReceived(layer, data, len); + } +} + +class MediaPipelineTransmit::PipelineListener + : public DirectMediaStreamTrackListener +{ +friend class MediaPipelineTransmit; +public: + explicit PipelineListener(const RefPtr<MediaSessionConduit>& conduit) + : conduit_(conduit), + track_id_(TRACK_INVALID), + mMutex("MediaPipelineTransmit::PipelineListener"), + track_id_external_(TRACK_INVALID), + active_(false), + enabled_(false), + direct_connect_(false) + { + } + + ~PipelineListener() + { + if (!NS_IsMainThread()) { + // release conduit on mainthread. Must use forget()! + nsresult rv = NS_DispatchToMainThread(new + ConduitDeleteEvent(conduit_.forget())); + MOZ_ASSERT(!NS_FAILED(rv),"Could not dispatch conduit shutdown to main"); + if (NS_FAILED(rv)) { + MOZ_CRASH(); + } + } else { + conduit_ = nullptr; + } +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + if (converter_) { + converter_->Shutdown(); + } +#endif + } + + // Dispatches setting the internal TrackID to TRACK_INVALID to the media + // graph thread to keep it in sync with other MediaStreamGraph operations + // like RemoveListener() and AddListener(). The TrackID will be updated on + // the next NewData() callback. + void UnsetTrackId(MediaStreamGraphImpl* graph); + + void SetActive(bool active) { active_ = active; } + void SetEnabled(bool enabled) { enabled_ = enabled; } + + // These are needed since nested classes don't have access to any particular + // instance of the parent + void SetAudioProxy(const RefPtr<AudioProxyThread>& proxy) + { + audio_processing_ = proxy; + } + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + void SetVideoFrameConverter(const RefPtr<VideoFrameConverter>& converter) + { + converter_ = converter; + } + + void OnVideoFrameConverted(unsigned char* aVideoFrame, + unsigned int aVideoFrameLength, + unsigned short aWidth, + unsigned short aHeight, + VideoType aVideoType, + uint64_t aCaptureTime) + { + MOZ_ASSERT(conduit_->type() == MediaSessionConduit::VIDEO); + static_cast<VideoSessionConduit*>(conduit_.get())->SendVideoFrame( + aVideoFrame, aVideoFrameLength, aWidth, aHeight, aVideoType, aCaptureTime); + } + + void OnVideoFrameConverted(webrtc::I420VideoFrame& aVideoFrame) + { + MOZ_ASSERT(conduit_->type() == MediaSessionConduit::VIDEO); + static_cast<VideoSessionConduit*>(conduit_.get())->SendVideoFrame(aVideoFrame); + } +#endif + + // Implement MediaStreamTrackListener + void NotifyQueuedChanges(MediaStreamGraph* aGraph, + StreamTime aTrackOffset, + const MediaSegment& aQueuedMedia) override; + + // Implement DirectMediaStreamTrackListener + void NotifyRealtimeTrackData(MediaStreamGraph* aGraph, + StreamTime aTrackOffset, + const MediaSegment& aMedia) override; + void NotifyDirectListenerInstalled(InstallationResult aResult) override; + void NotifyDirectListenerUninstalled() override; + +private: + void UnsetTrackIdImpl() { + MutexAutoLock lock(mMutex); + track_id_ = track_id_external_ = TRACK_INVALID; + } + + void NewData(MediaStreamGraph* graph, + StreamTime offset, + const MediaSegment& media); + + RefPtr<MediaSessionConduit> conduit_; + RefPtr<AudioProxyThread> audio_processing_; +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + RefPtr<VideoFrameConverter> converter_; +#endif + + // May be TRACK_INVALID until we see data from the track + TrackID track_id_; // this is the current TrackID this listener is attached to + Mutex mMutex; + // protected by mMutex + // May be TRACK_INVALID until we see data from the track + TrackID track_id_external_; // this is queried from other threads + + // active is true if there is a transport to send on + mozilla::Atomic<bool> active_; + // enabled is true if the media access control permits sending + // actual content; when false you get black/silence + mozilla::Atomic<bool> enabled_; + + // Written and read on the MediaStreamGraph thread + bool direct_connect_; +}; + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) +// Implements VideoConverterListener for MediaPipeline. +// +// We pass converted frames on to MediaPipelineTransmit::PipelineListener +// where they are further forwarded to VideoConduit. +// MediaPipelineTransmit calls Detach() during shutdown to ensure there is +// no cyclic dependencies between us and PipelineListener. +class MediaPipelineTransmit::VideoFrameFeeder + : public VideoConverterListener +{ +public: + explicit VideoFrameFeeder(const RefPtr<PipelineListener>& listener) + : listener_(listener), + mutex_("VideoFrameFeeder") + { + MOZ_COUNT_CTOR(VideoFrameFeeder); + } + + void Detach() + { + MutexAutoLock lock(mutex_); + + listener_ = nullptr; + } + + void OnVideoFrameConverted(unsigned char* aVideoFrame, + unsigned int aVideoFrameLength, + unsigned short aWidth, + unsigned short aHeight, + VideoType aVideoType, + uint64_t aCaptureTime) override + { + MutexAutoLock lock(mutex_); + + if (!listener_) { + return; + } + + listener_->OnVideoFrameConverted(aVideoFrame, aVideoFrameLength, + aWidth, aHeight, aVideoType, aCaptureTime); + } + + void OnVideoFrameConverted(webrtc::I420VideoFrame& aVideoFrame) override + { + MutexAutoLock lock(mutex_); + + if (!listener_) { + return; + } + + listener_->OnVideoFrameConverted(aVideoFrame); + } + +protected: + virtual ~VideoFrameFeeder() + { + MOZ_COUNT_DTOR(VideoFrameFeeder); + } + + RefPtr<PipelineListener> listener_; + Mutex mutex_; +}; +#endif + +class MediaPipelineTransmit::PipelineVideoSink : + public MediaStreamVideoSink +{ +public: + explicit PipelineVideoSink(const RefPtr<MediaSessionConduit>& conduit, + MediaPipelineTransmit::PipelineListener* listener) + : conduit_(conduit) + , pipelineListener_(listener) + { + } + + virtual void SetCurrentFrames(const VideoSegment& aSegment) override; + virtual void ClearFrames() override {} + +private: + ~PipelineVideoSink() { + // release conduit on mainthread. Must use forget()! + nsresult rv = NS_DispatchToMainThread(new + ConduitDeleteEvent(conduit_.forget())); + MOZ_ASSERT(!NS_FAILED(rv),"Could not dispatch conduit shutdown to main"); + if (NS_FAILED(rv)) { + MOZ_CRASH(); + } + } + RefPtr<MediaSessionConduit> conduit_; + MediaPipelineTransmit::PipelineListener* pipelineListener_; +}; + +MediaPipelineTransmit::MediaPipelineTransmit( + const std::string& pc, + nsCOMPtr<nsIEventTarget> main_thread, + nsCOMPtr<nsIEventTarget> sts_thread, + dom::MediaStreamTrack* domtrack, + const std::string& track_id, + int level, + RefPtr<MediaSessionConduit> conduit, + RefPtr<TransportFlow> rtp_transport, + RefPtr<TransportFlow> rtcp_transport, + nsAutoPtr<MediaPipelineFilter> filter) : + MediaPipeline(pc, TRANSMIT, main_thread, sts_thread, track_id, level, + conduit, rtp_transport, rtcp_transport, filter), + listener_(new PipelineListener(conduit)), + video_sink_(new PipelineVideoSink(conduit, listener_)), + domtrack_(domtrack) +{ + if (!IsVideo()) { + audio_processing_ = MakeAndAddRef<AudioProxyThread>(static_cast<AudioSessionConduit*>(conduit.get())); + listener_->SetAudioProxy(audio_processing_); + } +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + else { // Video + // For video we send frames to an async VideoFrameConverter that calls + // back to a VideoFrameFeeder that feeds I420 frames to VideoConduit. + + feeder_ = MakeAndAddRef<VideoFrameFeeder>(listener_); + + converter_ = MakeAndAddRef<VideoFrameConverter>(); + converter_->AddListener(feeder_); + + listener_->SetVideoFrameConverter(converter_); + } +#endif +} + +MediaPipelineTransmit::~MediaPipelineTransmit() +{ +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + if (feeder_) { + feeder_->Detach(); + } +#endif +} + +nsresult MediaPipelineTransmit::Init() { + AttachToTrack(track_id_); + + return MediaPipeline::Init(); +} + +void MediaPipelineTransmit::AttachToTrack(const std::string& track_id) { + ASSERT_ON_THREAD(main_thread_); + + description_ = pc_ + "| "; + description_ += conduit_->type() == MediaSessionConduit::AUDIO ? + "Transmit audio[" : "Transmit video["; + description_ += track_id; + description_ += "]"; + + // TODO(ekr@rtfm.com): Check for errors + MOZ_MTLOG(ML_DEBUG, "Attaching pipeline to track " + << static_cast<void *>(domtrack_) << " conduit type=" << + (conduit_->type() == MediaSessionConduit::AUDIO ?"audio":"video")); + + // Register the Listener directly with the source if we can. + // We also register it as a non-direct listener so we fall back to that + // if installing the direct listener fails. As a direct listener we get access + // to direct unqueued (and not resampled) data. + domtrack_->AddDirectListener(listener_); + domtrack_->AddListener(listener_); + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + domtrack_->AddDirectListener(video_sink_); +#endif + +#ifndef MOZILLA_INTERNAL_API + // this enables the unit tests that can't fiddle with principals and the like + listener_->SetEnabled(true); +#endif +} + +bool +MediaPipelineTransmit::IsVideo() const +{ + return !!domtrack_->AsVideoStreamTrack(); +} + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) +void MediaPipelineTransmit::UpdateSinkIdentity_m(MediaStreamTrack* track, + nsIPrincipal* principal, + const PeerIdentity* sinkIdentity) { + ASSERT_ON_THREAD(main_thread_); + + if (track != nullptr && track != domtrack_) { + // If a track is specified, then it might not be for this pipeline, + // since we receive notifications for all tracks on the PC. + // nullptr means that the PeerIdentity has changed and shall be applied + // to all tracks of the PC. + return; + } + + bool enableTrack = principal->Subsumes(domtrack_->GetPrincipal()); + if (!enableTrack) { + // first try didn't work, but there's a chance that this is still available + // if our track is bound to a peerIdentity, and the peer connection (our + // sink) is bound to the same identity, then we can enable the track. + const PeerIdentity* trackIdentity = domtrack_->GetPeerIdentity(); + if (sinkIdentity && trackIdentity) { + enableTrack = (*sinkIdentity == *trackIdentity); + } + } + + listener_->SetEnabled(enableTrack); +} +#endif + +void +MediaPipelineTransmit::DetachMedia() +{ + ASSERT_ON_THREAD(main_thread_); + if (domtrack_) { + domtrack_->RemoveDirectListener(listener_); + domtrack_->RemoveListener(listener_); + domtrack_->RemoveDirectListener(video_sink_); + domtrack_ = nullptr; + } + // Let the listener be destroyed with the pipeline (or later). +} + +nsresult MediaPipelineTransmit::TransportReady_s(TransportInfo &info) { + ASSERT_ON_THREAD(sts_thread_); + // Call base ready function. + MediaPipeline::TransportReady_s(info); + + // Should not be set for a transmitter + if (&info == &rtp_) { + listener_->SetActive(true); + } + + return NS_OK; +} + +nsresult MediaPipelineTransmit::ReplaceTrack(MediaStreamTrack& domtrack) { + // MainThread, checked in calls we make +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + nsString nsTrackId; + domtrack.GetId(nsTrackId); + std::string track_id(NS_ConvertUTF16toUTF8(nsTrackId).get()); +#else + std::string track_id = domtrack.GetId(); +#endif + MOZ_MTLOG(ML_DEBUG, "Reattaching pipeline " << description_ << " to track " + << static_cast<void *>(&domtrack) + << " track " << track_id << " conduit type=" << + (conduit_->type() == MediaSessionConduit::AUDIO ?"audio":"video")); + + DetachMedia(); + domtrack_ = &domtrack; // Detach clears it + // Unsets the track id after RemoveListener() takes effect. + listener_->UnsetTrackId(domtrack_->GraphImpl()); + track_id_ = track_id; + AttachToTrack(track_id); + return NS_OK; +} + +void MediaPipeline::DisconnectTransport_s(TransportInfo &info) { + MOZ_ASSERT(info.transport_); + ASSERT_ON_THREAD(sts_thread_); + + info.transport_->SignalStateChange.disconnect(this); + // We do this even if we're a transmitter, since we are still possibly + // registered to receive RTCP. + TransportLayerDtls *dtls = static_cast<TransportLayerDtls *>( + info.transport_->GetLayer(TransportLayerDtls::ID())); + MOZ_ASSERT(dtls); // DTLS is mandatory + MOZ_ASSERT(dtls->downward()); + dtls->downward()->SignalPacketReceived.disconnect(this); +} + +nsresult MediaPipeline::ConnectTransport_s(TransportInfo &info) { + MOZ_ASSERT(info.transport_); + ASSERT_ON_THREAD(sts_thread_); + + // Look to see if the transport is ready + if (info.transport_->state() == TransportLayer::TS_OPEN) { + nsresult res = TransportReady_s(info); + if (NS_FAILED(res)) { + MOZ_MTLOG(ML_ERROR, "Error calling TransportReady(); res=" + << static_cast<uint32_t>(res) << " in " << __FUNCTION__); + return res; + } + } else if (info.transport_->state() == TransportLayer::TS_ERROR) { + MOZ_MTLOG(ML_ERROR, ToString(info.type_) + << "transport is already in error state"); + TransportFailed_s(info); + return NS_ERROR_FAILURE; + } + + info.transport_->SignalStateChange.connect(this, + &MediaPipeline::StateChange); + + return NS_OK; +} + +MediaPipeline::TransportInfo* MediaPipeline::GetTransportInfo_s( + TransportFlow *flow) { + ASSERT_ON_THREAD(sts_thread_); + if (flow == rtp_.transport_) { + return &rtp_; + } + + if (flow == rtcp_.transport_) { + return &rtcp_; + } + + return nullptr; +} + +nsresult MediaPipeline::PipelineTransport::SendRtpPacket( + const void *data, int len) { + + nsAutoPtr<DataBuffer> buf(new DataBuffer(static_cast<const uint8_t *>(data), + len, len + SRTP_MAX_EXPANSION)); + + RUN_ON_THREAD(sts_thread_, + WrapRunnable( + RefPtr<MediaPipeline::PipelineTransport>(this), + &MediaPipeline::PipelineTransport::SendRtpRtcpPacket_s, + buf, true), + NS_DISPATCH_NORMAL); + + return NS_OK; +} + +nsresult MediaPipeline::PipelineTransport::SendRtpRtcpPacket_s( + nsAutoPtr<DataBuffer> data, + bool is_rtp) { + + ASSERT_ON_THREAD(sts_thread_); + if (!pipeline_) { + return NS_OK; // Detached + } + TransportInfo& transport = is_rtp ? pipeline_->rtp_ : pipeline_->rtcp_; + + if (!transport.send_srtp_) { + MOZ_MTLOG(ML_DEBUG, "Couldn't write RTP/RTCP packet; SRTP not set up yet"); + return NS_OK; + } + + MOZ_ASSERT(transport.transport_); + NS_ENSURE_TRUE(transport.transport_, NS_ERROR_NULL_POINTER); + + // libsrtp enciphers in place, so we need a big enough buffer. + MOZ_ASSERT(data->capacity() >= data->len() + SRTP_MAX_EXPANSION); + + int out_len; + nsresult res; + if (is_rtp) { + res = transport.send_srtp_->ProtectRtp(data->data(), + data->len(), + data->capacity(), + &out_len); + } else { + res = transport.send_srtp_->ProtectRtcp(data->data(), + data->len(), + data->capacity(), + &out_len); + } + if (!NS_SUCCEEDED(res)) { + return res; + } + + // paranoia; don't have uninitialized bytes included in data->len() + data->SetLength(out_len); + + MOZ_MTLOG(ML_DEBUG, pipeline_->description_ << " sending " << + (is_rtp ? "RTP" : "RTCP") << " packet"); + if (is_rtp) { + pipeline_->increment_rtp_packets_sent(out_len); + } else { + pipeline_->increment_rtcp_packets_sent(); + } + return pipeline_->SendPacket(transport.transport_, data->data(), out_len); +} + +nsresult MediaPipeline::PipelineTransport::SendRtcpPacket( + const void *data, int len) { + + nsAutoPtr<DataBuffer> buf(new DataBuffer(static_cast<const uint8_t *>(data), + len, len + SRTP_MAX_EXPANSION)); + + RUN_ON_THREAD(sts_thread_, + WrapRunnable( + RefPtr<MediaPipeline::PipelineTransport>(this), + &MediaPipeline::PipelineTransport::SendRtpRtcpPacket_s, + buf, false), + NS_DISPATCH_NORMAL); + + return NS_OK; +} + +void MediaPipelineTransmit::PipelineListener:: +UnsetTrackId(MediaStreamGraphImpl* graph) { +#ifndef USE_FAKE_MEDIA_STREAMS + class Message : public ControlMessage { + public: + explicit Message(PipelineListener* listener) : + ControlMessage(nullptr), listener_(listener) {} + virtual void Run() override + { + listener_->UnsetTrackIdImpl(); + } + RefPtr<PipelineListener> listener_; + }; + graph->AppendMessage(MakeUnique<Message>(this)); +#else + UnsetTrackIdImpl(); +#endif +} +// Called if we're attached with AddDirectListener() +void MediaPipelineTransmit::PipelineListener:: +NotifyRealtimeTrackData(MediaStreamGraph* graph, + StreamTime offset, + const MediaSegment& media) { + MOZ_MTLOG(ML_DEBUG, "MediaPipeline::NotifyRealtimeTrackData() listener=" << + this << ", offset=" << offset << + ", duration=" << media.GetDuration()); + + NewData(graph, offset, media); +} + +void MediaPipelineTransmit::PipelineListener:: +NotifyQueuedChanges(MediaStreamGraph* graph, + StreamTime offset, + const MediaSegment& queued_media) { + MOZ_MTLOG(ML_DEBUG, "MediaPipeline::NotifyQueuedChanges()"); + + // ignore non-direct data if we're also getting direct data + if (!direct_connect_) { + NewData(graph, offset, queued_media); + } +} + +void MediaPipelineTransmit::PipelineListener:: +NotifyDirectListenerInstalled(InstallationResult aResult) { + MOZ_MTLOG(ML_INFO, "MediaPipeline::NotifyDirectListenerInstalled() listener= " << + this << ", result=" << static_cast<int32_t>(aResult)); + + direct_connect_ = InstallationResult::SUCCESS == aResult; +} + +void MediaPipelineTransmit::PipelineListener:: +NotifyDirectListenerUninstalled() { + MOZ_MTLOG(ML_INFO, "MediaPipeline::NotifyDirectListenerUninstalled() listener=" << this); + + direct_connect_ = false; +} + +void MediaPipelineTransmit::PipelineListener:: +NewData(MediaStreamGraph* graph, + StreamTime offset, + const MediaSegment& media) { + if (!active_) { + MOZ_MTLOG(ML_DEBUG, "Discarding packets because transport not ready"); + return; + } + + if (conduit_->type() != + (media.GetType() == MediaSegment::AUDIO ? MediaSessionConduit::AUDIO : + MediaSessionConduit::VIDEO)) { + MOZ_ASSERT(false, "The media type should always be correct since the " + "listener is locked to a specific track"); + return; + } + + // TODO(ekr@rtfm.com): For now assume that we have only one + // track type and it's destined for us + // See bug 784517 + if (media.GetType() == MediaSegment::AUDIO) { + AudioSegment* audio = const_cast<AudioSegment *>( + static_cast<const AudioSegment *>(&media)); + + AudioSegment::ChunkIterator iter(*audio); + while(!iter.IsEnded()) { + TrackRate rate; +#ifdef USE_FAKE_MEDIA_STREAMS + rate = Fake_MediaStream::GraphRate(); +#else + rate = graph->GraphRate(); +#endif + audio_processing_->QueueAudioChunk(rate, *iter, enabled_); + iter.Next(); + } + } else { + // Ignore + } +} + +void MediaPipelineTransmit::PipelineVideoSink:: +SetCurrentFrames(const VideoSegment& aSegment) +{ + MOZ_ASSERT(pipelineListener_); + + if (!pipelineListener_->active_) { + MOZ_MTLOG(ML_DEBUG, "Discarding packets because transport not ready"); + return; + } + + if (conduit_->type() != MediaSessionConduit::VIDEO) { + // Ignore data of wrong kind in case we have a muxed stream + return; + } + +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + VideoSegment* video = const_cast<VideoSegment *>(&aSegment); + + VideoSegment::ChunkIterator iter(*video); + while(!iter.IsEnded()) { + pipelineListener_->converter_->QueueVideoChunk(*iter, !pipelineListener_->enabled_); + iter.Next(); + } +#endif +} + +class TrackAddedCallback { + public: + virtual void TrackAdded(TrackTicks current_ticks) = 0; + + NS_INLINE_DECL_THREADSAFE_REFCOUNTING(TrackAddedCallback); + + protected: + virtual ~TrackAddedCallback() {} +}; + +class GenericReceiveListener; + +class GenericReceiveCallback : public TrackAddedCallback +{ + public: + explicit GenericReceiveCallback(GenericReceiveListener* listener) + : listener_(listener) {} + + void TrackAdded(TrackTicks time); + + private: + RefPtr<GenericReceiveListener> listener_; +}; + +// Add a listener on the MSG thread using the MSG command queue +static void AddListener(MediaStream* source, MediaStreamListener* listener) { +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + class Message : public ControlMessage { + public: + Message(MediaStream* stream, MediaStreamListener* listener) + : ControlMessage(stream), + listener_(listener) {} + + virtual void Run() override { + mStream->AddListenerImpl(listener_.forget()); + } + private: + RefPtr<MediaStreamListener> listener_; + }; + + MOZ_ASSERT(listener); + + source->GraphImpl()->AppendMessage(MakeUnique<Message>(source, listener)); +#else + source->AddListener(listener); +#endif +} + +class GenericReceiveListener : public MediaStreamListener +{ + public: + GenericReceiveListener(SourceMediaStream *source, TrackID track_id) + : source_(source), + track_id_(track_id), + played_ticks_(0), + principal_handle_(PRINCIPAL_HANDLE_NONE) {} + + virtual ~GenericReceiveListener() {} + + void AddSelf() + { + AddListener(source_, this); + } + + void EndTrack() + { + source_->EndTrack(track_id_); + } + +#ifndef USE_FAKE_MEDIA_STREAMS + // Must be called on the main thread + void SetPrincipalHandle_m(const PrincipalHandle& principal_handle) + { + class Message : public ControlMessage + { + public: + Message(GenericReceiveListener* listener, + MediaStream* stream, + const PrincipalHandle& principal_handle) + : ControlMessage(stream), + listener_(listener), + principal_handle_(principal_handle) + {} + + void Run() override { + listener_->SetPrincipalHandle_msg(principal_handle_); + } + + RefPtr<GenericReceiveListener> listener_; + PrincipalHandle principal_handle_; + }; + + source_->GraphImpl()->AppendMessage(MakeUnique<Message>(this, source_, principal_handle)); + } + + // Must be called on the MediaStreamGraph thread + void SetPrincipalHandle_msg(const PrincipalHandle& principal_handle) + { + principal_handle_ = principal_handle; + } +#endif // USE_FAKE_MEDIA_STREAMS + + protected: + SourceMediaStream *source_; + const TrackID track_id_; + TrackTicks played_ticks_; + PrincipalHandle principal_handle_; +}; + +MediaPipelineReceive::MediaPipelineReceive( + const std::string& pc, + nsCOMPtr<nsIEventTarget> main_thread, + nsCOMPtr<nsIEventTarget> sts_thread, + SourceMediaStream *stream, + const std::string& track_id, + int level, + RefPtr<MediaSessionConduit> conduit, + RefPtr<TransportFlow> rtp_transport, + RefPtr<TransportFlow> rtcp_transport, + nsAutoPtr<MediaPipelineFilter> filter) : + MediaPipeline(pc, RECEIVE, main_thread, sts_thread, + track_id, level, conduit, rtp_transport, + rtcp_transport, filter), + stream_(stream), + segments_added_(0) +{ + MOZ_ASSERT(stream_); +} + +MediaPipelineReceive::~MediaPipelineReceive() +{ + MOZ_ASSERT(!stream_); // Check that we have shut down already. +} + +class MediaPipelineReceiveAudio::PipelineListener + : public GenericReceiveListener +{ +public: + PipelineListener(SourceMediaStream * source, TrackID track_id, + const RefPtr<MediaSessionConduit>& conduit) + : GenericReceiveListener(source, track_id), + conduit_(conduit) + { + } + + ~PipelineListener() + { + if (!NS_IsMainThread()) { + // release conduit on mainthread. Must use forget()! + nsresult rv = NS_DispatchToMainThread(new + ConduitDeleteEvent(conduit_.forget())); + MOZ_ASSERT(!NS_FAILED(rv),"Could not dispatch conduit shutdown to main"); + if (NS_FAILED(rv)) { + MOZ_CRASH(); + } + } else { + conduit_ = nullptr; + } + } + + // Implement MediaStreamListener + void NotifyPull(MediaStreamGraph* graph, StreamTime desired_time) override + { + MOZ_ASSERT(source_); + if (!source_) { + MOZ_MTLOG(ML_ERROR, "NotifyPull() called from a non-SourceMediaStream"); + return; + } + + // This comparison is done in total time to avoid accumulated roundoff errors. + while (source_->TicksToTimeRoundDown(WEBRTC_DEFAULT_SAMPLE_RATE, + played_ticks_) < desired_time) { + int16_t scratch_buffer[AUDIO_SAMPLE_BUFFER_MAX]; + + int samples_length; + + // This fetches 10ms of data, either mono or stereo + MediaConduitErrorCode err = + static_cast<AudioSessionConduit*>(conduit_.get())->GetAudioFrame( + scratch_buffer, + WEBRTC_DEFAULT_SAMPLE_RATE, + 0, // TODO(ekr@rtfm.com): better estimate of "capture" (really playout) delay + samples_length); + + if (err != kMediaConduitNoError) { + // Insert silence on conduit/GIPS failure (extremely unlikely) + MOZ_MTLOG(ML_ERROR, "Audio conduit failed (" << err + << ") to return data @ " << played_ticks_ + << " (desired " << desired_time << " -> " + << source_->StreamTimeToSeconds(desired_time) << ")"); + // if this is not enough we'll loop and provide more + samples_length = WEBRTC_DEFAULT_SAMPLE_RATE/100; + PodArrayZero(scratch_buffer); + } + + MOZ_ASSERT(samples_length * sizeof(uint16_t) < AUDIO_SAMPLE_BUFFER_MAX); + + MOZ_MTLOG(ML_DEBUG, "Audio conduit returned buffer of length " + << samples_length); + + RefPtr<SharedBuffer> samples = SharedBuffer::Create(samples_length * sizeof(uint16_t)); + int16_t *samples_data = static_cast<int16_t *>(samples->Data()); + AudioSegment segment; + // We derive the number of channels of the stream from the number of samples + // the AudioConduit gives us, considering it gives us packets of 10ms and we + // know the rate. + uint32_t channelCount = samples_length / (WEBRTC_DEFAULT_SAMPLE_RATE / 100); + AutoTArray<int16_t*,2> channels; + AutoTArray<const int16_t*,2> outputChannels; + size_t frames = samples_length / channelCount; + + channels.SetLength(channelCount); + + size_t offset = 0; + for (size_t i = 0; i < channelCount; i++) { + channels[i] = samples_data + offset; + offset += frames; + } + + DeinterleaveAndConvertBuffer(scratch_buffer, + frames, + channelCount, + channels.Elements()); + + outputChannels.AppendElements(channels); + + segment.AppendFrames(samples.forget(), outputChannels, frames, + principal_handle_); + + // Handle track not actually added yet or removed/finished + if (source_->AppendToTrack(track_id_, &segment)) { + played_ticks_ += frames; + } else { + MOZ_MTLOG(ML_ERROR, "AppendToTrack failed"); + // we can't un-read the data, but that's ok since we don't want to + // buffer - but don't i-loop! + return; + } + } + } + +private: + RefPtr<MediaSessionConduit> conduit_; +}; + +MediaPipelineReceiveAudio::MediaPipelineReceiveAudio( + const std::string& pc, + nsCOMPtr<nsIEventTarget> main_thread, + nsCOMPtr<nsIEventTarget> sts_thread, + SourceMediaStream* stream, + const std::string& media_stream_track_id, + TrackID numeric_track_id, + int level, + RefPtr<AudioSessionConduit> conduit, + RefPtr<TransportFlow> rtp_transport, + RefPtr<TransportFlow> rtcp_transport, + nsAutoPtr<MediaPipelineFilter> filter) : + MediaPipelineReceive(pc, main_thread, sts_thread, + stream, media_stream_track_id, level, conduit, + rtp_transport, rtcp_transport, filter), + listener_(new PipelineListener(stream, numeric_track_id, conduit)) +{} + +void MediaPipelineReceiveAudio::DetachMedia() +{ + ASSERT_ON_THREAD(main_thread_); + if (stream_ && listener_) { + listener_->EndTrack(); + stream_->RemoveListener(listener_); + stream_ = nullptr; + } +} + +nsresult MediaPipelineReceiveAudio::Init() { + ASSERT_ON_THREAD(main_thread_); + MOZ_MTLOG(ML_DEBUG, __FUNCTION__); + + description_ = pc_ + "| Receive audio["; + description_ += track_id_; + description_ += "]"; + + listener_->AddSelf(); + + return MediaPipelineReceive::Init(); +} + +#ifndef USE_FAKE_MEDIA_STREAMS +void MediaPipelineReceiveAudio::SetPrincipalHandle_m(const PrincipalHandle& principal_handle) +{ + listener_->SetPrincipalHandle_m(principal_handle); +} +#endif // USE_FAKE_MEDIA_STREAMS + +class MediaPipelineReceiveVideo::PipelineListener + : public GenericReceiveListener { +public: + PipelineListener(SourceMediaStream * source, TrackID track_id) + : GenericReceiveListener(source, track_id), + width_(0), + height_(0), +#if defined(MOZILLA_INTERNAL_API) + image_container_(), + image_(), +#endif + monitor_("Video PipelineListener") + { +#if !defined(MOZILLA_EXTERNAL_LINKAGE) + image_container_ = + LayerManager::CreateImageContainer(ImageContainer::ASYNCHRONOUS); +#endif + } + + // Implement MediaStreamListener + void NotifyPull(MediaStreamGraph* graph, StreamTime desired_time) override + { + #if defined(MOZILLA_INTERNAL_API) + ReentrantMonitorAutoEnter enter(monitor_); + + RefPtr<Image> image = image_; + StreamTime delta = desired_time - played_ticks_; + + // Don't append if we've already provided a frame that supposedly + // goes past the current aDesiredTime Doing so means a negative + // delta and thus messes up handling of the graph + if (delta > 0) { + VideoSegment segment; + segment.AppendFrame(image.forget(), delta, IntSize(width_, height_), + principal_handle_); + // Handle track not actually added yet or removed/finished + if (source_->AppendToTrack(track_id_, &segment)) { + played_ticks_ = desired_time; + } else { + MOZ_MTLOG(ML_ERROR, "AppendToTrack failed"); + return; + } + } + #endif + } + + // Accessors for external writes from the renderer + void FrameSizeChange(unsigned int width, + unsigned int height, + unsigned int number_of_streams) { + ReentrantMonitorAutoEnter enter(monitor_); + + width_ = width; + height_ = height; + } + + void RenderVideoFrame(const unsigned char* buffer, + size_t buffer_size, + uint32_t time_stamp, + int64_t render_time, + const RefPtr<layers::Image>& video_image) + { + RenderVideoFrame(buffer, buffer_size, width_, (width_ + 1) >> 1, + time_stamp, render_time, video_image); + } + + void RenderVideoFrame(const unsigned char* buffer, + size_t buffer_size, + uint32_t y_stride, + uint32_t cbcr_stride, + uint32_t time_stamp, + int64_t render_time, + const RefPtr<layers::Image>& video_image) + { +#ifdef MOZILLA_INTERNAL_API + ReentrantMonitorAutoEnter enter(monitor_); +#endif // MOZILLA_INTERNAL_API + +#if defined(MOZILLA_INTERNAL_API) + if (buffer) { + // Create a video frame using |buffer|. +#ifdef MOZ_WIDGET_GONK + RefPtr<PlanarYCbCrImage> yuvImage = new GrallocImage(); +#else + RefPtr<PlanarYCbCrImage> yuvImage = image_container_->CreatePlanarYCbCrImage(); +#endif + uint8_t* frame = const_cast<uint8_t*>(static_cast<const uint8_t*> (buffer)); + + PlanarYCbCrData yuvData; + yuvData.mYChannel = frame; + yuvData.mYSize = IntSize(y_stride, height_); + yuvData.mYStride = y_stride; + yuvData.mCbCrStride = cbcr_stride; + yuvData.mCbChannel = frame + height_ * yuvData.mYStride; + yuvData.mCrChannel = yuvData.mCbChannel + ((height_ + 1) >> 1) * yuvData.mCbCrStride; + yuvData.mCbCrSize = IntSize(yuvData.mCbCrStride, (height_ + 1) >> 1); + yuvData.mPicX = 0; + yuvData.mPicY = 0; + yuvData.mPicSize = IntSize(width_, height_); + yuvData.mStereoMode = StereoMode::MONO; + + if (!yuvImage->CopyData(yuvData)) { + MOZ_ASSERT(false); + return; + } + + image_ = yuvImage; + } +#ifdef WEBRTC_GONK + else { + // Decoder produced video frame that can be appended to the track directly. + MOZ_ASSERT(video_image); + image_ = video_image; + } +#endif // WEBRTC_GONK +#endif // MOZILLA_INTERNAL_API + } + +private: + int width_; + int height_; +#if defined(MOZILLA_INTERNAL_API) + RefPtr<layers::ImageContainer> image_container_; + RefPtr<layers::Image> image_; +#endif + mozilla::ReentrantMonitor monitor_; // Monitor for processing WebRTC frames. + // Protects image_ against: + // - Writing from the GIPS thread + // - Reading from the MSG thread +}; + +class MediaPipelineReceiveVideo::PipelineRenderer : public VideoRenderer +{ +public: + explicit PipelineRenderer(MediaPipelineReceiveVideo *pipeline) : + pipeline_(pipeline) {} + + void Detach() { pipeline_ = nullptr; } + + // Implement VideoRenderer + void FrameSizeChange(unsigned int width, + unsigned int height, + unsigned int number_of_streams) override + { + pipeline_->listener_->FrameSizeChange(width, height, number_of_streams); + } + + void RenderVideoFrame(const unsigned char* buffer, + size_t buffer_size, + uint32_t time_stamp, + int64_t render_time, + const ImageHandle& handle) override + { + pipeline_->listener_->RenderVideoFrame(buffer, buffer_size, + time_stamp, render_time, + handle.GetImage()); + } + + void RenderVideoFrame(const unsigned char* buffer, + size_t buffer_size, + uint32_t y_stride, + uint32_t cbcr_stride, + uint32_t time_stamp, + int64_t render_time, + const ImageHandle& handle) override + { + pipeline_->listener_->RenderVideoFrame(buffer, buffer_size, + y_stride, cbcr_stride, + time_stamp, render_time, + handle.GetImage()); + } + +private: + MediaPipelineReceiveVideo *pipeline_; // Raw pointer to avoid cycles +}; + + +MediaPipelineReceiveVideo::MediaPipelineReceiveVideo( + const std::string& pc, + nsCOMPtr<nsIEventTarget> main_thread, + nsCOMPtr<nsIEventTarget> sts_thread, + SourceMediaStream *stream, + const std::string& media_stream_track_id, + TrackID numeric_track_id, + int level, + RefPtr<VideoSessionConduit> conduit, + RefPtr<TransportFlow> rtp_transport, + RefPtr<TransportFlow> rtcp_transport, + nsAutoPtr<MediaPipelineFilter> filter) : + MediaPipelineReceive(pc, main_thread, sts_thread, + stream, media_stream_track_id, level, conduit, + rtp_transport, rtcp_transport, filter), + renderer_(new PipelineRenderer(this)), + listener_(new PipelineListener(stream, numeric_track_id)) +{} + +void MediaPipelineReceiveVideo::DetachMedia() +{ + ASSERT_ON_THREAD(main_thread_); + + // stop generating video and thus stop invoking the PipelineRenderer + // and PipelineListener - the renderer has a raw ptr to the Pipeline to + // avoid cycles, and the render callbacks are invoked from a different + // thread so simple null-checks would cause TSAN bugs without locks. + static_cast<VideoSessionConduit*>(conduit_.get())->DetachRenderer(); + if (stream_ && listener_) { + listener_->EndTrack(); + stream_->RemoveListener(listener_); + stream_ = nullptr; + } +} + +nsresult MediaPipelineReceiveVideo::Init() { + ASSERT_ON_THREAD(main_thread_); + MOZ_MTLOG(ML_DEBUG, __FUNCTION__); + + description_ = pc_ + "| Receive video["; + description_ += track_id_; + description_ += "]"; + +#if defined(MOZILLA_INTERNAL_API) + listener_->AddSelf(); +#endif + + // Always happens before we can DetachMedia() + static_cast<VideoSessionConduit *>(conduit_.get())-> + AttachRenderer(renderer_); + + return MediaPipelineReceive::Init(); +} + +#ifndef USE_FAKE_MEDIA_STREAMS +void MediaPipelineReceiveVideo::SetPrincipalHandle_m(const PrincipalHandle& principal_handle) +{ + listener_->SetPrincipalHandle_m(principal_handle); +} +#endif // USE_FAKE_MEDIA_STREAMS + +} // end namespace |