components/devtools_bridge/session_dependency_factory.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "components/devtools_bridge/session_dependency_factory.h"
   6
   7 #include "base/bind.h"
   8 #include "base/location.h"
   9 #include "base/task_runner.h"
  10 #include "base/threading/thread.h"
  11 #include "components/devtools_bridge/abstract_data_channel.h"
  12 #include "components/devtools_bridge/abstract_peer_connection.h"
  13 #include "components/devtools_bridge/rtc_configuration.h"
  14 #include "third_party/libjingle/source/talk/app/webrtc/mediaconstraintsinterface.h"
  15 #include "third_party/libjingle/source/talk/app/webrtc/peerconnectioninterface.h"
  16 #include "third_party/webrtc/base/bind.h"
  17 #include "third_party/webrtc/base/messagehandler.h"
  18 #include "third_party/webrtc/base/messagequeue.h"
  19 #include "third_party/webrtc/base/ssladapter.h"
  20 #include "third_party/webrtc/base/thread.h"
  21
  22 namespace devtools_bridge {
  23
  24 class RTCConfiguration::Impl
  25     : public RTCConfiguration,
  26       public webrtc::PeerConnectionInterface::RTCConfiguration {
  27  public:
  28   void AddIceServer(const std::string& uri,
  29                     const std::string& username,
  30                     const std::string& credential) override {
  31     webrtc::PeerConnectionInterface::IceServer server;
  32     server.uri = uri;
  33     server.username = username;
  34     server.password = credential;
  35     servers.push_back(server);
  36   }
  37
  38   const Impl& impl() const override {
  39     return *this;
  40   }
  41
  42  private:
  43   webrtc::PeerConnectionInterface::RTCConfiguration base_;
  44 };
  45
  46 namespace {
  47
  48 template <typename T>
  49 void CheckedRelease(rtc::scoped_refptr<T>* ptr) {
  50   CHECK_EQ(0, ptr->release()->Release());
  51 }
  52
  53 class MediaConstraints
  54     : public webrtc::MediaConstraintsInterface {
  55  public:
  56   ~MediaConstraints() override {}
  57
  58   const Constraints& GetMandatory() const override { return mandatory_; }
  59
  60   const Constraints& GetOptional() const override { return optional_; }
  61
  62   void AddMandatory(const std::string& key, const std::string& value) {
  63     mandatory_.push_back(Constraint(key, value));
  64   }
  65
  66  private:
  67   Constraints mandatory_;
  68   Constraints optional_;
  69 };
  70
  71 /**
  72  * Posts tasks on signaling thread. If stopped (when SesseionDependencyFactry
  73  * is destroying) ignores posted tasks.
  74  */
  75 class SignalingThreadTaskRunner : public base::TaskRunner,
  76                                   private rtc::MessageHandler {
  77  public:
  78   explicit SignalingThreadTaskRunner(rtc::Thread* thread) : thread_(thread) {}
  79
  80   bool PostDelayedTask(const tracked_objects::Location& from_here,
  81                        const base::Closure& task,
  82                        base::TimeDelta delay) override {
  83     DCHECK(delay.ToInternalValue() == 0);
  84
  85     rtc::CritScope scope(&critical_section_);
  86
  87     if (thread_)
  88       thread_->Send(this, 0, new Task(task));
  89
  90     return true;
  91   }
  92
  93   bool RunsTasksOnCurrentThread() const override {
  94     rtc::CritScope scope(&critical_section_);
  95
  96     return thread_ != NULL && thread_->IsCurrent();
  97   }
  98
  99   void Stop() {
 100     rtc::CritScope scope(&critical_section_);
 101     thread_ = NULL;
 102   }
 103
 104  private:
 105   typedef rtc::TypedMessageData<base::Closure> Task;
 106
 107   ~SignalingThreadTaskRunner() override {}
 108
 109   void OnMessage(rtc::Message* msg) override {
 110     static_cast<Task*>(msg->pdata)->data().Run();
 111   }
 112
 113   mutable rtc::CriticalSection critical_section_;
 114   rtc::Thread* thread_;  // Guarded by |critical_section_|.
 115 };
 116
 117 class DataChannelObserverImpl : public webrtc::DataChannelObserver {
 118  public:
 119   DataChannelObserverImpl(
 120       webrtc::DataChannelInterface* data_channel,
 121       scoped_ptr<AbstractDataChannel::Observer> observer)
 122       : data_channel_(data_channel),
 123         observer_(observer.Pass()) {
 124   }
 125
 126   void InitState() {
 127     open_ = data_channel_->state() == webrtc::DataChannelInterface::kOpen;
 128   }
 129
 130   void OnStateChange() override {
 131     bool open = data_channel_->state() == webrtc::DataChannelInterface::kOpen;
 132
 133     if (open == open_) return;
 134
 135     open_ = open;
 136     if (open) {
 137       observer_->OnOpen();
 138     } else {
 139       observer_->OnClose();
 140     }
 141   }
 142
 143   void OnMessage(const webrtc::DataBuffer& buffer) override {
 144     observer_->OnMessage(buffer.data.data(), buffer.size());
 145   }
 146
 147  private:
 148   webrtc::DataChannelInterface* const data_channel_;
 149   scoped_ptr<AbstractDataChannel::Observer> const observer_;
 150   bool open_;
 151 };
 152
 153 /**
 154  * Thread-safe view on AbstractDataChannel.
 155  */
 156 class DataChannelProxyImpl : public AbstractDataChannel::Proxy {
 157  public:
 158   DataChannelProxyImpl(
 159       SessionDependencyFactory* factory,
 160       rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel)
 161       : data_channel_(data_channel),
 162         signaling_thread_task_runner_(
 163             factory->signaling_thread_task_runner()) {
 164   }
 165
 166   void StopOnSignalingThread() {
 167     data_channel_ = NULL;
 168   }
 169
 170   void SendBinaryMessage(const void* data, size_t length) override {
 171     auto buffer = make_scoped_ptr(new webrtc::DataBuffer(rtc::Buffer(), true));
 172     buffer->data.SetData(data, length);
 173
 174     signaling_thread_task_runner_->PostTask(
 175         FROM_HERE, base::Bind(
 176             &DataChannelProxyImpl::SendMessageOnSignalingThread,
 177             this,
 178             base::Passed(&buffer)));
 179   }
 180
 181   void Close() override {
 182     signaling_thread_task_runner_->PostTask(
 183         FROM_HERE, base::Bind(&DataChannelProxyImpl::CloseOnSignalingThread,
 184                               this));
 185   }
 186
 187  private:
 188
 189   ~DataChannelProxyImpl() override {}
 190
 191   void SendMessageOnSignalingThread(scoped_ptr<webrtc::DataBuffer> message) {
 192     if (data_channel_ != NULL)
 193       data_channel_->Send(*message);
 194   }
 195
 196   void CloseOnSignalingThread() {
 197     if (data_channel_ != NULL)
 198       data_channel_->Close();
 199   }
 200
 201   // Accessed on signaling thread.
 202   rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel_;
 203
 204   const scoped_refptr<base::TaskRunner> signaling_thread_task_runner_;
 205 };
 206
 207 class DataChannelImpl : public AbstractDataChannel {
 208  public:
 209   DataChannelImpl(
 210       SessionDependencyFactory* factory,
 211       rtc::Thread* const signaling_thread,
 212       rtc::scoped_refptr<webrtc::DataChannelInterface> impl)
 213       : factory_(factory),
 214         signaling_thread_(signaling_thread),
 215         impl_(impl) {
 216   }
 217
 218   ~DataChannelImpl() override {
 219     if (proxy_.get()) {
 220       signaling_thread_->Invoke<void>(rtc::Bind(
 221           &DataChannelProxyImpl::StopOnSignalingThread, proxy_.get()));
 222     }
 223   }
 224
 225   void RegisterObserver(scoped_ptr<Observer> observer) override {
 226     observer_.reset(new DataChannelObserverImpl(impl_.get(), observer.Pass()));
 227     signaling_thread_->Invoke<void>(rtc::Bind(
 228         &DataChannelImpl::RegisterObserverOnSignalingThread, this));
 229   }
 230
 231   void UnregisterObserver() override {
 232     DCHECK(observer_.get() != NULL);
 233     impl_->UnregisterObserver();
 234     observer_.reset();
 235   }
 236
 237   void SendBinaryMessage(void* data, size_t length) override {
 238     SendMessage(data, length, true);
 239   }
 240
 241   void SendTextMessage(void* data, size_t length) override {
 242     SendMessage(data, length, false);
 243   }
 244
 245   void SendMessage(void* data, size_t length, bool is_binary) {
 246     impl_->Send(webrtc::DataBuffer(rtc::Buffer(data, length), is_binary));
 247   }
 248
 249   void Close() override {
 250     impl_->Close();
 251   }
 252
 253   scoped_refptr<Proxy> proxy() override {
 254     if (!proxy_.get())
 255       proxy_ = new DataChannelProxyImpl(factory_, impl_);
 256     return proxy_;
 257   }
 258
 259  private:
 260   void RegisterObserverOnSignalingThread() {
 261     // State initialization and observer registration happen atomically
 262     // if done on the signaling thread (see rtc::Thread::Send).
 263     observer_->InitState();
 264     impl_->RegisterObserver(observer_.get());
 265   }
 266
 267   SessionDependencyFactory* const factory_;
 268   scoped_refptr<DataChannelProxyImpl> proxy_;
 269   rtc::Thread* const signaling_thread_;
 270   scoped_ptr<DataChannelObserverImpl> observer_;
 271   const rtc::scoped_refptr<webrtc::DataChannelInterface> impl_;
 272 };
 273
 274 class PeerConnectionObserverImpl
 275     : public webrtc::PeerConnectionObserver {
 276  public:
 277   PeerConnectionObserverImpl(AbstractPeerConnection::Delegate* delegate)
 278       : delegate_(delegate),
 279         connected_(false) {
 280   }
 281
 282   void OnAddStream(webrtc::MediaStreamInterface* stream) override {}
 283
 284   void OnRemoveStream(webrtc::MediaStreamInterface* stream) override {}
 285
 286   void OnDataChannel(webrtc::DataChannelInterface* data_channel) override {}
 287
 288   void OnRenegotiationNeeded() override {}
 289
 290   void OnSignalingChange(
 291       webrtc::PeerConnectionInterface::SignalingState new_state) override {}
 292
 293   void OnIceConnectionChange(
 294       webrtc::PeerConnectionInterface::IceConnectionState new_state) override {
 295     bool connected =
 296         new_state == webrtc::PeerConnectionInterface::kIceConnectionConnected ||
 297         new_state == webrtc::PeerConnectionInterface::kIceConnectionCompleted;
 298
 299     if (connected != connected_) {
 300       connected_ = connected;
 301       delegate_->OnIceConnectionChange(connected_);
 302     }
 303   }
 304
 305   void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override {
 306     std::string sdp;
 307     candidate->ToString(&sdp);
 308
 309     delegate_->OnIceCandidate(
 310         candidate->sdp_mid(), candidate->sdp_mline_index(), sdp);
 311   }
 312
 313  private:
 314   AbstractPeerConnection::Delegate* const delegate_;
 315   bool connected_;
 316 };
 317
 318 /**
 319  * Helper object which may outlive PeerConnectionImpl. Provides access
 320  * to the connection and the delegate to operaion callback objects
 321  * in a safe way. Always accessible on the signaling thread.
 322  */
 323 class PeerConnectionHolder : public rtc::RefCountInterface {
 324  public:
 325   PeerConnectionHolder(
 326       rtc::Thread* signaling_thread,
 327       webrtc::PeerConnectionInterface* connection,
 328       AbstractPeerConnection::Delegate* delegate)
 329       : signaling_thread_(signaling_thread),
 330         connection_(connection),
 331         delegate_(delegate),
 332         disposed_(false) {
 333   }
 334
 335   ~PeerConnectionHolder() override { DCHECK(disposed_); }
 336
 337   void Dispose() {
 338     DCHECK(!IsDisposed());
 339     disposed_ = true;
 340   }
 341
 342   webrtc::PeerConnectionInterface* connection() {
 343     DCHECK(!IsDisposed());
 344     return connection_;
 345   }
 346
 347   AbstractPeerConnection::Delegate* delegate() {
 348     DCHECK(!IsDisposed());
 349     return delegate_;
 350   }
 351
 352   bool IsDisposed() {
 353     DCHECK(signaling_thread_->IsCurrent());
 354     return disposed_;
 355   }
 356
 357  private:
 358   rtc::Thread* const signaling_thread_;
 359   webrtc::PeerConnectionInterface* const connection_;
 360   AbstractPeerConnection::Delegate* const delegate_;
 361   bool disposed_;
 362 };
 363
 364 class CreateAndSetHandler
 365     : public webrtc::CreateSessionDescriptionObserver,
 366       public webrtc::SetSessionDescriptionObserver {
 367  public:
 368   explicit CreateAndSetHandler(
 369       rtc::scoped_refptr<PeerConnectionHolder> holder)
 370       : holder_(holder) {
 371   }
 372
 373   void OnSuccess(webrtc::SessionDescriptionInterface* desc) override {
 374     if (holder_->IsDisposed()) return;
 375
 376     type_ = desc->type();
 377     if (desc->ToString(&description_)) {
 378       holder_->connection()->SetLocalDescription(this, desc);
 379     } else {
 380       OnFailure("Can't serialize session description");
 381     }
 382   }
 383
 384   void OnSuccess() override {
 385     if (holder_->IsDisposed()) return;
 386
 387     if (type_ == webrtc::SessionDescriptionInterface::kOffer) {
 388       holder_->delegate()->OnLocalOfferCreatedAndSetSet(description_);
 389     } else {
 390       DCHECK_EQ(webrtc::SessionDescriptionInterface::kAnswer, type_);
 391
 392       holder_->delegate()->OnLocalAnswerCreatedAndSetSet(description_);
 393     }
 394   }
 395
 396   void OnFailure(const std::string& error) override {
 397     if (holder_->IsDisposed()) return;
 398
 399     holder_->delegate()->OnFailure(error);
 400   }
 401
 402  private:
 403   const rtc::scoped_refptr<PeerConnectionHolder> holder_;
 404   std::string type_;
 405   std::string description_;
 406 };
 407
 408 class SetRemoteDescriptionHandler
 409    : public webrtc::SetSessionDescriptionObserver {
 410  public:
 411   SetRemoteDescriptionHandler(
 412       rtc::scoped_refptr<PeerConnectionHolder> holder)
 413       : holder_(holder) {
 414   }
 415
 416   void OnSuccess() override {
 417     if (holder_->IsDisposed()) return;
 418
 419     holder_->delegate()->OnRemoteDescriptionSet();
 420   }
 421
 422   void OnFailure(const std::string& error) override {
 423     if (holder_->IsDisposed()) return;
 424
 425     holder_->delegate()->OnFailure(error);
 426   }
 427
 428  private:
 429   const rtc::scoped_refptr<PeerConnectionHolder> holder_;
 430 };
 431
 432 class PeerConnectionImpl : public AbstractPeerConnection {
 433  public:
 434   PeerConnectionImpl(
 435       SessionDependencyFactory* const factory,
 436       rtc::Thread* signaling_thread,
 437       rtc::scoped_refptr<webrtc::PeerConnectionInterface> connection,
 438       scoped_ptr<PeerConnectionObserverImpl> observer,
 439       scoped_ptr<AbstractPeerConnection::Delegate> delegate)
 440       : factory_(factory),
 441         holder_(new rtc::RefCountedObject<PeerConnectionHolder>(
 442             signaling_thread, connection.get(), delegate.get())),
 443         signaling_thread_(signaling_thread),
 444         connection_(connection),
 445         observer_(observer.Pass()),
 446         delegate_(delegate.Pass()) {
 447   }
 448
 449   ~PeerConnectionImpl() override {
 450     signaling_thread_->Invoke<void>(rtc::Bind(
 451         &PeerConnectionImpl::DisposeOnSignalingThread, this));
 452   }
 453
 454   void CreateAndSetLocalOffer() override {
 455     connection_->CreateOffer(MakeCreateAndSetHandler(), NULL);
 456   }
 457
 458   void CreateAndSetLocalAnswer() override {
 459     connection_->CreateAnswer(MakeCreateAndSetHandler(), NULL);
 460   }
 461
 462   void SetRemoteOffer(const std::string& description) override {
 463     SetRemoteDescription(
 464         webrtc::SessionDescriptionInterface::kOffer, description);
 465   }
 466
 467   void SetRemoteAnswer(const std::string& description) override {
 468     SetRemoteDescription(
 469         webrtc::SessionDescriptionInterface::kAnswer, description);
 470   }
 471
 472   void SetRemoteDescription(
 473       const std::string& type, const std::string& description) {
 474     webrtc::SdpParseError error;
 475     scoped_ptr<webrtc::SessionDescriptionInterface> value(
 476         webrtc::CreateSessionDescription(type, description, &error));
 477     if (value == NULL) {
 478       OnParseError(error);
 479       return;
 480     }
 481     // Takes ownership on |value|.
 482     connection_->SetRemoteDescription(
 483         new rtc::RefCountedObject<SetRemoteDescriptionHandler>(holder_),
 484         value.release());
 485   }
 486
 487   void AddIceCandidate(const std::string& sdp_mid,
 488                        int sdp_mline_index,
 489                        const std::string& sdp) override {
 490     webrtc::SdpParseError error;
 491     auto candidate = webrtc::CreateIceCandidate(
 492         sdp_mid, sdp_mline_index, sdp, &error);
 493     if (candidate == NULL) {
 494       OnParseError(error);
 495       return;
 496     }
 497     // Doesn't takes ownership.
 498     connection_->AddIceCandidate(candidate);
 499     delete candidate;
 500   }
 501
 502   scoped_ptr<AbstractDataChannel> CreateDataChannel(int channelId) override {
 503     webrtc::DataChannelInit init;
 504     init.ordered = true;
 505     init.negotiated = true;
 506     init.id = channelId;
 507
 508     return make_scoped_ptr(new DataChannelImpl(
 509         factory_,
 510         signaling_thread_,
 511         connection_->CreateDataChannel("", &init)));
 512   }
 513
 514  private:
 515   webrtc::CreateSessionDescriptionObserver* MakeCreateAndSetHandler() {
 516     return new rtc::RefCountedObject<CreateAndSetHandler>(holder_);
 517   }
 518
 519   void DisposeOnSignalingThread() {
 520     DCHECK(signaling_thread_->IsCurrent());
 521
 522     CheckedRelease(&connection_);
 523     holder_->Dispose();
 524   }
 525
 526   void OnParseError(const webrtc::SdpParseError& error) {
 527     // TODO(serya): Send on signaling thread.
 528   }
 529
 530   SessionDependencyFactory* const factory_;
 531   const rtc::scoped_refptr<PeerConnectionHolder> holder_;
 532   rtc::Thread* const signaling_thread_;
 533   rtc::scoped_refptr<webrtc::PeerConnectionInterface> connection_;
 534   const scoped_ptr<PeerConnectionObserverImpl> observer_;
 535   const scoped_ptr<AbstractPeerConnection::Delegate> delegate_;
 536 };
 537
 538 class SessionDependencyFactoryImpl : public SessionDependencyFactory {
 539  public:
 540   SessionDependencyFactoryImpl(
 541       const base::Closure& cleanup_on_signaling_thread)
 542       : cleanup_on_signaling_thread_(cleanup_on_signaling_thread) {
 543     signaling_thread_.SetName("signaling_thread", NULL);
 544     signaling_thread_.Start();
 545     worker_thread_.SetName("worker_thread", NULL);
 546     worker_thread_.Start();
 547
 548     factory_ = webrtc::CreatePeerConnectionFactory(
 549         &worker_thread_, &signaling_thread_, NULL, NULL, NULL);
 550   }
 551
 552   ~SessionDependencyFactoryImpl() override {
 553     if (signaling_thread_task_runner_.get())
 554       signaling_thread_task_runner_->Stop();
 555
 556     signaling_thread_.Invoke<void>(rtc::Bind(
 557         &SessionDependencyFactoryImpl::DisposeOnSignalingThread, this));
 558   }
 559
 560   scoped_ptr<AbstractPeerConnection> CreatePeerConnection(
 561       scoped_ptr<RTCConfiguration> config,
 562       scoped_ptr<AbstractPeerConnection::Delegate> delegate) override {
 563     auto observer = make_scoped_ptr(
 564         new PeerConnectionObserverImpl(delegate.get()));
 565
 566     MediaConstraints constraints;
 567     constraints.AddMandatory(
 568         MediaConstraints::kEnableDtlsSrtp, MediaConstraints::kValueTrue);
 569
 570     auto connection = factory_->CreatePeerConnection(
 571         config->impl(), &constraints, NULL, NULL, observer.get());
 572
 573     return make_scoped_ptr(new PeerConnectionImpl(
 574         this, &signaling_thread_, connection, observer.Pass(),
 575         delegate.Pass()));
 576   }
 577
 578   scoped_refptr<base::TaskRunner> signaling_thread_task_runner() override {
 579     if (!signaling_thread_task_runner_.get()) {
 580       signaling_thread_task_runner_ =
 581           new SignalingThreadTaskRunner(&signaling_thread_);
 582     }
 583     return signaling_thread_task_runner_;
 584   }
 585
 586   scoped_refptr<base::TaskRunner> io_thread_task_runner() override {
 587     if (!io_thread_.get()) {
 588       io_thread_.reset(new base::Thread("devtools bridge IO thread"));
 589       base::Thread::Options options;
 590       options.message_loop_type = base::MessageLoop::TYPE_IO;
 591       CHECK(io_thread_->StartWithOptions(options));
 592     }
 593     return io_thread_->task_runner();
 594   }
 595
 596  private:
 597   void DisposeOnSignalingThread() {
 598     DCHECK(signaling_thread_.IsCurrent());
 599     CheckedRelease(&factory_);
 600     if (!cleanup_on_signaling_thread_.is_null())
 601       cleanup_on_signaling_thread_.Run();
 602   }
 603
 604   scoped_ptr<base::Thread> io_thread_;
 605   scoped_refptr<SignalingThreadTaskRunner> signaling_thread_task_runner_;
 606   base::Closure cleanup_on_signaling_thread_;
 607   rtc::Thread signaling_thread_;
 608   rtc::Thread worker_thread_;
 609   rtc::scoped_refptr<webrtc::PeerConnectionFactoryInterface> factory_;
 610 };
 611
 612 } // namespace
 613
 614 // RTCCOnfiguration
 615
 616 // static
 617 scoped_ptr<RTCConfiguration> RTCConfiguration::CreateInstance() {
 618   return make_scoped_ptr(new RTCConfiguration::Impl());
 619 }
 620
 621 // SessionDependencyFactory
 622
 623 // static
 624 bool SessionDependencyFactory::InitializeSSL() {
 625   return rtc::InitializeSSL();
 626 }
 627
 628 // static
 629 bool SessionDependencyFactory::CleanupSSL() {
 630   return rtc::CleanupSSL();
 631 }
 632
 633 // static
 634 scoped_ptr<SessionDependencyFactory> SessionDependencyFactory::CreateInstance(
 635     const base::Closure& cleanup_on_signaling_thread) {
 636   return make_scoped_ptr(new SessionDependencyFactoryImpl(
 637       cleanup_on_signaling_thread));
 638 }
 639
 640 }  // namespace devtools_bridge