Basic implementation of Pepper Transport API.

webkit/plugins/ppapi/ppb_transport_impl.cc

 // Copyright (c) 2010 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 "webkit/plugins/ppapi/ppb_transport_impl.h"
 
-#include "base/singleton.h"
-#include "base/threading/thread_local.h"
+#include "ppapi/c/pp_completion_callback.h"

[brettw, 2011/02/10 07:19:16]

Be sure the file list it alphabetized.

[Sergey Ulanov, 2011/02/10 21:14:54]

Done.

+#include "ppapi/c/pp_errors.h"
+#include "third_party/libjingle/source/talk/p2p/base/p2ptransportchannel.h"
+#include "third_party/libjingle/source/talk/p2p/client/httpportallocator.h"
 #include "ppapi/c/dev/ppb_transport_dev.h"
 #include "webkit/plugins/ppapi/common.h"
 #include "webkit/plugins/ppapi/plugin_module.h"
 #include "webkit/plugins/ppapi/ppapi_plugin_instance.h"
+#include "webkit/plugins/ppapi/var.h"
 
 namespace webkit {
 namespace ppapi {
 
 namespace {
 
-// Creates a new transport object with the specified name
-// using the specified protocol.
-PP_Resource CreateTransport(PP_Instance instance,
-                            const char* name,
+PP_Resource CreateTransport(PP_Instance instance_id, const char* name,
                             const char* proto) {
-  // TODO(juberti): implement me
-  PP_Resource p(0);
-  return p;
+  PluginInstance* instance = ResourceTracker::Get()->GetInstance(instance_id);
+  if (!instance)
+    return 0;
+
+  scoped_refptr<PPB_Transport_Impl> t(new PPB_Transport_Impl(instance));
+  if (!t->Init(name, proto))
+    return 0;
+
+  return t->GetReference();
 }
 
-// Returns whether or not resource is PPB_Transport_Impl
 PP_Bool IsTransport(PP_Resource resource) {
-  return BoolToPPBool(!!Resource::GetAs<PPB_Transport_Impl>(resource));
+  return BoolToPPBool(Resource::GetAs<PPB_Transport_Impl>(resource) != NULL);
 }
 
-// Returns whether the transport is currently writable
-// (i.e. can send data to the remote peer)
-PP_Bool IsWritable(PP_Resource transport) {
-  // TODO(juberti): impelement me
-  return PP_FALSE;
+PP_Bool IsWritable(PP_Resource resource) {
+  scoped_refptr<PPB_Transport_Impl> t(
+      Resource::GetAs<PPB_Transport_Impl>(resource));
+  return BoolToPPBool((t.get()) ? t->IsWritable() : false);
 }
 
-
-// TODO(juberti): other getters/setters
-// connect state
-// connect type, protocol
-// RTT
-
-
-// Establishes a connection to the remote peer.
-// Returns PP_ERROR_WOULDBLOCK and notifies on |cb|
-// when connectivity is established (or timeout occurs).
-int32_t Connect(PP_Resource transport,
-                PP_CompletionCallback cb) {
-  // TODO(juberti): impelement me
-  return 0;
+int32_t Connect(PP_Resource resource, PP_CompletionCallback callback) {
+  scoped_refptr<PPB_Transport_Impl> t(
+      Resource::GetAs<PPB_Transport_Impl>(resource));
+  return (t.get()) ? t->Connect(callback) : PP_ERROR_BADRESOURCE;
 }
 
-
-// Obtains another ICE candidate address to be provided
-// to the remote peer. Returns PP_ERROR_WOULDBLOCK
-// if there are no more addresses to be sent.
-int32_t GetNextAddress(PP_Resource transport,
-                       PP_Var* address,
-                       PP_CompletionCallback cb) {
-  // TODO(juberti): implement me
-  return 0;
+int32_t GetNextAddress(PP_Resource resource, PP_Var* address,

[brettw, 2011/02/10 07:19:16]

I was a little unclear about how this is supposed to work from just looking at
the interface declaration. It seems the address is an out param and that it must
remain valid I take it address is an out param and that the caller must ensure
it remains valid?

I think using this from the caller correctly will be difficult. Often you would
pass it the address of some class member and use the callback factory to make
sure your object doesn't get callbacks after it's destroyed.

I wonder if we can change the interface to make the completion callback just
indicates when you can re-call this function to get a synchronous result? I
think this would be much easier to code to.

[Sergey Ulanov, 2011/02/10 21:14:54]

FileIO API uses exactly the same approach (see PPB_FileIO_Impl::ReadCallback()
and PPB_FileIO_Impl::Read()). Sockets interface in chrome does it too (see
src/net/socket/socket.h). I don't think it makes it much more difficult to use
the interface. In the same time, this approach helps to reduce number of memcpy
for read/write operations.

[brettw, 2011/02/14 17:13:58]

Yes, I think this makes sense for reading since we want to avoid having to copy
the buffer. The same is true for your case which is why I suggested you needed a
Close function rather than changing that case as well (sorry, guess I missed
yours).

In this case, there's no issue with copying, so I'd prefer the version with
slightly easier memory management.

A more compelling reason for changing is that the current API is almost
impossible to write a C++ wrapper around. We'll need to convert a PP_Var* to a
pp::Var*. This may involve cleaning up the ownership of the "old" value of the
existing pp::Var, and this requires us to do some checking at the time of
setting the value. When the value is set at a random time, it's not possible for
us to do this without writing some kind of complicated proxy for the callback
function.

[Sergey Ulanov, 2011/02/14 22:22:55]

Yes, I seen now. Thanks for explaining it. I've also update comments in
ppb_transport_dev.h to reflect this behavior.

+                       PP_CompletionCallback callback) {
+  scoped_refptr<PPB_Transport_Impl> t(
+      Resource::GetAs<PPB_Transport_Impl>(resource));
+  return (t.get())? t->GetNextAddress(address, callback) : PP_ERROR_BADRESOURCE;
 }
 
-
-// Provides an ICE candidate address that was received
-// from the remote peer.
-int32_t ReceiveRemoteAddress(PP_Resource transport,
-                             PP_Var address) {
-  // TODO(juberti): implement me
-  return 0;
+int32_t ReceiveRemoteAddress(PP_Resource resource, PP_Var address) {
+  scoped_refptr<PPB_Transport_Impl> t(
+      Resource::GetAs<PPB_Transport_Impl>(resource));
+  return (t.get())? t->ReceiveRemoteAddress(address) : PP_ERROR_BADRESOURCE;
 }
 
-
-// Like recv(), receives data. Returns PP_ERROR_WOULDBLOCK
-// if there is currently no data to receive.
-int32_t Recv(PP_Resource transport,
-             void* data,
-             uint32_t len,
-             PP_CompletionCallback cb) {
-  // TODO(juberti): implement me
-  return 0;
+int32_t Recv(PP_Resource resource, void* data, uint32_t len,
+             PP_CompletionCallback callback) {
+  scoped_refptr<PPB_Transport_Impl> t(
+      Resource::GetAs<PPB_Transport_Impl>(resource));
+  return (t.get())? t->Recv(data, len, callback) : PP_ERROR_BADRESOURCE;
 }
 
-
-// Like send(), sends data. Returns PP_ERROR_WOULDBLOCK
-// if the socket is currently flow-controlled.
-int32_t Send(PP_Resource transport,
-             const void* data,
-             uint32_t len,
-             PP_CompletionCallback cb) {
-  // TODO(juberti): implement me
-  return 0;
+int32_t Send(PP_Resource resource, const void* data, uint32_t len,
+             PP_CompletionCallback callback) {
+  scoped_refptr<PPB_Transport_Impl> t(
+      Resource::GetAs<PPB_Transport_Impl>(resource));
+  return (t.get())? t->Send(data, len, callback) : PP_ERROR_BADRESOURCE;
 }
 
-
 // Disconnects from the remote peer.
-int32_t Close(PP_Resource transport) {
-  // TODO(juberti): implement me
-  return 0;
+int32_t Close(PP_Resource resource) {
+  scoped_refptr<PPB_Transport_Impl> t(
+      Resource::GetAs<PPB_Transport_Impl>(resource));
+  return (t.get())? t->Close() : PP_ERROR_BADRESOURCE;
 }
 
-
 const PPB_Transport_Dev ppb_transport = {
   &CreateTransport,
   &IsTransport,
   &IsWritable,
   &Connect,
   &GetNextAddress,
   &ReceiveRemoteAddress,
   &Recv,
   &Send,
   &Close,
 };
 
 }  // namespace
 
 PPB_Transport_Impl::PPB_Transport_Impl(PluginInstance* instance)
-    : Resource(instance) {
-  // TODO(juberti): impl
+    : Resource(instance),
+      network_manager_(new talk_base::NetworkManager()),
+      allocator_(new cricket::HttpPortAllocator(network_manager_.get(), "")) {
+  std::vector<talk_base::SocketAddress> stun_hosts;
+  stun_hosts.push_back(talk_base::SocketAddress("stun.l.google.com", 19302));
+  allocator_->SetStunHosts(stun_hosts);
+  // TODO(sergeyu): Use port allocator that works inside sandbox.
+}
+
+PPB_Transport_Impl::~PPB_Transport_Impl() {
 }
 
 const PPB_Transport_Dev* PPB_Transport_Impl::GetInterface() {
   return &ppb_transport;
 }
 
-PPB_Transport_Impl::~PPB_Transport_Impl() {
-  // TODO(juberti): teardown
-}
-
 PPB_Transport_Impl* PPB_Transport_Impl::AsPPB_Transport_Impl() {
   return this;
 }
 
 bool PPB_Transport_Impl::Init(const char* name, const char* proto) {
-  // TODO(juberti): impl
+  // For now, always create http://www.google.com/transport/p2p .
+  channel_.reset(new cricket::P2PTransportChannel(
+      name, "", NULL, allocator_.get()));
+  channel_->SignalRequestSignaling.connect(
+      this, &PPB_Transport_Impl::OnRequestSignaling);
+  channel_->SignalWritableState.connect(
+      this, &PPB_Transport_Impl::OnWriteableState);
+  channel_->SignalCandidateReady.connect(
+      this, &PPB_Transport_Impl::OnCandidateReady);
+  channel_->SignalReadPacket.connect(
+      this, &PPB_Transport_Impl::OnReadPacket);
+  return true;
+}
+
+bool PPB_Transport_Impl::IsWritable() const {
+  return channel_->writable();
+}
+
+int32_t PPB_Transport_Impl::Connect(PP_CompletionCallback callback) {
+  // TODO(juberti): Fail if we're already connected.
+  if (connect_callback_.get() && !connect_callback_->completed())
+    return PP_ERROR_INPROGRESS;
+
+  channel_->Connect();
+
+  PP_Resource resource_id = GetReferenceNoAddRef();
+  CHECK(resource_id);
+  connect_callback_ = new TrackedCompletionCallback(
+      instance()->module()->GetCallbackTracker(), resource_id, callback);
+  return PP_ERROR_WOULDBLOCK;
+}
+
+int32_t PPB_Transport_Impl::GetNextAddress(PP_Var* address,
+                                           PP_CompletionCallback callback) {
+  if (next_address_callback_.get() && !next_address_callback_->completed())
+    return PP_ERROR_INPROGRESS;
+
+  if (!local_candidates_.empty()) {
+    Serialize(local_candidates_.front(), address);
+    local_candidates_.pop_front();
+    return PP_OK;
+  }
+
+  next_address_ = address;
+
+  PP_Resource resource_id = GetReferenceNoAddRef();
+  CHECK(resource_id);
+  next_address_callback_ = new TrackedCompletionCallback(
+      instance()->module()->GetCallbackTracker(), resource_id, callback);
+  return PP_ERROR_WOULDBLOCK;
+}
+
+int32_t PPB_Transport_Impl::ReceiveRemoteAddress(PP_Var address) {
+  cricket::Candidate candidate;
+  if (!Deserialize(address, &candidate)) {
+    return PP_ERROR_FAILED;
+  }
+
+  channel_->OnCandidate(candidate);
+  return PP_OK;
+}
+
+int32_t PPB_Transport_Impl::Recv(void* data, uint32_t len,
+                                 PP_CompletionCallback callback) {
+  if (recv_callback_.get() && !recv_callback_->completed())
+    return PP_ERROR_INPROGRESS;
+
+  // TODO(juberti): Should we store packets that are received when
+  // no callback is installed?
+
+  recv_buffer_ = data;

[brettw, 2011/02/10 07:19:16]

This has the same problem as GetNextAddress with respect to lifetimes. In this
case, it's really being like the URLLoader and we probably don't want to make a
copy. For the URLLoader, we added an explicit Close() function that you can call
to cancel all the pending callbacks. This way, when your object goes out of
scope, you can explicitly ensure that your buffer isn't used regardless of who
else may hold a ref to the Transport object.

[Sergey Ulanov, 2011/02/10 21:14:54]

We do have Close() in this interface. I've added line to abort all pending
callbacks when transport is closed.

+  recv_buffer_size_ = len;
+
+  PP_Resource resource_id = GetReferenceNoAddRef();
+  CHECK(resource_id);
+  recv_callback_ = new TrackedCompletionCallback(
+      instance()->module()->GetCallbackTracker(), resource_id, callback);
+  return PP_ERROR_WOULDBLOCK;
+}
+
+int32_t PPB_Transport_Impl::Send(const void* data, uint32_t len,
+                                 PP_CompletionCallback callback) {
+  return channel_->SendPacket(static_cast<const char*>(data), len);
+}
+
+int32_t PPB_Transport_Impl::Close() {
+  channel_->Reset();
+  return PP_OK;
+}
+
+void PPB_Transport_Impl::OnRequestSignaling() {
+  channel_->OnSignalingReady();
+}
+
+void PPB_Transport_Impl::OnCandidateReady(
+    cricket::TransportChannelImpl* channel,
+    const cricket::Candidate& candidate) {
+  if (next_address_callback_.get() && next_address_callback_->completed()) {
+    scoped_refptr<TrackedCompletionCallback> callback;
+    callback.swap(next_address_callback_);
+    Serialize(candidate, next_address_);
+    callback->Run(PP_OK);
+  } else {
+    local_candidates_.push_back(candidate);
+  }
+}
+
+void PPB_Transport_Impl::OnWriteableState(cricket::TransportChannel* channel) {
+  if (connect_callback_.get() && connect_callback_->completed()) {
+    scoped_refptr<TrackedCompletionCallback> callback;
+    callback.swap(connect_callback_);
+    callback->Run(PP_OK);
+  }
+}
+
+void PPB_Transport_Impl::OnReadPacket(cricket::TransportChannel* channel,
+                                      const char* data, size_t len) {
+  if (recv_callback_.get() && recv_callback_->completed()) {
+    scoped_refptr<TrackedCompletionCallback> callback;
+    callback.swap(recv_callback_);
+
+    if (len <= recv_buffer_size_) {
+      memcpy(recv_buffer_, data, len);
+      callback->Run(PP_OK);
+    } else {
+      callback->Run(PP_ERROR_FAILED);
+    }
+  }
+  // TODO(sergeyu): Buffer incoming packet if there is no pending read.
+}
+
+bool PPB_Transport_Impl::Serialize(const cricket::Candidate& candidate,
+                                   PP_Var* address) {
+  // TODO(juberti): Come up with a real wire format!
+  std::string blob = candidate.ToString();
+  *address = StringVar::StringToPPVar(instance()->module(), blob);
+  return true;
+}
+
+bool PPB_Transport_Impl::Deserialize(PP_Var address,
+                                     cricket::Candidate* candidate) {
+  // TODO(juberti): Implement this.
   return false;
 }
 
 }  // namespace ppapi
 }  // namespace webkit
-