Implement P2P Transport Dev using P2PTransportImpl.

BUG=None
TEST=Unittests

Review URL: http://codereview.chromium.org/6823021

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@81331 0039d316-1c4b-4281-b951-d872f2087c98

19 files changed, 401 insertions, 175 deletions

diff --git a/chrome/test/ui/ppapi_uitest.cc b/chrome/test/ui/ppapi_uitest.cc
index 4496e20..489eeaf 100644
--- a/chrome/test/ui/ppapi_uitest.cc
+++ b/chrome/test/ui/ppapi_uitest.cc
@@ -171,6 +171,6 @@ TEST_F(PPAPITest, DISABLED_DirectoryReader) {
   RunTestViaHTTP("DirectoryReader");
 }
 
-TEST_F(PPAPITest, DISABLED_Transport) {
+TEST_F(PPAPITest, Transport) {
   RunTest("Transport");
 }
diff --git a/content/renderer/p2p/p2p_transport_impl.cc b/content/renderer/p2p/p2p_transport_impl.cc
index d5f2afe..2353134 100644
--- a/content/renderer/p2p/p2p_transport_impl.cc
+++ b/content/renderer/p2p/p2p_transport_impl.cc
@@ -23,6 +23,11 @@ P2PTransportImpl::P2PTransportImpl(
       socket_factory_(socket_factory) {
 }
 
+P2PTransportImpl::P2PTransportImpl(P2PSocketDispatcher* socket_dispatcher)
+    : network_manager_(new IpcNetworkManager(socket_dispatcher)),
+      socket_factory_(new IpcPacketSocketFactory(socket_dispatcher)) {
+}
+
 P2PTransportImpl::~P2PTransportImpl() {
 }
 
@@ -40,8 +45,9 @@ bool P2PTransportImpl::Init(const std::string& name, const std::string& config,
   // TODO(sergeyu): Implement PortAllocator that can parse |config|
   // and use it here instead of BasicPortAllocator.
   allocator_.reset(new cricket::BasicPortAllocator(
-      network_manager_, socket_factory_));
+      network_manager_.get(), socket_factory_.get()));
 
+  DCHECK(!channel_.get());
   channel_.reset(new cricket::P2PTransportChannel(
       name, "", NULL, allocator_.get()));
   channel_->SignalRequestSignaling.connect(
diff --git a/content/renderer/p2p/p2p_transport_impl.h b/content/renderer/p2p/p2p_transport_impl.h
index d473614..2c8b947 100644
--- a/content/renderer/p2p/p2p_transport_impl.h
+++ b/content/renderer/p2p/p2p_transport_impl.h
@@ -10,7 +10,7 @@
 #include "third_party/libjingle/source/talk/base/sigslot.h"
 #include "webkit/glue/p2p_transport.h"
 
-class RenderView;
+class P2PSocketDispatcher;
 
 namespace cricket {
 class Candidate;
@@ -33,11 +33,16 @@ class P2PTransportImpl : public webkit_glue::P2PTransport,
                          public sigslot::has_slots<> {
  public:
   // Create P2PTransportImpl using specified NetworkManager and
-  // PacketSocketFactory. Caller keeps ownership of |network_manager|
-  // and |socket_factory|.
+  // PacketSocketFactory. Takes ownership of |network_manager| and
+  // |socket_factory|.
   P2PTransportImpl(talk_base::NetworkManager* network_manager,
                    talk_base::PacketSocketFactory* socket_factory);
 
+  // Creates P2PTransportImpl using specified
+  // P2PSocketDispatcher. This constructor creates IpcNetworkManager
+  // and IpcPacketSocketFactory, and keeps ownership of these objects.
+  P2PTransportImpl(P2PSocketDispatcher* socket_dispatcher);
+
   virtual ~P2PTransportImpl();
 
   // webkit_glue::P2PTransport interface.
@@ -63,8 +68,8 @@ class P2PTransportImpl : public webkit_glue::P2PTransport,
   EventHandler* event_handler_;
   State state_;
 
-  talk_base::NetworkManager* network_manager_;
-  talk_base::PacketSocketFactory* socket_factory_;
+  scoped_ptr<talk_base::NetworkManager> network_manager_;
+  scoped_ptr<talk_base::PacketSocketFactory> socket_factory_;
 
   scoped_ptr<cricket::PortAllocator> allocator_;
   scoped_ptr<cricket::P2PTransportChannel> channel_;
diff --git a/content/renderer/p2p/p2p_transport_impl_unittest.cc b/content/renderer/p2p/p2p_transport_impl_unittest.cc
index f7b9bfe..393fd0d 100644
--- a/content/renderer/p2p/p2p_transport_impl_unittest.cc
+++ b/content/renderer/p2p/p2p_transport_impl_unittest.cc
@@ -203,28 +203,19 @@ class P2PTransportImplTest : public testing::Test {
 
     net::IPAddressNumber ip;
     ASSERT_TRUE(net::ParseIPLiteralToNumber(kTestAddress1, &ip));
-    network_manager1_.reset(new jingle_glue::FakeNetworkManager(ip));
-    socket_factory1_.reset(
-        new jingle_glue::FakeSocketFactory(socket_manager_, ip));
-    transport1_.reset(new P2PTransportImpl(network_manager1_.get(),
-                                           socket_factory1_.get()));
+    transport1_.reset(new P2PTransportImpl(
+        new jingle_glue::FakeNetworkManager(ip),
+        new jingle_glue::FakeSocketFactory(socket_manager_, ip)));
 
     ASSERT_TRUE(net::ParseIPLiteralToNumber(kTestAddress2, &ip));
-    network_manager2_.reset(new jingle_glue::FakeNetworkManager(ip));
-    socket_factory2_.reset(
-        new jingle_glue::FakeSocketFactory(socket_manager_, ip));
-    transport2_.reset(new P2PTransportImpl(network_manager2_.get(),
-                                           socket_factory2_.get()));
+    transport2_.reset(new P2PTransportImpl(
+        new jingle_glue::FakeNetworkManager(ip),
+        new jingle_glue::FakeSocketFactory(socket_manager_, ip)));
   }
 
   MessageLoop message_loop_;
 
-  scoped_ptr<jingle_glue::FakeNetworkManager> network_manager1_;
-  scoped_ptr<jingle_glue::FakeNetworkManager> network_manager2_;
   scoped_refptr<jingle_glue::FakeSocketManager> socket_manager_;
-  scoped_ptr<jingle_glue::FakeSocketFactory> socket_factory1_;
-  scoped_ptr<jingle_glue::FakeSocketFactory> socket_factory2_;
-
   scoped_ptr<P2PTransportImpl> transport1_;
   MockP2PEventHandler event_handler1_;
   scoped_ptr<P2PTransportImpl> transport2_;
diff --git a/content/renderer/pepper_plugin_delegate_impl.cc b/content/renderer/pepper_plugin_delegate_impl.cc
index f8b5355..eaa98c5 100644
--- a/content/renderer/pepper_plugin_delegate_impl.cc
+++ b/content/renderer/pepper_plugin_delegate_impl.cc
@@ -29,6 +29,7 @@
 #include "content/renderer/content_renderer_client.h"
 #include "content/renderer/renderer_gl_context.h"
 #include "content/renderer/gpu_channel_host.h"
+#include "content/renderer/p2p/p2p_transport_impl.h"
 #include "content/renderer/pepper_platform_context_3d_impl.h"
 #include "content/renderer/render_view.h"
 #include "content/renderer/render_widget_fullscreen_pepper.h"
@@ -924,3 +925,7 @@ void PepperPluginDelegateImpl::HasUnsupportedFeature() {
 P2PSocketDispatcher* PepperPluginDelegateImpl::GetP2PSocketDispatcher() {
   return render_view_->p2p_socket_dispatcher();
 }
+
+webkit_glue::P2PTransport* PepperPluginDelegateImpl::CreateP2PTransport() {
+  return new P2PTransportImpl(render_view_->p2p_socket_dispatcher());
+}
diff --git a/content/renderer/pepper_plugin_delegate_impl.h b/content/renderer/pepper_plugin_delegate_impl.h
index afb4804..7fa2361 100644
--- a/content/renderer/pepper_plugin_delegate_impl.h
+++ b/content/renderer/pepper_plugin_delegate_impl.h
@@ -190,6 +190,7 @@ class PepperPluginDelegateImpl
   virtual void SetContentRestriction(int restrictions);
   virtual void HasUnsupportedFeature();
   virtual P2PSocketDispatcher* GetP2PSocketDispatcher();
+  virtual webkit_glue::P2PTransport* CreateP2PTransport();
 
  private:
   // Pointer to the RenderView that owns us.
diff --git a/jingle/glue/thread_wrapper.cc b/jingle/glue/thread_wrapper.cc
index fe26094..979064b 100644
--- a/jingle/glue/thread_wrapper.cc
+++ b/jingle/glue/thread_wrapper.cc
@@ -30,6 +30,7 @@ JingleThreadWrapper::JingleThreadWrapper(MessageLoop* message_loop)
   DCHECK_EQ(message_loop_, MessageLoop::current());
 
   talk_base::ThreadManager::SetCurrent(this);
+  talk_base::MessageQueueManager::Instance()->Add(this);
   message_loop_->AddDestructionObserver(this);
 }
 
@@ -38,6 +39,7 @@ JingleThreadWrapper::~JingleThreadWrapper() {
 
 void JingleThreadWrapper::WillDestroyCurrentMessageLoop() {
   talk_base::ThreadManager::SetCurrent(NULL);
+  talk_base::MessageQueueManager::Instance()->Remove(this);
   message_loop_->RemoveDestructionObserver(this);
   delete this;
 }
diff --git a/jingle/glue/thread_wrapper_unittest.cc b/jingle/glue/thread_wrapper_unittest.cc
index 18f8bb4..042970e 100644
--- a/jingle/glue/thread_wrapper_unittest.cc
+++ b/jingle/glue/thread_wrapper_unittest.cc
@@ -35,20 +35,15 @@ class ThreadWrapperTest : public testing::Test {
   }
 
   virtual void SetUp() OVERRIDE {
-    wrapper_ = new JingleThreadWrapper(&message_loop_);
+    JingleThreadWrapper::EnsureForCurrentThread();
   }
 
   // ThreadWrapper destroyes itself when |message_loop_| is destroyed.
-  JingleThreadWrapper* wrapper_;
   MessageLoop message_loop_;
   MockMessageHandler handler1_;
   MockMessageHandler handler2_;
 };
 
-TEST_F(ThreadWrapperTest, Create) {
-  EXPECT_EQ(thread(), static_cast<talk_base::Thread*>(wrapper_));
-}
-
 MATCHER_P3(MatchMessageAndDeleteData, handler, message_id, data, "") {
   delete arg->pdata;
   return arg->phandler == handler &&
@@ -156,4 +151,17 @@ TEST_F(ThreadWrapperTest, ClearDelayed) {
   message_loop_.Run();
 }
 
+// Verify that the queue is cleared when a handler is destroyed.
+TEST_F(ThreadWrapperTest, ClearDestoroyed) {
+  MockMessageHandler* handler_ptr;
+  {
+    MockMessageHandler handler;
+    handler_ptr = &handler;
+    thread()->Post(&handler, kTestMessage1, NULL);
+  }
+  talk_base::MessageList removed;
+  thread()->Clear(handler_ptr, talk_base::MQID_ANY, &removed);
+  DCHECK_EQ(0U, removed.size());
+}
+
 }  // namespace jingle_glue
diff --git a/ppapi/tests/test_case.cc b/ppapi/tests/test_case.cc
index ab54d51..bfcfa8c 100644
--- a/ppapi/tests/test_case.cc
+++ b/ppapi/tests/test_case.cc
@@ -9,6 +9,7 @@
 #include "ppapi/tests/test_utils.h"
 #include "ppapi/tests/testing_instance.h"
 
+// static
 std::string TestCase::MakeFailureMessage(const char* file,
                                          int line,
                                          const char* cmd) {
diff --git a/ppapi/tests/test_case.h b/ppapi/tests/test_case.h
index b06ecb8..9ccde68 100644
--- a/ppapi/tests/test_case.h
+++ b/ppapi/tests/test_case.h
@@ -34,7 +34,8 @@ class TestCase {
   // first displayed.
   virtual void RunTest() = 0;
 
-  std::string MakeFailureMessage(const char* file, int line, const char* cmd);
+  static std::string MakeFailureMessage(const char* file, int line,
+                                        const char* cmd);
 
   // Returns the scriptable test object for the current test, if any.
   // Internally, this uses CreateTestObject which each test overrides.
@@ -129,4 +130,3 @@ class TestCaseFactory {
 #define PASS() return std::string()
 
 #endif  // PPAPI_TESTS_TEST_CASE_H_
-
diff --git a/ppapi/tests/test_transport.cc b/ppapi/tests/test_transport.cc
index 4035676..bcf9701 100644
--- a/ppapi/tests/test_transport.cc
+++ b/ppapi/tests/test_transport.cc
@@ -4,9 +4,11 @@
 
 #include "ppapi/tests/test_transport.h"
 
+#include <stdlib.h>
 #include <string.h>
 
-#include <string>
+#include <list>
+#include <map>
 
 #include "ppapi/c/dev/ppb_testing_dev.h"
 #include "ppapi/c/pp_errors.h"
@@ -19,6 +21,72 @@
 
 REGISTER_TEST_CASE(Transport);
 
+#define RUN_SUBTEST(function) { \
+    std::string result = function(); \
+    if (!result.empty()) \
+      return result; \
+  }
+
+namespace {
+
+const char kTestChannelName[] = "test";
+const int kNumPackets = 100;
+const int kSendBufferSize = 1200;
+const int kReadBufferSize = 65536;
+
+class StreamReader {
+ public:
+  explicit StreamReader(pp::Transport_Dev* transport)
+      : ALLOW_THIS_IN_INITIALIZER_LIST(callback_factory_(this)),
+        transport_(transport),
+        errors_(0) {
+    Read();
+  }
+
+  const std::list<std::vector<char> >& received() { return received_; }
+  std::list<std::string> errors() { return errors_; }
+
+ private:
+  void Read() {
+    while (true) {
+      buffer_.resize(kReadBufferSize);
+      int result = transport_->Recv(
+          &buffer_[0], buffer_.size(),
+          callback_factory_.NewCallback(&StreamReader::OnReadFinished));
+      if (result > 0)
+        DidFinishRead(result);
+      else
+        break;
+    }
+  }
+
+  void OnReadFinished(int result) {
+    DidFinishRead(result);
+    if (result > 0)
+      Read();
+  }
+
+  void DidFinishRead(int result) {
+    if (result > 0) {
+      if (result > static_cast<int>(buffer_.size())) {
+        errors_.push_back(TestCase::MakeFailureMessage(
+            __FILE__, __LINE__,
+            "Recv() returned value that is bigger than the buffer."));
+      }
+      buffer_.resize(result);
+      received_.push_back(buffer_);
+    }
+  }
+
+  pp::CompletionCallbackFactory<StreamReader> callback_factory_;
+  pp::Transport_Dev* transport_;
+  std::vector<char> buffer_;
+  std::list<std::vector<char> > received_;
+  std::list<std::string> errors_;
+};
+
+}  // namespace
+
 bool TestTransport::Init() {
   transport_interface_ = reinterpret_cast<PPB_Transport_Dev const*>(
       pp::Module::Get()->GetBrowserInterface(PPB_TRANSPORT_DEV_INTERFACE));
@@ -26,42 +94,135 @@ bool TestTransport::Init() {
 }
 
 void TestTransport::RunTest() {
-  RUN_TEST(Basics);
-  // TODO(juberti): more Transport tests here...
+  RUN_TEST(Create);
+  RUN_TEST(Connect);
+  RUN_TEST(SendData);
+  RUN_TEST(ConnectAndClose);
 }
 
-std::string TestTransport::TestBasics() {
-  pp::Transport_Dev transport1(instance_, "test", "");
-  pp::Transport_Dev transport2(instance_, "test", "");
+std::string TestTransport::InitTargets() {
+  transport1_.reset(new pp::Transport_Dev(instance_, kTestChannelName, ""));
+  transport2_.reset(new pp::Transport_Dev(instance_, kTestChannelName, ""));
+
+  ASSERT_TRUE(transport1_.get() != NULL);
+  ASSERT_TRUE(transport2_.get() != NULL);
 
+  PASS();
+}
+
+std::string TestTransport::Connect() {
   TestCompletionCallback connect_cb1(instance_->pp_instance());
   TestCompletionCallback connect_cb2(instance_->pp_instance());
-  ASSERT_EQ(transport1.Connect(connect_cb1), PP_OK_COMPLETIONPENDING);
-  ASSERT_EQ(transport2.Connect(connect_cb2), PP_OK_COMPLETIONPENDING);
+  ASSERT_EQ(transport1_->Connect(connect_cb1), PP_OK_COMPLETIONPENDING);
+  ASSERT_EQ(transport2_->Connect(connect_cb2), PP_OK_COMPLETIONPENDING);
 
   pp::Var address1;
   pp::Var address2;
   TestCompletionCallback next_address_cb1(instance_->pp_instance());
   TestCompletionCallback next_address_cb2(instance_->pp_instance());
-  ASSERT_EQ(transport1.GetNextAddress(&address1, next_address_cb1),
+  ASSERT_EQ(transport1_->GetNextAddress(&address1, next_address_cb1),
             PP_OK_COMPLETIONPENDING);
-  ASSERT_EQ(transport2.GetNextAddress(&address2, next_address_cb2),
+  ASSERT_EQ(transport2_->GetNextAddress(&address2, next_address_cb2),
             PP_OK_COMPLETIONPENDING);
-  ASSERT_EQ(next_address_cb1.WaitForResult(), 0);
-  ASSERT_EQ(next_address_cb2.WaitForResult(), 0);
-  ASSERT_EQ(transport1.GetNextAddress(&address1, next_address_cb1), PP_OK);
-  ASSERT_EQ(transport2.GetNextAddress(&address2, next_address_cb2), PP_OK);
+  ASSERT_EQ(next_address_cb1.WaitForResult(), PP_OK);
+  ASSERT_EQ(next_address_cb2.WaitForResult(), PP_OK);
+  ASSERT_EQ(transport1_->GetNextAddress(&address1, next_address_cb1), PP_OK);
+  ASSERT_EQ(transport2_->GetNextAddress(&address2, next_address_cb2), PP_OK);
+
+  ASSERT_EQ(transport1_->ReceiveRemoteAddress(address2), PP_OK);
+  ASSERT_EQ(transport2_->ReceiveRemoteAddress(address1), PP_OK);
+
+  ASSERT_EQ(connect_cb1.WaitForResult(), PP_OK);
+  ASSERT_EQ(connect_cb2.WaitForResult(), PP_OK);
+
+  ASSERT_TRUE(transport1_->IsWritable());
+  ASSERT_TRUE(transport2_->IsWritable());
+
+  PASS();
+}
+
+std::string TestTransport::Clean() {
+  transport1_.reset();
+  transport2_.reset();
+
+  PASS();
+}
+
+std::string TestTransport::TestCreate() {
+  RUN_SUBTEST(InitTargets);
+
+  Clean();
+
+  PASS();
+}
+
+std::string TestTransport::TestConnect() {
+  RUN_SUBTEST(InitTargets);
+  RUN_SUBTEST(Connect);
+
+  Clean();
+
+  PASS();
+}
+
+std::string TestTransport::TestSendData() {
+  RUN_SUBTEST(InitTargets);
+  RUN_SUBTEST(Connect);
+
+  StreamReader reader(transport1_.get());
+
+  std::map<int, std::vector<char> > sent_packets;
+  for (int i = 0; i < kNumPackets; ++i) {
+    std::vector<char> send_buffer(kSendBufferSize);
+    for (size_t j = 0; j < send_buffer.size(); ++j) {
+      send_buffer[j] = rand() % 256;
+    }
+    // Put packet index in the beginning.
+    memcpy(&send_buffer[0], &i, sizeof(i));
+
+    TestCompletionCallback send_cb(instance_->pp_instance());
+    ASSERT_EQ(
+        transport2_->Send(&send_buffer[0], send_buffer.size(), send_cb),
+        static_cast<int>(send_buffer.size()));
+    sent_packets[i] = send_buffer;
+  }
+
+  // Wait for 1 second.
+  TestCompletionCallback wait_cb(instance_->pp_instance());
+  pp::Module::Get()->core()->CallOnMainThread(1000, wait_cb);
+  ASSERT_EQ(wait_cb.WaitForResult(), PP_OK);
+
+  ASSERT_TRUE(reader.errors().size() == 0);
+  ASSERT_TRUE(reader.received().size() > 0);
+  for (std::list<std::vector<char> >::const_iterator it =
+           reader.received().begin(); it != reader.received().end(); ++it) {
+    int index;
+    memcpy(&index, &(*it)[0], sizeof(index));
+    ASSERT_TRUE(sent_packets[index] == *it);
+  }
+
+  Clean();
+
+  PASS();
+}
+
+std::string TestTransport::TestConnectAndClose() {
+  RUN_SUBTEST(InitTargets);
+  RUN_SUBTEST(Connect);
 
-  ASSERT_EQ(transport1.ReceiveRemoteAddress(address2), 0);
-  ASSERT_EQ(transport2.ReceiveRemoteAddress(address1), 0);
+  std::vector<char> recv_buffer(kReadBufferSize);
+  TestCompletionCallback recv_cb(instance_->pp_instance());
+  ASSERT_EQ(
+      transport1_->Recv(&recv_buffer[0], recv_buffer.size(), recv_cb),
+      PP_OK_COMPLETIONPENDING);
 
-  ASSERT_EQ(connect_cb1.WaitForResult(), 0);
-  ASSERT_EQ(connect_cb2.WaitForResult(), 0);
+  // Close the transport and verify that callback is aborted.
+  ASSERT_EQ(transport1_->Close(), PP_OK);
 
-  ASSERT_TRUE(transport1.IsWritable());
-  ASSERT_TRUE(transport2.IsWritable());
+  ASSERT_EQ(recv_cb.run_count(), 1);
+  ASSERT_EQ(recv_cb.result(), PP_ERROR_ABORTED);
 
-  // TODO(sergeyu): Verify that data can be sent/received.
+  Clean();
 
   PASS();
 }
diff --git a/ppapi/tests/test_transport.h b/ppapi/tests/test_transport.h
index 43e4306..f883b6a 100644
--- a/ppapi/tests/test_transport.h
+++ b/ppapi/tests/test_transport.h
@@ -7,13 +7,14 @@
 
 #include <string>
 
+#include "base/scoped_ptr.h"
 #include "ppapi/tests/test_case.h"
 
 struct PPB_Transport_Dev;
 
 namespace pp {
-class Transport;
-}
+class Transport_Dev;
+}  // namespace pp
 
 class TestTransport : public TestCase {
  public:
@@ -24,10 +25,20 @@ class TestTransport : public TestCase {
   virtual void RunTest();
 
  private:
-  std::string TestBasics();
+  std::string InitTargets();
+  std::string Connect();
+  std::string Clean();
+
+  std::string TestCreate();
+  std::string TestConnect();
+  std::string TestSendData();
+  std::string TestConnectAndClose();
 
   // Used by the tests that access the C API directly.
   const PPB_Transport_Dev* transport_interface_;
+
+  scoped_ptr<pp::Transport_Dev> transport1_;
+  scoped_ptr<pp::Transport_Dev> transport2_;
 };
 
 #endif  // PPAPI_TESTS_TEST_TRANSPORT_H_
diff --git a/ppapi/tests/test_utils.h b/ppapi/tests/test_utils.h
index 6fde9bf..b71cc05 100644
--- a/ppapi/tests/test_utils.h
+++ b/ppapi/tests/test_utils.h
@@ -26,6 +26,8 @@ class TestCompletionCallback {
   unsigned run_count() const { return run_count_; }
   void reset_run_count() { run_count_ = 0; }
 
+  int32_t result() const { return result_; }
+
  private:
   static void Handler(void* user_data, int32_t result);
 
diff --git a/webkit/glue/webkit_glue.gypi b/webkit/glue/webkit_glue.gypi
index 3b54df3..16fe656 100644
--- a/webkit/glue/webkit_glue.gypi
+++ b/webkit/glue/webkit_glue.gypi
@@ -115,8 +115,6 @@
         '<(DEPTH)/skia/skia.gyp:skia',
         '<(DEPTH)/third_party/icu/icu.gyp:icui18n',
         '<(DEPTH)/third_party/icu/icu.gyp:icuuc',
-        '<(DEPTH)/third_party/libjingle/libjingle.gyp:libjingle',
-        '<(DEPTH)/third_party/libjingle/libjingle.gyp:libjingle_p2p',
         '<(DEPTH)/third_party/npapi/npapi.gyp:npapi',
         '<(DEPTH)/ppapi/ppapi.gyp:ppapi_c',
         'webkit_resources',
diff --git a/webkit/plugins/ppapi/mock_plugin_delegate.cc b/webkit/plugins/ppapi/mock_plugin_delegate.cc
index 388b6af..f81e9c2 100644
--- a/webkit/plugins/ppapi/mock_plugin_delegate.cc
+++ b/webkit/plugins/ppapi/mock_plugin_delegate.cc
@@ -224,5 +224,9 @@ P2PSocketDispatcher* MockPluginDelegate::GetP2PSocketDispatcher() {
   return NULL;
 }
 
+webkit_glue::P2PTransport* MockPluginDelegate::CreateP2PTransport() {
+  return NULL;
+}
+
 }  // namespace ppapi
 }  // namespace webkit
diff --git a/webkit/plugins/ppapi/mock_plugin_delegate.h b/webkit/plugins/ppapi/mock_plugin_delegate.h
index ce172a0..dd5e8f1 100644
--- a/webkit/plugins/ppapi/mock_plugin_delegate.h
+++ b/webkit/plugins/ppapi/mock_plugin_delegate.h
@@ -98,6 +98,7 @@ class MockPluginDelegate : public PluginDelegate {
   virtual void SetContentRestriction(int restrictions);
   virtual void HasUnsupportedFeature();
   virtual P2PSocketDispatcher* GetP2PSocketDispatcher();
+  virtual webkit_glue::P2PTransport* CreateP2PTransport();
 };
 
 }  // namespace ppapi
diff --git a/webkit/plugins/ppapi/plugin_delegate.h b/webkit/plugins/ppapi/plugin_delegate.h
index 62ad9c1..33fb820 100644
--- a/webkit/plugins/ppapi/plugin_delegate.h
+++ b/webkit/plugins/ppapi/plugin_delegate.h
@@ -54,6 +54,10 @@ class WebFileChooserCompletion;
 struct WebFileChooserParams;
 }
 
+namespace webkit_glue {
+class P2PTransport;
+}  // namespace webkit_glue
+
 struct PP_Flash_NetAddress;
 struct PP_VideoDecoderConfig_Dev;
 
@@ -368,9 +372,12 @@ class PluginDelegate {
   // Socket dispatcher for P2P connections. Returns to NULL if P2P API
   // is disabled.
   //
-  // TODO(sergeyu): Replace this with a higher-level P2P API
-  // implementation.
+  // TODO(sergeyu): Stop using GetP2PSocketDispatcher() in remoting
+  // client and remove it from here.
   virtual P2PSocketDispatcher* GetP2PSocketDispatcher() = 0;
+
+  // Creates P2PTransport object.
+  virtual webkit_glue::P2PTransport* CreateP2PTransport() = 0;
 };
 
 }  // namespace ppapi
diff --git a/webkit/plugins/ppapi/ppb_transport_impl.cc b/webkit/plugins/ppapi/ppb_transport_impl.cc
index c4d440e..3f48721 100644
--- a/webkit/plugins/ppapi/ppb_transport_impl.cc
+++ b/webkit/plugins/ppapi/ppb_transport_impl.cc
@@ -4,12 +4,13 @@
 
 #include "webkit/plugins/ppapi/ppb_transport_impl.h"
 
+#include "base/message_loop.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/socket/socket.h"
 #include "ppapi/c/dev/ppb_transport_dev.h"
 #include "ppapi/c/pp_completion_callback.h"
 #include "ppapi/c/pp_errors.h"
-#include "third_party/libjingle/source/talk/base/basicpacketsocketfactory.h"
-#include "third_party/libjingle/source/talk/p2p/base/p2ptransportchannel.h"
-#include "third_party/libjingle/source/talk/p2p/client/httpportallocator.h"
 #include "webkit/plugins/ppapi/common.h"
 #include "webkit/plugins/ppapi/plugin_module.h"
 #include "webkit/plugins/ppapi/ppapi_plugin_instance.h"
@@ -95,22 +96,32 @@ const PPB_Transport_Dev ppb_transport = {
   &Close,
 };
 
+int MapNetError(int result) {
+  if (result > 0)
+    return result;
+
+  switch (result) {
+    case net::OK:
+      return PP_OK;
+    case net::ERR_IO_PENDING:
+      return PP_OK_COMPLETIONPENDING;
+    case net::ERR_INVALID_ARGUMENT:
+      return PP_ERROR_BADARGUMENT;
+    default:
+      return PP_ERROR_FAILED;
+  }
+}
+
 }  // namespace
 
 PPB_Transport_Impl::PPB_Transport_Impl(PluginInstance* instance)
     : Resource(instance),
-      network_manager_(new talk_base::NetworkManager()),
-      // TODO(sergeyu): Use IpcPacketSocketFactory here when it is
-      // implemented, and when we have talk_base::Thread wrapper for
-      // Chromium threads.
-      socket_factory_(new talk_base::BasicPacketSocketFactory(
-          talk_base::Thread::Current())),
-      allocator_(new cricket::HttpPortAllocator(
-          network_manager_.get(), socket_factory_.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.
+      started_(false),
+      writable_(false),
+      ALLOW_THIS_IN_INITIALIZER_LIST(
+          channel_write_callback_(this, &PPB_Transport_Impl::OnWritten)),
+      ALLOW_THIS_IN_INITIALIZER_LIST(
+          channel_read_callback_(this, &PPB_Transport_Impl::OnRead)) {
 }
 
 PPB_Transport_Impl::~PPB_Transport_Impl() {
@@ -125,30 +136,33 @@ PPB_Transport_Impl* PPB_Transport_Impl::AsPPB_Transport_Impl() {
 }
 
 bool PPB_Transport_Impl::Init(const char* name, const char* proto) {
-  // 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;
+  name_ = name;
+  proto_ = proto;
+  p2p_transport_.reset(instance()->delegate()->CreateP2PTransport());
+  return p2p_transport_.get() != NULL;
 }
 
 bool PPB_Transport_Impl::IsWritable() const {
-  return channel_->writable();
+  if (!p2p_transport_.get())
+    return false;
+
+  return 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())
+  if (!p2p_transport_.get())
+    return PP_ERROR_FAILED;
+
+  // TODO(sergeyu): Use |proto_| here.
+
+  // Connect() has already been called.
+  if (started_)
     return PP_ERROR_INPROGRESS;
 
-  channel_->Connect();
+  if (!p2p_transport_->Init(name_, "", this))
+    return PP_ERROR_FAILED;
+
+  started_ = true;
 
   PP_Resource resource_id = GetReferenceNoAddRef();
   CHECK(resource_id);
@@ -159,11 +173,15 @@ int32_t PPB_Transport_Impl::Connect(PP_CompletionCallback callback) {
 
 int32_t PPB_Transport_Impl::GetNextAddress(PP_Var* address,
                                            PP_CompletionCallback callback) {
+  if (!p2p_transport_.get())
+    return PP_ERROR_FAILED;
+
   if (next_address_callback_.get() && !next_address_callback_->completed())
     return PP_ERROR_INPROGRESS;
 
   if (!local_candidates_.empty()) {
-    Serialize(local_candidates_.front(), address);
+    *address = StringVar::StringToPPVar(instance()->module(),
+                                        local_candidates_.front());
     local_candidates_.pop_front();
     return PP_OK;
   }
@@ -176,98 +194,111 @@ int32_t PPB_Transport_Impl::GetNextAddress(PP_Var* address,
 }
 
 int32_t PPB_Transport_Impl::ReceiveRemoteAddress(PP_Var address) {
-  cricket::Candidate candidate;
-  if (!Deserialize(address, &candidate)) {
+  if (!p2p_transport_.get())
     return PP_ERROR_FAILED;
-  }
 
-  channel_->OnCandidate(candidate);
-  return PP_OK;
+  scoped_refptr<StringVar> address_str = StringVar::FromPPVar(address);
+  if (!address_str)
+    return PP_ERROR_BADARGUMENT;
+
+  return p2p_transport_->AddRemoteCandidate(address_str->value()) ?
+      PP_OK : PP_ERROR_FAILED;
 }
 
 int32_t PPB_Transport_Impl::Recv(void* data, uint32_t len,
                                  PP_CompletionCallback callback) {
+  if (!p2p_transport_.get())
+    return PP_ERROR_FAILED;
+
   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?
+  net::Socket* channel = p2p_transport_->GetChannel();
+  if (!channel)
+    return PP_ERROR_FAILED;
 
-  recv_buffer_ = data;
-  recv_buffer_size_ = len;
+  scoped_refptr<net::IOBuffer> buffer =
+      new net::WrappedIOBuffer(static_cast<const char*>(data));
+  int result = MapNetError(channel->Read(buffer, len, &channel_read_callback_));
+  if (result == PP_OK_COMPLETIONPENDING) {
+    PP_Resource resource_id = GetReferenceNoAddRef();
+    CHECK(resource_id);
+    recv_callback_ = new TrackedCompletionCallback(
+        instance()->module()->GetCallbackTracker(), resource_id, callback);
+  }
 
-  PP_Resource resource_id = GetReferenceNoAddRef();
-  CHECK(resource_id);
-  recv_callback_ = new TrackedCompletionCallback(
-      instance()->module()->GetCallbackTracker(), resource_id, callback);
-  return PP_OK_COMPLETIONPENDING;
+  return result;
 }
 
 int32_t PPB_Transport_Impl::Send(const void* data, uint32_t len,
                                  PP_CompletionCallback callback) {
-  return channel_->SendPacket(static_cast<const char*>(data), len);
+  if (!p2p_transport_.get())
+    return PP_ERROR_FAILED;
+
+  if (send_callback_.get() && !send_callback_->completed())
+    return PP_ERROR_INPROGRESS;
+
+  net::Socket* channel = p2p_transport_->GetChannel();
+  if (!channel)
+    return PP_ERROR_FAILED;
+
+  scoped_refptr<net::IOBuffer> buffer =
+      new net::WrappedIOBuffer(static_cast<const char*>(data));
+  int result = MapNetError(channel->Write(buffer, len,
+                                          &channel_write_callback_));
+  if (result == PP_OK_COMPLETIONPENDING) {
+    PP_Resource resource_id = GetReferenceNoAddRef();
+    CHECK(resource_id);
+    send_callback_ = new TrackedCompletionCallback(
+        instance()->module()->GetCallbackTracker(), resource_id, callback);
+  }
+
+  return result;
 }
 
 int32_t PPB_Transport_Impl::Close() {
-  channel_->Reset();
+  if (!p2p_transport_.get())
+    return PP_ERROR_FAILED;
+
+  p2p_transport_.reset();
   instance()->module()->GetCallbackTracker()->AbortAll();
   return PP_OK;
 }
 
-void PPB_Transport_Impl::OnRequestSignaling() {
-  channel_->OnSignalingReady();
-}
-
-void PPB_Transport_Impl::OnCandidateReady(
-    cricket::TransportChannelImpl* channel,
-    const cricket::Candidate& candidate) {
+void PPB_Transport_Impl::OnCandidateReady(const std::string& address) {
   // Store the candidate first before calling the callback.
-  local_candidates_.push_back(candidate);
+  local_candidates_.push_back(address);
 
-  if (next_address_callback_.get() && next_address_callback_->completed()) {
+  if (next_address_callback_.get() && !next_address_callback_->completed()) {
     scoped_refptr<TrackedCompletionCallback> callback;
     callback.swap(next_address_callback_);
     callback->Run(PP_OK);
   }
 }
 
-void PPB_Transport_Impl::OnWriteableState(cricket::TransportChannel* channel) {
-  if (connect_callback_.get() && connect_callback_->completed()) {
+void PPB_Transport_Impl::OnStateChange(webkit_glue::P2PTransport::State state) {
+  writable_ = (state | webkit_glue::P2PTransport::STATE_WRITABLE) != 0;
+  if (writable_ && 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.
-}
+void PPB_Transport_Impl::OnRead(int result) {
+  DCHECK(recv_callback_.get() && !recv_callback_->completed());
 
-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();
-  Var::PluginReleasePPVar(*address);
-  *address = StringVar::StringToPPVar(instance()->module(), blob);
-  return true;
+  scoped_refptr<TrackedCompletionCallback> callback;
+  callback.swap(recv_callback_);
+  callback->Run(MapNetError(result));
 }
 
-bool PPB_Transport_Impl::Deserialize(PP_Var address,
-                                     cricket::Candidate* candidate) {
-  // TODO(juberti): Implement this.
-  return false;
+void PPB_Transport_Impl::OnWritten(int result) {
+  DCHECK(send_callback_.get() && !send_callback_->completed());
+
+  scoped_refptr<TrackedCompletionCallback> callback;
+  callback.swap(send_callback_);
+  callback->Run(MapNetError(result));
 }
 
 }  // namespace ppapi
diff --git a/webkit/plugins/ppapi/ppb_transport_impl.h b/webkit/plugins/ppapi/ppb_transport_impl.h
index 1a7ce1a..c368b07 100644
--- a/webkit/plugins/ppapi/ppb_transport_impl.h
+++ b/webkit/plugins/ppapi/ppb_transport_impl.h
@@ -5,30 +5,22 @@
 #ifndef WEBKIT_PLUGINS_PPAPI_PPB_TRANSPORT_IMPL_H_
 #define WEBKIT_PLUGINS_PPAPI_PPB_TRANSPORT_IMPL_H_
 
+#include <list>
+#include <string>
+
 #include "base/basictypes.h"
 #include "base/memory/scoped_ptr.h"
+#include "net/base/completion_callback.h"
 #include "ppapi/c/dev/ppb_transport_dev.h"
-#include "third_party/libjingle/source/talk/base/sigslot.h"
-#include "third_party/libjingle/source/talk/p2p/base/candidate.h"
+#include "webkit/glue/p2p_transport.h"
 #include "webkit/plugins/ppapi/callbacks.h"
 #include "webkit/plugins/ppapi/resource.h"
 
-namespace talk_base {
-class NetworkManager;
-class PacketSocketFactory;
-}  // namespace talk_base
-
-namespace cricket {
-class HttpPortAllocator;
-class P2PTransportChannel;
-class TransportChannel;
-class TransportChannelImpl;
-}  // namespace cricket
-
 namespace webkit {
 namespace ppapi {
 
-class PPB_Transport_Impl : public Resource, public sigslot::has_slots<> {
+class PPB_Transport_Impl : public Resource,
+                           public webkit_glue::P2PTransport::EventHandler {
  public:
   static const PPB_Transport_Dev* GetInterface();
 
@@ -38,7 +30,7 @@ class PPB_Transport_Impl : public Resource, public sigslot::has_slots<> {
   bool Init(const char* name, const char* proto);
 
   // Resource override.
-  virtual PPB_Transport_Impl* AsPPB_Transport_Impl();
+  virtual PPB_Transport_Impl* AsPPB_Transport_Impl() OVERRIDE;
 
   bool IsWritable() const;
   int32_t Connect(PP_CompletionCallback cb);
@@ -48,29 +40,29 @@ class PPB_Transport_Impl : public Resource, public sigslot::has_slots<> {
   int32_t Send(const void* data, uint32_t len, PP_CompletionCallback cb);
   int32_t Close();
 
- private:
-  void OnRequestSignaling();
-  void OnCandidateReady(cricket::TransportChannelImpl* channel,
-                        const cricket::Candidate& candidate);
-  void OnWriteableState(cricket::TransportChannel*);
-  void OnReadPacket(cricket::TransportChannel*, const char*, size_t);
+  // webkit_glue::P2PTransport::EventHandler implementation.
+  virtual void OnCandidateReady(const std::string& address) OVERRIDE;
+  virtual void OnStateChange(webkit_glue::P2PTransport::State state) OVERRIDE;
 
-  bool Serialize(const cricket::Candidate& candidate, PP_Var* address);
-  bool Deserialize(PP_Var address, cricket::Candidate* candidate);
+ private:
+  void OnRead(int result);
+  void OnWritten(int result);
 
-  scoped_ptr<talk_base::NetworkManager> network_manager_;
-  scoped_ptr<talk_base::PacketSocketFactory> socket_factory_;
-  scoped_ptr<cricket::HttpPortAllocator> allocator_;
-  scoped_ptr<cricket::P2PTransportChannel> channel_;
-  std::list<cricket::Candidate> local_candidates_;
+  std::string name_;
+  std::string proto_;
+  bool started_;
+  scoped_ptr<webkit_glue::P2PTransport> p2p_transport_;
+  bool writable_;
+  std::list<std::string> local_candidates_;
 
   scoped_refptr<TrackedCompletionCallback> connect_callback_;
-
   scoped_refptr<TrackedCompletionCallback> next_address_callback_;
 
   scoped_refptr<TrackedCompletionCallback> recv_callback_;
-  void* recv_buffer_;
-  uint32_t recv_buffer_size_;
+  scoped_refptr<TrackedCompletionCallback> send_callback_;
+
+  net::CompletionCallbackImpl<PPB_Transport_Impl> channel_write_callback_;
+  net::CompletionCallbackImpl<PPB_Transport_Impl> channel_read_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(PPB_Transport_Impl);
 };