Split libjingle's signaling thread from the UI thread

content/renderer/media/rtc_peer_connection_handler.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "content/renderer/media/rtc_peer_connection_handler.h"
 
 #include <string>
 #include <utility>
 #include <vector>
 
@@ skipping 23 matching lines @@
 #include "third_party/WebKit/public/platform/WebMediaStreamSource.h"
 #include "third_party/WebKit/public/platform/WebRTCConfiguration.h"
 #include "third_party/WebKit/public/platform/WebRTCDataChannelInit.h"
 #include "third_party/WebKit/public/platform/WebRTCICECandidate.h"
 #include "third_party/WebKit/public/platform/WebRTCOfferOptions.h"
 #include "third_party/WebKit/public/platform/WebRTCSessionDescription.h"
 #include "third_party/WebKit/public/platform/WebRTCSessionDescriptionRequest.h"
 #include "third_party/WebKit/public/platform/WebRTCVoidRequest.h"
 #include "third_party/WebKit/public/platform/WebURL.h"
 
+using webrtc::DataChannelInterface;
+using webrtc::IceCandidateInterface;
+using webrtc::MediaStreamInterface;
+using webrtc::PeerConnectionInterface;
+using webrtc::PeerConnectionObserver;
 using webrtc::StatsReport;
+using webrtc::StatsReportCopyable;
 using webrtc::StatsReports;
 
 namespace content {
 
 // Converter functions from libjingle types to WebKit types.
 blink::WebRTCPeerConnectionHandlerClient::ICEGatheringState
 GetWebKitIceGatheringState(
     webrtc::PeerConnectionInterface::IceGatheringState state) {
   using blink::WebRTCPeerConnectionHandlerClient;
   switch (state) {
@@ skipping 109 matching lines @@
   }
 }
 
 class SessionDescriptionRequestTracker {
  public:
   SessionDescriptionRequestTracker(RTCPeerConnectionHandler* handler,
                                    PeerConnectionTracker::Action action)
       : handler_(handler), action_(action) {}
 
   void TrackOnSuccess(const webrtc::SessionDescriptionInterface* desc) {
+    DCHECK(thread_checker_.CalledOnValidThread());
     std::string value;
     if (desc) {
       desc->ToString(&value);
       value = "type: " + desc->type() + ", sdp: " + value;
     }
-    if (handler_->peer_connection_tracker())
+    if (handler_->peer_connection_tracker()) {
       handler_->peer_connection_tracker()->TrackSessionDescriptionCallback(
           handler_, action_, "OnSuccess", value);
+    }
   }
 
   void TrackOnFailure(const std::string& error) {
-    if (handler_->peer_connection_tracker())
+    DCHECK(thread_checker_.CalledOnValidThread());
+    if (handler_->peer_connection_tracker()) {
       handler_->peer_connection_tracker()->TrackSessionDescriptionCallback(
           handler_, action_, "OnFailure", error);
+    }
   }
 
  private:
   RTCPeerConnectionHandler* handler_;
   PeerConnectionTracker::Action action_;
+  base::ThreadChecker thread_checker_;
 };
 
 // Class mapping responses from calls to libjingle CreateOffer/Answer and
 // the blink::WebRTCSessionDescriptionRequest.
 class CreateSessionDescriptionRequest
     : public webrtc::CreateSessionDescriptionObserver {
  public:
   explicit CreateSessionDescriptionRequest(
+      const scoped_refptr<base::SingleThreadTaskRunner>& main_thread,
       const blink::WebRTCSessionDescriptionRequest& request,
       RTCPeerConnectionHandler* handler,
       PeerConnectionTracker::Action action)
-      : webkit_request_(request), tracker_(handler, action) {}
+      : main_thread_(main_thread),
+        webkit_request_(request),
+        tracker_(handler, action) {
+  }
 
   void OnSuccess(webrtc::SessionDescriptionInterface* desc) override {
+    if (!main_thread_->BelongsToCurrentThread()) {

[perkj_chrome, 2014/10/30 12:42:35]

Why is this? To avoid having a separate method OnSuccessOnMainThread with a
thread check? 
Shouldn't this always be called from libjingles signaling thread?

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Yes, this is first always called on the signaling thread. It's just for
convenience to keep the implementation in one place.  In places where we fetch
webrtc state on the signaling thread for optimizations sake, that's not
possible.

+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(&CreateSessionDescriptionRequest::OnSuccess, this, desc));
+      return;
+    }
+
     tracker_.TrackOnSuccess(desc);
     webkit_request_.requestSucceeded(CreateWebKitSessionDescription(desc));
     delete desc;
   }
   void OnFailure(const std::string& error) override {
+    if (!main_thread_->BelongsToCurrentThread()) {
+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(&CreateSessionDescriptionRequest::OnFailure, this, error));
+      return;
+    }
+
     tracker_.TrackOnFailure(error);
     webkit_request_.requestFailed(base::UTF8ToUTF16(error));
   }
 
  protected:
   ~CreateSessionDescriptionRequest() override {}
 
- private:
+  const scoped_refptr<base::SingleThreadTaskRunner> main_thread_;
   blink::WebRTCSessionDescriptionRequest webkit_request_;
   SessionDescriptionRequestTracker tracker_;
 };
 
 // Class mapping responses from calls to libjingle
 // SetLocalDescription/SetRemoteDescription and a blink::WebRTCVoidRequest.
 class SetSessionDescriptionRequest
     : public webrtc::SetSessionDescriptionObserver {
  public:
   explicit SetSessionDescriptionRequest(
+      const scoped_refptr<base::SingleThreadTaskRunner>& main_thread,
       const blink::WebRTCVoidRequest& request,
       RTCPeerConnectionHandler* handler,
       PeerConnectionTracker::Action action)
-      : webkit_request_(request), tracker_(handler, action) {}
+      : main_thread_(main_thread),
+        webkit_request_(request),
+        tracker_(handler, action) {
+  }
 
   void OnSuccess() override {
+    if (!main_thread_->BelongsToCurrentThread()) {

[perkj_chrome, 2014/10/30 12:42:35]

Same question as 222.

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Let me know if you prefer to split these up into separate methods

[perkj_chrome, 2014/10/31 08:37:24]

I personally think this is fine at least on this type of wrappers. But I got
review feedback earlier that preferered separate methods. If you split this you
should split all and that means a lot of new methods....

+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(&SetSessionDescriptionRequest::OnSuccess, this));
+      return;
+    }
     tracker_.TrackOnSuccess(NULL);
     webkit_request_.requestSucceeded();
   }
   void OnFailure(const std::string& error) override {
+    if (!main_thread_->BelongsToCurrentThread()) {
+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(&SetSessionDescriptionRequest::OnFailure, this, error));
+      return;
+    }
     tracker_.TrackOnFailure(error);
     webkit_request_.requestFailed(base::UTF8ToUTF16(error));
   }
 
  protected:
   ~SetSessionDescriptionRequest() override {}
 
  private:
+  const scoped_refptr<base::SingleThreadTaskRunner> main_thread_;
   blink::WebRTCVoidRequest webkit_request_;
   SessionDescriptionRequestTracker tracker_;
 };
 
 // Class mapping responses from calls to libjingle
 // GetStats into a blink::WebRTCStatsCallback.
 class StatsResponse : public webrtc::StatsObserver {
  public:
   explicit StatsResponse(const scoped_refptr<LocalRTCStatsRequest>& request)
-      : request_(request.get()), response_(request_->createResponse().get()) {
+      : request_(request.get()),
+        main_thread_(base::ThreadTaskRunnerHandle::Get()) {
     // Measure the overall time it takes to satisfy a getStats request.
     TRACE_EVENT_ASYNC_BEGIN0("webrtc", "getStats_Native", this);
   }
 
   void OnComplete(const StatsReports& reports) override {
-    TRACE_EVENT0("webrtc", "StatsResponse::OnComplete")
-    for (StatsReports::const_iterator it = reports.begin();
-         it != reports.end(); ++it) {
-      if ((*it)->values.size() > 0) {
-        AddReport(*(*it));
-      }
+    TRACE_EVENT0("webrtc", "StatsResponse::OnComplete");
+    DCHECK(reports_.empty());
+    reports_.reserve(reports.size());

[perkj_chrome, 2014/10/30 12:42:35]

reports_ are used to two threads.
Please add thread check that this is called on the signaling thread.

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

ah, right, there's no need for that (meant to do this actually so thanks for the
reminder). fixed.
done.

+    for (auto it : reports)
+      reports_.push_back(StatsReportCopyable(*it));
+
+    main_thread_->PostTask(FROM_HERE,
+        base::Bind(&StatsResponse::DeliverCallback, this));
+  }
+
+ private:
+  void DeliverCallback() {
+    DCHECK(main_thread_->BelongsToCurrentThread());
+    TRACE_EVENT0("webrtc", "StatsResponse::DeliverCallback");
+
+    // TODO: response_ doesn't need to be a member.

[perkj_chrome, 2014/10/30 12:42:35]

please fix.

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Done.

+    response_ = request_->createResponse().get();
+    for (const auto& report : reports_) {
+      if (report.values.size() > 0)
+        AddReport(report);
     }
 
+    reports_.clear();
+
     // Record the getSync operation as done before calling into Blink so that
     // we don't skew the perf measurements of the native code with whatever the
     // callback might be doing.
     TRACE_EVENT_ASYNC_END0("webrtc", "getStats_Native", this);
-
     request_->requestSucceeded(response_);
   }
 
- private:
   void AddReport(const StatsReport& report) {
     int idx = response_->addReport(blink::WebString::fromUTF8(report.id),
                                    blink::WebString::fromUTF8(report.type),
                                    report.timestamp);
     for (StatsReport::Values::const_iterator value_it = report.values.begin();
          value_it != report.values.end(); ++value_it) {
       AddStatistic(idx, value_it->display_name(), value_it->value);
     }
   }
 
   void AddStatistic(int idx, const char* name, const std::string& value) {
     response_->addStatistic(idx,
                             blink::WebString::fromUTF8(name),
                             blink::WebString::fromUTF8(value));
   }
 
   rtc::scoped_refptr<LocalRTCStatsRequest> request_;
+  std::vector<StatsReportCopyable> reports_;
   rtc::scoped_refptr<LocalRTCStatsResponse> response_;
+  const scoped_refptr<base::SingleThreadTaskRunner> main_thread_;
 };
 
 // Implementation of LocalRTCStatsRequest.
 LocalRTCStatsRequest::LocalRTCStatsRequest(blink::WebRTCStatsRequest impl)
-    : impl_(impl),
-      response_(NULL) {
+    : impl_(impl) {
 }
 
 LocalRTCStatsRequest::LocalRTCStatsRequest() {}
 LocalRTCStatsRequest::~LocalRTCStatsRequest() {}
 
 bool LocalRTCStatsRequest::hasSelector() const {
   return impl_.hasSelector();
 }
 
 blink::WebMediaStreamTrack LocalRTCStatsRequest::component() const {
   return impl_.component();
 }
 
 scoped_refptr<LocalRTCStatsResponse> LocalRTCStatsRequest::createResponse() {
-  DCHECK(!response_);
-  response_ = new rtc::RefCountedObject<LocalRTCStatsResponse>(
-      impl_.createResponse());
-  return response_.get();
+  return scoped_refptr<LocalRTCStatsResponse>(
+      new rtc::RefCountedObject<LocalRTCStatsResponse>(impl_.createResponse()));
 }
 
 void LocalRTCStatsRequest::requestSucceeded(
     const LocalRTCStatsResponse* response) {
   impl_.requestSucceeded(response->webKitStatsResponse());
 }
 
 // Implementation of LocalRTCStatsResponse.
 blink::WebRTCStatsResponse LocalRTCStatsResponse::webKitStatsResponse() const {
   return impl_;
 }
 
 size_t LocalRTCStatsResponse::addReport(blink::WebString type,
                                         blink::WebString id,
                                         double timestamp) {
   return impl_.addReport(type, id, timestamp);
 }
 
 void LocalRTCStatsResponse::addStatistic(size_t report,
                                          blink::WebString name,
                                          blink::WebString value) {
   impl_.addStatistic(report, name, value);
 }
 
 namespace {
 
+void GetStatsOnSignalingThread(
+    const scoped_refptr<webrtc::PeerConnectionInterface>& pc,
+    webrtc::PeerConnectionInterface::StatsOutputLevel level,
+    const scoped_refptr<webrtc::StatsObserver>& observer,
+    const scoped_refptr<webrtc::MediaStreamTrackInterface>& track) {
+  TRACE_EVENT0("webrtc", "GetStatsOnSignalingThread");
+  if (!pc->GetStats(observer.get(), track.get(), level)) {
+    DVLOG(1) << "GetStats failed.";
+    // TODO(hta): Consider how to get an error back.
+    observer->OnComplete(StatsReports());
+  }
+}
+
 class PeerConnectionUMAObserver : public webrtc::UMAObserver {
  public:
   PeerConnectionUMAObserver() {}
   ~PeerConnectionUMAObserver() override {}
 
   void IncrementCounter(
       webrtc::PeerConnectionUMAMetricsCounter counter) override {
+    // Runs on libjingle's signaling thread.
     UMA_HISTOGRAM_ENUMERATION("WebRTC.PeerConnection.IPMetrics",
                               counter,
                               webrtc::kBoundary);
   }
 
   void AddHistogramSample(webrtc::PeerConnectionUMAMetricsName type,
                           int value) override {
+    // Runs on libjingle's signaling thread.
     switch (type) {
       case webrtc::kTimeToConnect:
         UMA_HISTOGRAM_MEDIUM_TIMES(
             "WebRTC.PeerConnection.TimeToConnect",
             base::TimeDelta::FromMilliseconds(value));
         break;
       case webrtc::kNetworkInterfaces_IPv4:
         UMA_HISTOGRAM_COUNTS_100("WebRTC.PeerConnection.IPv4Interfaces",
                                  value);
         break;
       case webrtc::kNetworkInterfaces_IPv6:
         UMA_HISTOGRAM_COUNTS_100("WebRTC.PeerConnection.IPv6Interfaces",
                                  value);
         break;
       default:
         NOTREACHED();
     }
   }
 };
 
 base::LazyInstance<std::set<RTCPeerConnectionHandler*> >::Leaky
     g_peer_connection_handlers = LAZY_INSTANCE_INITIALIZER;
 
 }  // namespace
 
+class RTCPeerConnectionHandler::Observer

[perkj_chrome, 2014/10/30 12:42:35]

Please add a comment about the purpose of this class.
Ie, move PC callbacks from the sigaling thread to the main render thread.

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Done.

+    : public base::RefCountedThreadSafe<RTCPeerConnectionHandler::Observer>,
+      public PeerConnectionObserver {
+ public:
+  Observer(const base::WeakPtr<RTCPeerConnectionHandler>& handler)
+      : handler_(handler), main_thread_(base::ThreadTaskRunnerHandle::Get()) {}
+
+ protected:
+  friend class base::RefCountedThreadSafe<RTCPeerConnectionHandler::Observer>;
+  virtual ~Observer() {}
+
+  void OnError() override {
+    // TODO(perkj): Implement.

[perkj_chrome, 2014/10/30 12:42:35]

Or rater remove from the pc interface.

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Done.

+    NOTIMPLEMENTED();
+  }
+
+  void OnSignalingChange(
+      PeerConnectionInterface::SignalingState new_state) override {
+    if (!main_thread_->BelongsToCurrentThread()) {
+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(&RTCPeerConnectionHandler::Observer::OnSignalingChange,
+              this, new_state));
+    } else if (handler_) {
+      handler_->OnSignalingChange(new_state);
+    }
+  }
+
+  void OnAddStream(MediaStreamInterface* stream) override {
+    DCHECK(stream);
+    scoped_ptr<RemoteMediaStreamImpl> remote_stream(
+        new RemoteMediaStreamImpl(main_thread_, stream));

[perkj_chrome, 2014/10/30 12:42:35]

So this is instantiating |remote_stream| on the signaling thread? 
Wouldnt it be much simpler to instantiate it on the main render thread instead. 

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Discussed offline

+
+    // The webkit object owned by RemoteMediaStreamImpl, will be initialized
+    // asynchronously and the posted task will execude after that initialization
+    // is done.
+    main_thread_->PostTask(FROM_HERE,
+        base::Bind(&RTCPeerConnectionHandler::Observer::OnAddStreamImpl,
+            this, base::Passed(&remote_stream)));
+  }
+
+  void OnRemoveStream(MediaStreamInterface* stream) override {
+    main_thread_->PostTask(FROM_HERE,
+        base::Bind(&RTCPeerConnectionHandler::Observer::OnRemoveStreamImpl,
+            this, make_scoped_refptr(stream)));
+  }
+
+  void OnDataChannel(DataChannelInterface* data_channel) override {
+    scoped_ptr<RtcDataChannelHandler> handler(
+        new RtcDataChannelHandler(main_thread_, data_channel));
+    main_thread_->PostTask(FROM_HERE,
+        base::Bind(&RTCPeerConnectionHandler::Observer::OnDataChannelImpl,
+            this, base::Passed(&handler)));
+  }
+
+  void OnRenegotiationNeeded() override {
+    if (!main_thread_->BelongsToCurrentThread()) {
+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(&RTCPeerConnectionHandler::Observer::OnRenegotiationNeeded,
+              this));
+    } else if (handler_) {
+      handler_->OnRenegotiationNeeded();
+    }
+  }
+
+  void OnIceConnectionChange(
+      PeerConnectionInterface::IceConnectionState new_state) override {
+    if (!main_thread_->BelongsToCurrentThread()) {
+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(
+              &RTCPeerConnectionHandler::Observer::OnIceConnectionChange, this,
+              new_state));
+    } else if (handler_) {
+      handler_->OnIceConnectionChange(new_state);
+    }
+  }
+
+  void OnIceGatheringChange(
+      PeerConnectionInterface::IceGatheringState new_state) override {
+    if (!main_thread_->BelongsToCurrentThread()) {
+      main_thread_->PostTask(FROM_HERE,
+          base::Bind(&RTCPeerConnectionHandler::Observer::OnIceGatheringChange,
+              this, new_state));
+    } else if (handler_) {
+      handler_->OnIceGatheringChange(new_state);
+    }
+  }
+
+  void OnIceCandidate(const IceCandidateInterface* candidate) override {
+    std::string sdp;
+    if (!candidate->ToString(&sdp)) {
+      NOTREACHED() << "OnIceCandidate: Could not get SDP string.";
+      return;
+    }
+
+    main_thread_->PostTask(FROM_HERE,
+        base::Bind(&RTCPeerConnectionHandler::Observer::OnIceCandidateImpl,
+            this, sdp, candidate->sdp_mid(), candidate->sdp_mline_index(),
+            candidate->candidate().component(),
+            candidate->candidate().address().family()));
+  }
+
+  void OnAddStreamImpl(scoped_ptr<RemoteMediaStreamImpl> stream) {
+    DCHECK(stream->webkit_stream().extraData()) << "Initialization not done";
+    if (handler_)
+      handler_->OnAddStream(stream.Pass());
+  }
+
+  void OnRemoveStreamImpl(const scoped_refptr<MediaStreamInterface>& stream) {
+    if (handler_)
+      handler_->OnRemoveStream(stream);
+  }
+
+  void OnDataChannelImpl(scoped_ptr<RtcDataChannelHandler> handler) {
+    if (handler_)
+      handler_->OnDataChannel(handler.Pass());
+  }
+
+  void OnIceCandidateImpl(const std::string& sdp, const std::string& sdp_mid,
+      int sdp_mline_index, int component, int address_family) {
+    if (handler_) {
+      handler_->OnIceCandidate(sdp, sdp_mid, sdp_mline_index, component,
+          address_family);
+    }
+  }
+
+ private:
+  const base::WeakPtr<RTCPeerConnectionHandler> handler_;
+  const scoped_refptr<base::SingleThreadTaskRunner> main_thread_;
+};
+
 RTCPeerConnectionHandler::RTCPeerConnectionHandler(
     blink::WebRTCPeerConnectionHandlerClient* client,
-    PeerConnectionDependencyFactory* dependency_factory)
+    PeerConnectionDependencyFactory* dependency_factory,
+    const scoped_refptr<base::SingleThreadTaskRunner>& signaling_thread)
     : client_(client),
       dependency_factory_(dependency_factory),
       frame_(NULL),
+      signaling_thread_(signaling_thread),
       peer_connection_tracker_(NULL),
       num_data_channels_created_(0),
       num_local_candidates_ipv4_(0),
-      num_local_candidates_ipv6_(0) {
+      num_local_candidates_ipv6_(0),
+      weak_factory_(this) {
   g_peer_connection_handlers.Get().insert(this);
 }
 
 RTCPeerConnectionHandler::~RTCPeerConnectionHandler() {
+  DCHECK(thread_checker_.CalledOnValidThread());
   g_peer_connection_handlers.Get().erase(this);
   if (peer_connection_tracker_)
     peer_connection_tracker_->UnregisterPeerConnection(this);
   STLDeleteValues(&remote_streams_);
 
   UMA_HISTOGRAM_COUNTS_10000(
       "WebRTC.NumDataChannelsPerPeerConnection", num_data_channels_created_);
 }
 
 // static
 void RTCPeerConnectionHandler::DestructAllHandlers() {
   std::set<RTCPeerConnectionHandler*> handlers(
       g_peer_connection_handlers.Get().begin(),
       g_peer_connection_handlers.Get().end());
-  for (std::set<RTCPeerConnectionHandler*>::iterator handler = handlers.begin();
-       handler != handlers.end();
-       ++handler) {
-    (*handler)->client_->releasePeerConnectionHandler();
+  for (auto handler : handlers) {
+    if (handler->client_)
+      handler->client_->releasePeerConnectionHandler();
   }
 }
 
+// static
 void RTCPeerConnectionHandler::ConvertOfferOptionsToConstraints(
     const blink::WebRTCOfferOptions& options,
     RTCMediaConstraints* output) {
   output->AddMandatory(
       webrtc::MediaConstraintsInterface::kOfferToReceiveAudio,
       options.offerToReceiveAudio() > 0 ? "true" : "false",
       true);
 
   output->AddMandatory(
       webrtc::MediaConstraintsInterface::kOfferToReceiveVideo,
       options.offerToReceiveVideo() > 0 ? "true" : "false",
       true);
 
   if (!options.voiceActivityDetection()) {
     output->AddMandatory(
         webrtc::MediaConstraintsInterface::kVoiceActivityDetection,
         "false",
         true);
   }
 
   if (options.iceRestart()) {
     output->AddMandatory(
         webrtc::MediaConstraintsInterface::kIceRestart, "true", true);
   }
 }
 
 void RTCPeerConnectionHandler::associateWithFrame(blink::WebFrame* frame) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(frame);
   frame_ = frame;
 }
 
 bool RTCPeerConnectionHandler::initialize(
     const blink::WebRTCConfiguration& server_configuration,
     const blink::WebMediaConstraints& options) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(frame_);
 
   peer_connection_tracker_ =
       RenderThreadImpl::current()->peer_connection_tracker();
 
   webrtc::PeerConnectionInterface::RTCConfiguration config;
   GetNativeRtcConfiguration(server_configuration, &config);
 
   RTCMediaConstraints constraints(options);
 
-  native_peer_connection_ =
-      dependency_factory_->CreatePeerConnection(
-          config, &constraints, frame_, this);
+  peer_connection_observer_ = new Observer(weak_factory_.GetWeakPtr());
+  native_peer_connection_ = dependency_factory_->CreatePeerConnection(
+      config, &constraints, frame_, peer_connection_observer_.get());
 
   if (!native_peer_connection_.get()) {
     LOG(ERROR) << "Failed to initialize native PeerConnection.";
     return false;
   }
-  if (peer_connection_tracker_)
+
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->RegisterPeerConnection(
         this, config, constraints, frame_);
+  }
 
   uma_observer_ = new rtc::RefCountedObject<PeerConnectionUMAObserver>();
   native_peer_connection_->RegisterUMAObserver(uma_observer_.get());
   return true;
 }
 
 bool RTCPeerConnectionHandler::InitializeForTest(
     const blink::WebRTCConfiguration& server_configuration,
     const blink::WebMediaConstraints& options,
     PeerConnectionTracker* peer_connection_tracker) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   webrtc::PeerConnectionInterface::RTCConfiguration config;
   GetNativeRtcConfiguration(server_configuration, &config);
 
+  peer_connection_observer_ = new Observer(weak_factory_.GetWeakPtr());
   RTCMediaConstraints constraints(options);
-  native_peer_connection_ =
-      dependency_factory_->CreatePeerConnection(
-          config, &constraints, NULL, this);
+  native_peer_connection_ = dependency_factory_->CreatePeerConnection(
+      config, &constraints, NULL, peer_connection_observer_.get());
   if (!native_peer_connection_.get()) {
     LOG(ERROR) << "Failed to initialize native PeerConnection.";
     return false;
   }
   peer_connection_tracker_ = peer_connection_tracker;
   return true;
 }
 
 void RTCPeerConnectionHandler::createOffer(
     const blink::WebRTCSessionDescriptionRequest& request,
     const blink::WebMediaConstraints& options) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::createOffer");
+
   scoped_refptr<CreateSessionDescriptionRequest> description_request(
       new rtc::RefCountedObject<CreateSessionDescriptionRequest>(
-          request, this, PeerConnectionTracker::ACTION_CREATE_OFFER));
+          base::ThreadTaskRunnerHandle::Get(), request, this,
+          PeerConnectionTracker::ACTION_CREATE_OFFER));
+
   RTCMediaConstraints constraints(options);
   native_peer_connection_->CreateOffer(description_request.get(), &constraints);
 
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackCreateOffer(this, constraints);
 }
 
 void RTCPeerConnectionHandler::createOffer(
     const blink::WebRTCSessionDescriptionRequest& request,
     const blink::WebRTCOfferOptions& options) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::createOffer");
+
   scoped_refptr<CreateSessionDescriptionRequest> description_request(
       new rtc::RefCountedObject<CreateSessionDescriptionRequest>(
-          request, this, PeerConnectionTracker::ACTION_CREATE_OFFER));
+          base::ThreadTaskRunnerHandle::Get(), request, this,
+          PeerConnectionTracker::ACTION_CREATE_OFFER));
 
   RTCMediaConstraints constraints;
   ConvertOfferOptionsToConstraints(options, &constraints);
   native_peer_connection_->CreateOffer(description_request.get(), &constraints);
 
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackCreateOffer(this, constraints);
 }
 
 void RTCPeerConnectionHandler::createAnswer(
     const blink::WebRTCSessionDescriptionRequest& request,
     const blink::WebMediaConstraints& options) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::createAnswer");
   scoped_refptr<CreateSessionDescriptionRequest> description_request(
       new rtc::RefCountedObject<CreateSessionDescriptionRequest>(
-          request, this, PeerConnectionTracker::ACTION_CREATE_ANSWER));
+          base::ThreadTaskRunnerHandle::Get(), request, this,
+          PeerConnectionTracker::ACTION_CREATE_ANSWER));
   RTCMediaConstraints constraints(options);
   native_peer_connection_->CreateAnswer(description_request.get(),
                                         &constraints);
 
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackCreateAnswer(this, constraints);
 }
 
 void RTCPeerConnectionHandler::setLocalDescription(
     const blink::WebRTCVoidRequest& request,
     const blink::WebRTCSessionDescription& description) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::setLocalDescription");
   webrtc::SdpParseError error;
   webrtc::SessionDescriptionInterface* native_desc =
       CreateNativeSessionDescription(description, &error);
   if (!native_desc) {
     std::string reason_str = "Failed to parse SessionDescription. ";
     reason_str.append(error.line);
     reason_str.append(" ");
     reason_str.append(error.description);
     LOG(ERROR) << reason_str;
     request.requestFailed(blink::WebString::fromUTF8(reason_str));
     return;
   }
-  if (peer_connection_tracker_)
+
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackSetSessionDescription(
         this, description, PeerConnectionTracker::SOURCE_LOCAL);
+  }
 
   scoped_refptr<SetSessionDescriptionRequest> set_request(
       new rtc::RefCountedObject<SetSessionDescriptionRequest>(
-          request, this, PeerConnectionTracker::ACTION_SET_LOCAL_DESCRIPTION));
+          base::ThreadTaskRunnerHandle::Get(), request, this,
+          PeerConnectionTracker::ACTION_SET_LOCAL_DESCRIPTION));
   native_peer_connection_->SetLocalDescription(set_request.get(), native_desc);
 }
 
 void RTCPeerConnectionHandler::setRemoteDescription(
     const blink::WebRTCVoidRequest& request,
     const blink::WebRTCSessionDescription& description) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::setRemoteDescription");
   webrtc::SdpParseError error;
   webrtc::SessionDescriptionInterface* native_desc =
       CreateNativeSessionDescription(description, &error);
   if (!native_desc) {
     std::string reason_str = "Failed to parse SessionDescription. ";
     reason_str.append(error.line);
     reason_str.append(" ");
     reason_str.append(error.description);
     LOG(ERROR) << reason_str;
     request.requestFailed(blink::WebString::fromUTF8(reason_str));
     return;
   }
-  if (peer_connection_tracker_)
+
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackSetSessionDescription(
         this, description, PeerConnectionTracker::SOURCE_REMOTE);
+  }
 
   scoped_refptr<SetSessionDescriptionRequest> set_request(
       new rtc::RefCountedObject<SetSessionDescriptionRequest>(
-          request, this, PeerConnectionTracker::ACTION_SET_REMOTE_DESCRIPTION));
+          base::ThreadTaskRunnerHandle::Get(), request, this,
+          PeerConnectionTracker::ACTION_SET_REMOTE_DESCRIPTION));
+  // TODO(tommi): This cannot be posted to the signaling thread unfortunately

[perkj_chrome, 2014/10/30 12:42:35]

Can we talk about this?  Is this only due to our tests are wrong or that the
spec actually say otherwise? This is async in the spec so it sounds like this
should also be async.

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Added a TODO to do this in a following cl as discussed.

+  // because the caller depends on the state being changed synchronously.

[perkj_chrome, 2014/10/30 12:42:35]

We should at least but a similar explanation with setLocalDescription.

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Done.

+  // I'm guessing this is so that the local and remote getter functions will
+  // immediately return what was just set.
+  // This unfortunately does block the UI thread for some time (tens of ms on
+  // beefy machines) each time it's called.
   native_peer_connection_->SetRemoteDescription(set_request.get(), native_desc);
 }
 
 blink::WebRTCSessionDescription
 RTCPeerConnectionHandler::localDescription() {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::localDescription");
   const webrtc::SessionDescriptionInterface* native_desc =
       native_peer_connection_->local_description();

[perkj_chrome, 2014/10/30 12:42:35]

Are these methods the problem? 

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Yes - or rather the html test page had a bug which I've now fixed.

   blink::WebRTCSessionDescription description =
       CreateWebKitSessionDescription(native_desc);
   return description;
 }
 
 blink::WebRTCSessionDescription
 RTCPeerConnectionHandler::remoteDescription() {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::remoteDescription");
   const webrtc::SessionDescriptionInterface* native_desc =
       native_peer_connection_->remote_description();
   blink::WebRTCSessionDescription description =
       CreateWebKitSessionDescription(native_desc);
   return description;
 }
 
 bool RTCPeerConnectionHandler::updateICE(
     const blink::WebRTCConfiguration& server_configuration,
     const blink::WebMediaConstraints& options) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::updateICE");
   webrtc::PeerConnectionInterface::RTCConfiguration config;
   GetNativeRtcConfiguration(server_configuration, &config);
   RTCMediaConstraints constraints(options);
 
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackUpdateIce(this, config, constraints);
 
-  return native_peer_connection_->UpdateIce(config.servers,
-                                            &constraints);
+  return native_peer_connection_->UpdateIce(config.servers, &constraints);
 }
 
 bool RTCPeerConnectionHandler::addICECandidate(
-  const blink::WebRTCVoidRequest& request,
-  const blink::WebRTCICECandidate& candidate) {
+    const blink::WebRTCVoidRequest& request,
+    const blink::WebRTCICECandidate& candidate) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::addICECandidate");
   // Libjingle currently does not accept callbacks for addICECandidate.
   // For that reason we are going to call callbacks from here.
   bool result = addICECandidate(candidate);
   base::MessageLoop::current()->PostTask(
       FROM_HERE,
       base::Bind(&RTCPeerConnectionHandler::OnaddICECandidateResult,
-                 base::Unretained(this), request, result));
+                 weak_factory_.GetWeakPtr(), request, result));
   // On failure callback will be triggered.
   return true;
 }
 
 bool RTCPeerConnectionHandler::addICECandidate(
     const blink::WebRTCICECandidate& candidate) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::addICECandidate");
   scoped_ptr<webrtc::IceCandidateInterface> native_candidate(
       dependency_factory_->CreateIceCandidate(
           base::UTF16ToUTF8(candidate.sdpMid()),
           candidate.sdpMLineIndex(),
           base::UTF16ToUTF8(candidate.candidate())));
   bool return_value = false;
 
   if (native_candidate) {
     return_value =
         native_peer_connection_->AddIceCandidate(native_candidate.get());
     LOG_IF(ERROR, !return_value) << "Error processing ICE candidate.";
   } else {
     LOG(ERROR) << "Could not create native ICE candidate.";
   }
 
   if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackAddIceCandidate(
         this, candidate, PeerConnectionTracker::SOURCE_REMOTE, return_value);
   }
   return return_value;
 }
 
 void RTCPeerConnectionHandler::OnaddICECandidateResult(
     const blink::WebRTCVoidRequest& webkit_request, bool result) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnaddICECandidateResult");
   if (!result) {
     // We don't have the actual error code from the libjingle, so for now
     // using a generic error string.
     return webkit_request.requestFailed(
         base::UTF8ToUTF16("Error processing ICE candidate"));
   }
 
   return webkit_request.requestSucceeded();
 }
 
 bool RTCPeerConnectionHandler::addStream(
     const blink::WebMediaStream& stream,
     const blink::WebMediaConstraints& options) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::addStream");
   for (ScopedVector<WebRtcMediaStreamAdapter>::iterator adapter_it =
       local_streams_.begin(); adapter_it != local_streams_.end();
       ++adapter_it) {
     if ((*adapter_it)->IsEqual(stream)) {
       DVLOG(1) << "RTCPeerConnectionHandler::addStream called with the same "
                << "stream twice. id=" << stream.id().utf8();
       return false;
     }
   }
 
-  if (peer_connection_tracker_)
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackAddStream(
         this, stream, PeerConnectionTracker::SOURCE_LOCAL);
+  }
 
   PerSessionWebRTCAPIMetrics::GetInstance()->IncrementStreamCounter();
 
   WebRtcMediaStreamAdapter* adapter =
       new WebRtcMediaStreamAdapter(stream, dependency_factory_);
   local_streams_.push_back(adapter);
 
   webrtc::MediaStreamInterface* webrtc_stream = adapter->webrtc_media_stream();
   track_metrics_.AddStream(MediaStreamTrackMetrics::SENT_STREAM,
                            webrtc_stream);
 
   RTCMediaConstraints constraints(options);
   return native_peer_connection_->AddStream(webrtc_stream, &constraints);
 }
 
 void RTCPeerConnectionHandler::removeStream(
     const blink::WebMediaStream& stream) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::removeStream");
   // Find the webrtc stream.
   scoped_refptr<webrtc::MediaStreamInterface> webrtc_stream;
   for (ScopedVector<WebRtcMediaStreamAdapter>::iterator adapter_it =
            local_streams_.begin(); adapter_it != local_streams_.end();
        ++adapter_it) {
     if ((*adapter_it)->IsEqual(stream)) {
       webrtc_stream = (*adapter_it)->webrtc_media_stream();
       local_streams_.erase(adapter_it);
       break;
     }
   }
   DCHECK(webrtc_stream.get());
   native_peer_connection_->RemoveStream(webrtc_stream.get());
 
-  if (peer_connection_tracker_)
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackRemoveStream(
         this, stream, PeerConnectionTracker::SOURCE_LOCAL);
+  }
   PerSessionWebRTCAPIMetrics::GetInstance()->DecrementStreamCounter();
   track_metrics_.RemoveStream(MediaStreamTrackMetrics::SENT_STREAM,
                               webrtc_stream.get());
 }
 
 void RTCPeerConnectionHandler::getStats(
     const blink::WebRTCStatsRequest& request) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   scoped_refptr<LocalRTCStatsRequest> inner_request(
       new rtc::RefCountedObject<LocalRTCStatsRequest>(request));
-  getStats(inner_request.get());
+  getStats(inner_request);
 }
 
-void RTCPeerConnectionHandler::getStats(LocalRTCStatsRequest* request) {
+void RTCPeerConnectionHandler::getStats(
+    const scoped_refptr<LocalRTCStatsRequest>& request) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::getStats");
   rtc::scoped_refptr<webrtc::StatsObserver> observer(
       new rtc::RefCountedObject<StatsResponse>(request));
   webrtc::MediaStreamTrackInterface* track = NULL;
   if (request->hasSelector()) {
     blink::WebMediaStreamSource::Type type =
         request->component().source().type();
     std::string track_id = request->component().id().utf8();
     if (type == blink::WebMediaStreamSource::TypeAudio) {
+      // TODO: native_peer_connection_ will introduce a thread hop. Do this

[perkj_chrome, 2014/10/30 12:42:35]

What is the status of this?

[tommi (sloooow) - chröme, 2014/10/30 20:37:36]

Move all of this over to the signaling thread as discussed.
Also added TODOs for other methods such as setRemote and Local description to
move them to the signaling thread.

+      // on the signaling thread.
       track =
           native_peer_connection_->local_streams()->FindAudioTrack(track_id);
       if (!track) {
         track =
             native_peer_connection_->remote_streams()->FindAudioTrack(track_id);
       }
     } else {
       DCHECK_EQ(blink::WebMediaStreamSource::TypeVideo, type);
       track =
           native_peer_connection_->local_streams()->FindVideoTrack(track_id);
@@ skipping 11 matching lines @@
   }
   GetStats(observer,
            track,
            webrtc::PeerConnectionInterface::kStatsOutputLevelStandard);
 }
 
 void RTCPeerConnectionHandler::GetStats(
     webrtc::StatsObserver* observer,
     webrtc::MediaStreamTrackInterface* track,
     webrtc::PeerConnectionInterface::StatsOutputLevel level) {
-  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::GetStats");
-  if (!native_peer_connection_->GetStats(observer, track, level)) {
-    DVLOG(1) << "GetStats failed.";
-    // TODO(hta): Consider how to get an error back.
-    observer->OnComplete(StatsReports());
-    return;
-  }
+  DCHECK(thread_checker_.CalledOnValidThread());
+  scoped_refptr<webrtc::StatsObserver> ref_observer(observer);
+  scoped_refptr<webrtc::MediaStreamTrackInterface> ref_track(track);
+
+  // TODO(tommi): Make other methods run from the signaling thread too.
+  signaling_thread_->PostTask(FROM_HERE,
+      base::Bind(&GetStatsOnSignalingThread, native_peer_connection_, level,
+                 ref_observer, ref_track));
 }
 
 void RTCPeerConnectionHandler::CloseClientPeerConnection() {
-  client_->closePeerConnection();
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (client_)
+    client_->closePeerConnection();
 }
 
 blink::WebRTCDataChannelHandler* RTCPeerConnectionHandler::createDataChannel(
     const blink::WebString& label, const blink::WebRTCDataChannelInit& init) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::createDataChannel");
   DVLOG(1) << "createDataChannel label " << base::UTF16ToUTF8(label);
 
   webrtc::DataChannelInit config;
   // TODO(jiayl): remove the deprecated reliable field once Libjingle is updated
   // to handle that.
   config.reliable = false;
   config.id = init.id;
   config.ordered = init.ordered;
   config.negotiated = init.negotiated;
   config.maxRetransmits = init.maxRetransmits;
   config.maxRetransmitTime = init.maxRetransmitTime;
   config.protocol = base::UTF16ToUTF8(init.protocol);
 
   rtc::scoped_refptr<webrtc::DataChannelInterface> webrtc_channel(
       native_peer_connection_->CreateDataChannel(base::UTF16ToUTF8(label),
                                                  &config));
   if (!webrtc_channel) {
     DLOG(ERROR) << "Could not create native data channel.";
     return NULL;
   }
-  if (peer_connection_tracker_)
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackCreateDataChannel(
         this, webrtc_channel.get(), PeerConnectionTracker::SOURCE_LOCAL);
+  }
 
   ++num_data_channels_created_;
 
   return new RtcDataChannelHandler(base::ThreadTaskRunnerHandle::Get(),
                                    webrtc_channel);
 }
 
 blink::WebRTCDTMFSenderHandler* RTCPeerConnectionHandler::createDTMFSender(
     const blink::WebMediaStreamTrack& track) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::createDTMFSender");
   DVLOG(1) << "createDTMFSender.";
 
   MediaStreamTrack* native_track = MediaStreamTrack::GetTrack(track);
-  if (!native_track ||
+  if (!native_track || !native_track->is_local_track() ||
       track.source().type() != blink::WebMediaStreamSource::TypeAudio) {
-    DLOG(ERROR) << "Could not create DTMF sender from a non-audio track.";
-    return NULL;
+    DLOG(ERROR) << "The DTMF sender requires a local audio track.";
+    return nullptr;
   }
 
   webrtc::AudioTrackInterface* audio_track = native_track->GetAudioAdapter();
   rtc::scoped_refptr<webrtc::DtmfSenderInterface> sender(
       native_peer_connection_->CreateDtmfSender(audio_track));
   if (!sender) {
     DLOG(ERROR) << "Could not create native DTMF sender.";
-    return NULL;
+    return nullptr;
   }
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackCreateDTMFSender(this, track);
 
   return new RtcDtmfSenderHandler(sender);
 }
 
 void RTCPeerConnectionHandler::stop() {
+  DCHECK(thread_checker_.CalledOnValidThread());
   DVLOG(1) << "RTCPeerConnectionHandler::stop";
 
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackStop(this);
   native_peer_connection_->Close();
-}
-
-void RTCPeerConnectionHandler::OnError() {
-  // TODO(perkj): Implement.
-  NOTIMPLEMENTED();
+  // The client_ pointer is not considered valid after this point and no further
+  // callbacks must be made.
+  client_ = nullptr;
 }
 
 void RTCPeerConnectionHandler::OnSignalingChange(
     webrtc::PeerConnectionInterface::SignalingState new_state) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnSignalingChange");
+
   blink::WebRTCPeerConnectionHandlerClient::SignalingState state =
       GetWebKitSignalingState(new_state);
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackSignalingStateChange(this, state);
-  client_->didChangeSignalingState(state);
+  if (client_)
+    client_->didChangeSignalingState(state);
 }
 
 // Called any time the IceConnectionState changes
 void RTCPeerConnectionHandler::OnIceConnectionChange(
     webrtc::PeerConnectionInterface::IceConnectionState new_state) {
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnIceConnectionChange");
+  DCHECK(thread_checker_.CalledOnValidThread());
   if (new_state == webrtc::PeerConnectionInterface::kIceConnectionChecking) {
     ice_connection_checking_start_ = base::TimeTicks::Now();
   } else if (new_state ==
       webrtc::PeerConnectionInterface::kIceConnectionConnected) {
     // If the state becomes connected, send the time needed for PC to become
     // connected from checking to UMA. UMA data will help to know how much
     // time needed for PC to connect with remote peer.
     UMA_HISTOGRAM_MEDIUM_TIMES(
         "WebRTC.PeerConnection.TimeToConnect",
         base::TimeTicks::Now() - ice_connection_checking_start_);
   }
 
   track_metrics_.IceConnectionChange(new_state);
   blink::WebRTCPeerConnectionHandlerClient::ICEConnectionState state =
       GetWebKitIceConnectionState(new_state);
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackIceConnectionStateChange(this, state);
-  client_->didChangeICEConnectionState(state);
+  if(client_)
+    client_->didChangeICEConnectionState(state);
 }
 
 // Called any time the IceGatheringState changes
 void RTCPeerConnectionHandler::OnIceGatheringChange(
     webrtc::PeerConnectionInterface::IceGatheringState new_state) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnIceGatheringChange");
+
   if (new_state == webrtc::PeerConnectionInterface::kIceGatheringComplete) {
     // If ICE gathering is completed, generate a NULL ICE candidate,
     // to signal end of candidates.
-    blink::WebRTCICECandidate null_candidate;
-    client_->didGenerateICECandidate(null_candidate);
+    if (client_) {
+      blink::WebRTCICECandidate null_candidate;
+      client_->didGenerateICECandidate(null_candidate);
+    }
 
     UMA_HISTOGRAM_COUNTS_100("WebRTC.PeerConnection.IPv4LocalCandidates",
                              num_local_candidates_ipv4_);
 
     UMA_HISTOGRAM_COUNTS_100("WebRTC.PeerConnection.IPv6LocalCandidates",
                              num_local_candidates_ipv6_);
   } else if (new_state ==
              webrtc::PeerConnectionInterface::kIceGatheringGathering) {
     // ICE restarts will change gathering state back to "gathering",
     // reset the counter.
     num_local_candidates_ipv6_ = 0;
     num_local_candidates_ipv4_ = 0;
   }
 
   blink::WebRTCPeerConnectionHandlerClient::ICEGatheringState state =
       GetWebKitIceGatheringState(new_state);
   if (peer_connection_tracker_)
     peer_connection_tracker_->TrackIceGatheringStateChange(this, state);
-  client_->didChangeICEGatheringState(state);
+  if (client_)
+    client_->didChangeICEGatheringState(state);
+}
+
+void RTCPeerConnectionHandler::OnRenegotiationNeeded() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnRenegotiationNeeded");
+  if (peer_connection_tracker_)
+    peer_connection_tracker_->TrackOnRenegotiationNeeded(this);
+  if (client_)
+    client_->negotiationNeeded();
+}
+
+PeerConnectionTracker* RTCPeerConnectionHandler::peer_connection_tracker() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  return peer_connection_tracker_;
 }
 
 void RTCPeerConnectionHandler::OnAddStream(
-    webrtc::MediaStreamInterface* stream_interface) {
-  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnAddStream");
-  DCHECK(stream_interface);
-  DCHECK(remote_streams_.find(stream_interface) == remote_streams_.end());
+    scoped_ptr<RemoteMediaStreamImpl> stream) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(remote_streams_.find(stream->webrtc_stream().get()) ==
+         remote_streams_.end());
+  DCHECK(stream->webkit_stream().extraData()) << "Initialization not done";
+  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnAddStreamImpl");
 
-  RemoteMediaStreamImpl* remote_stream =
-      new RemoteMediaStreamImpl(stream_interface);
+  // Ownership is with remote_streams_ now.
+  RemoteMediaStreamImpl* s = stream.release();
   remote_streams_.insert(
       std::pair<webrtc::MediaStreamInterface*, RemoteMediaStreamImpl*> (
-          stream_interface, remote_stream));
+          s->webrtc_stream().get(), s));
 
-  if (peer_connection_tracker_)
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackAddStream(
-        this, remote_stream->webkit_stream(),
-        PeerConnectionTracker::SOURCE_REMOTE);
+        this, s->webkit_stream(), PeerConnectionTracker::SOURCE_REMOTE);
+  }
 
   PerSessionWebRTCAPIMetrics::GetInstance()->IncrementStreamCounter();
 
   track_metrics_.AddStream(MediaStreamTrackMetrics::RECEIVED_STREAM,
-                           stream_interface);
-
-  client_->didAddRemoteStream(remote_stream->webkit_stream());
+                           s->webrtc_stream().get());
+  if (client_)
+    client_->didAddRemoteStream(s->webkit_stream());
 }
 
 void RTCPeerConnectionHandler::OnRemoveStream(
-    webrtc::MediaStreamInterface* stream_interface) {
-  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnRemoveStream");
-  DCHECK(stream_interface);
-  RemoteStreamMap::iterator it = remote_streams_.find(stream_interface);
+    const scoped_refptr<webrtc::MediaStreamInterface>& stream) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnRemoveStreamImpl");
+  RemoteStreamMap::iterator it = remote_streams_.find(stream.get());
   if (it == remote_streams_.end()) {
     NOTREACHED() << "Stream not found";
     return;
   }
 
   track_metrics_.RemoveStream(MediaStreamTrackMetrics::RECEIVED_STREAM,
-                              stream_interface);
+                              stream.get());
   PerSessionWebRTCAPIMetrics::GetInstance()->DecrementStreamCounter();
 
   scoped_ptr<RemoteMediaStreamImpl> remote_stream(it->second);
   const blink::WebMediaStream& webkit_stream = remote_stream->webkit_stream();
   DCHECK(!webkit_stream.isNull());
   remote_streams_.erase(it);
 
-  if (peer_connection_tracker_)
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackRemoveStream(
         this, webkit_stream, PeerConnectionTracker::SOURCE_REMOTE);
+  }
 
-  client_->didRemoveRemoteStream(webkit_stream);
+  if (client_)
+    client_->didRemoveRemoteStream(webkit_stream);
+}
+
+void RTCPeerConnectionHandler::OnDataChannel(
+    scoped_ptr<RtcDataChannelHandler> handler) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnDataChannelImpl");
+
+  if (peer_connection_tracker_) {
+    peer_connection_tracker_->TrackCreateDataChannel(
+        this, handler->channel().get(), PeerConnectionTracker::SOURCE_REMOTE);
+  }
+
+  if (client_)
+    client_->didAddRemoteDataChannel(handler.release());
 }
 
 void RTCPeerConnectionHandler::OnIceCandidate(
-    const webrtc::IceCandidateInterface* candidate) {
-  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnIceCandidate");
-  DCHECK(candidate);
-  std::string sdp;
-  if (!candidate->ToString(&sdp)) {
-    NOTREACHED() << "OnIceCandidate: Could not get SDP string.";
-    return;
-  }
+    const std::string& sdp, const std::string& sdp_mid, int sdp_mline_index,
+    int component, int address_family) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnIceCandidateImpl");
   blink::WebRTCICECandidate web_candidate;
   web_candidate.initialize(base::UTF8ToUTF16(sdp),
-                           base::UTF8ToUTF16(candidate->sdp_mid()),
-                           candidate->sdp_mline_index());
-  if (peer_connection_tracker_)
+                           base::UTF8ToUTF16(sdp_mid),
+                           sdp_mline_index);
+  if (peer_connection_tracker_) {
     peer_connection_tracker_->TrackAddIceCandidate(
         this, web_candidate, PeerConnectionTracker::SOURCE_LOCAL, true);
+  }
 
   // Only the first m line's first component is tracked to avoid
   // miscounting when doing BUNDLE or rtcp mux.
-  if (candidate->sdp_mline_index() == 0 &&
-      candidate->candidate().component() == 1) {
-    if (candidate->candidate().address().family() == AF_INET) {
-      num_local_candidates_ipv4_++;
-    } else if (candidate->candidate().address().family() == AF_INET6) {
-      num_local_candidates_ipv6_++;
+  if (sdp_mline_index == 0 && component == 1) {
+    if (address_family == AF_INET) {
+      ++num_local_candidates_ipv4_;
+    } else if (address_family == AF_INET6) {
+      ++num_local_candidates_ipv6_;
     } else {
       NOTREACHED();
     }
   }
-  client_->didGenerateICECandidate(web_candidate);
-}
-
-void RTCPeerConnectionHandler::OnDataChannel(
-    webrtc::DataChannelInterface* data_channel) {
-  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnDataChannel");
-  if (peer_connection_tracker_)
-    peer_connection_tracker_->TrackCreateDataChannel(
-        this, data_channel, PeerConnectionTracker::SOURCE_REMOTE);
-
-  DVLOG(1) << "RTCPeerConnectionHandler::OnDataChannel "
-           << data_channel->label();
-  client_->didAddRemoteDataChannel(new RtcDataChannelHandler(
-      base::ThreadTaskRunnerHandle::Get(), data_channel));
-}
-
-void RTCPeerConnectionHandler::OnRenegotiationNeeded() {
-  TRACE_EVENT0("webrtc", "RTCPeerConnectionHandler::OnRenegotiationNeeded");
-  if (peer_connection_tracker_)
-    peer_connection_tracker_->TrackOnRenegotiationNeeded(this);
-  client_->negotiationNeeded();
-}
-
-PeerConnectionTracker* RTCPeerConnectionHandler::peer_connection_tracker() {
-  return peer_connection_tracker_;
+  if (client_)
+    client_->didGenerateICECandidate(web_candidate);
 }
 
 webrtc::SessionDescriptionInterface*
 RTCPeerConnectionHandler::CreateNativeSessionDescription(
     const blink::WebRTCSessionDescription& description,
     webrtc::SdpParseError* error) {
   std::string sdp = base::UTF16ToUTF8(description.sdp());
   std::string type = base::UTF16ToUTF8(description.type());
   webrtc::SessionDescriptionInterface* native_desc =
       dependency_factory_->CreateSessionDescription(type, sdp, error);
 
   LOG_IF(ERROR, !native_desc) << "Failed to create native session description."
                               << " Type: " << type << " SDP: " << sdp;
 
   return native_desc;
 }
 
 }  // namespace content