Add P2PDatagramSocket and P2PStreamSocket interfaces.

remoting/protocol/channel_multiplexer.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 "remoting/protocol/channel_multiplexer.h"
 
 #include <string.h>
 
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/callback_helpers.h"
 #include "base/location.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/thread_task_runner_handle.h"
 #include "net/base/net_errors.h"
-#include "net/socket/stream_socket.h"
 #include "remoting/protocol/message_serialization.h"
+#include "remoting/protocol/p2p_stream_socket.h"
 
 namespace remoting {
 namespace protocol {
 
 namespace {
 const int kChannelIdUnknown = -1;
 const int kMaxPacketSize = 1024;
 
 class PendingPacket {
  public:
@@ skipping 41 matching lines @@
  public:
   MuxChannel(ChannelMultiplexer* multiplexer, const std::string& name,
              int send_id);
   ~MuxChannel();
 
   const std::string& name() { return name_; }
   int receive_id() { return receive_id_; }
   void set_receive_id(int id) { receive_id_ = id; }
 
   // Called by ChannelMultiplexer.
-  scoped_ptr<net::StreamSocket> CreateSocket();
+  scoped_ptr<P2PStreamSocket> CreateSocket();
   void OnIncomingPacket(scoped_ptr<MultiplexPacket> packet,
                         const base::Closure& done_task);
   void OnBaseChannelError(int error);
 
   // Called by MuxSocket.
   void OnSocketDestroyed();
   bool DoWrite(scoped_ptr<MultiplexPacket> packet,
                const base::Closure& done_task);
-  int DoRead(net::IOBuffer* buffer, int buffer_len);
+  int DoRead(const scoped_refptr<net::IOBuffer>& buffer, int buffer_len);
 
  private:
   ChannelMultiplexer* multiplexer_;
   std::string name_;
   int send_id_;
   bool id_sent_;
   int receive_id_;
   MuxSocket* socket_;
   std::list<PendingPacket*> pending_packets_;
 
   DISALLOW_COPY_AND_ASSIGN(MuxChannel);
 };
 
-class ChannelMultiplexer::MuxSocket : public net::StreamSocket,
+class ChannelMultiplexer::MuxSocket : public P2PStreamSocket,
                                       public base::NonThreadSafe,
                                       public base::SupportsWeakPtr<MuxSocket> {
  public:
   MuxSocket(MuxChannel* channel);
   ~MuxSocket() override;
 
   void OnWriteComplete();
   void OnBaseChannelError(int error);
   void OnPacketReceived();
 
-  // net::StreamSocket interface.
-  int Read(net::IOBuffer* buffer,
-           int buffer_len,
+  // P2PStreamSocket interface.
+  int Read(const scoped_refptr<net::IOBuffer>& buffer, int buffer_len,
            const net::CompletionCallback& callback) override;
-  int Write(net::IOBuffer* buffer,
-            int buffer_len,
+  int Write(const scoped_refptr<net::IOBuffer>& buffer, int buffer_len,
             const net::CompletionCallback& callback) override;
 
-  int SetReceiveBufferSize(int32 size) override {
-    NOTIMPLEMENTED();
-    return net::ERR_NOT_IMPLEMENTED;
-  }
-  int SetSendBufferSize(int32 size) override {
-    NOTIMPLEMENTED();
-    return net::ERR_NOT_IMPLEMENTED;
-  }
-
-  int Connect(const net::CompletionCallback& callback) override {
-    NOTIMPLEMENTED();
-    return net::ERR_NOT_IMPLEMENTED;
-  }
-  void Disconnect() override { NOTIMPLEMENTED(); }
-  bool IsConnected() const override {
-    NOTIMPLEMENTED();
-    return true;
-  }
-  bool IsConnectedAndIdle() const override {
-    NOTIMPLEMENTED();
-    return false;
-  }
-  int GetPeerAddress(net::IPEndPoint* address) const override {
-    NOTIMPLEMENTED();
-    return net::ERR_NOT_IMPLEMENTED;
-  }
-  int GetLocalAddress(net::IPEndPoint* address) const override {
-    NOTIMPLEMENTED();
-    return net::ERR_NOT_IMPLEMENTED;
-  }
-  const net::BoundNetLog& NetLog() const override {
-    NOTIMPLEMENTED();
-    return net_log_;
-  }
-  void SetSubresourceSpeculation() override { NOTIMPLEMENTED(); }
-  void SetOmniboxSpeculation() override { NOTIMPLEMENTED(); }
-  bool WasEverUsed() const override { return true; }
-  bool UsingTCPFastOpen() const override { return false; }
-  bool WasNpnNegotiated() const override { return false; }
-  net::NextProto GetNegotiatedProtocol() const override {
-    return net::kProtoUnknown;
-  }
-  bool GetSSLInfo(net::SSLInfo* ssl_info) override {
-    NOTIMPLEMENTED();
-    return false;
-  }
-  void GetConnectionAttempts(net::ConnectionAttempts* out) const override {
-    out->clear();
-  }
-  void ClearConnectionAttempts() override {}
-  void AddConnectionAttempts(const net::ConnectionAttempts& attempts) override {
-  }
-
  private:
   MuxChannel* channel_;
 
   int base_channel_error_ = net::OK;
 
   net::CompletionCallback read_callback_;
   scoped_refptr<net::IOBuffer> read_buffer_;
   int read_buffer_size_;
 
   bool write_pending_;
   int write_result_;
   net::CompletionCallback write_callback_;
 
-  net::BoundNetLog net_log_;
-
   DISALLOW_COPY_AND_ASSIGN(MuxSocket);
 };
 
 
 ChannelMultiplexer::MuxChannel::MuxChannel(
     ChannelMultiplexer* multiplexer,
     const std::string& name,
     int send_id)
     : multiplexer_(multiplexer),
       name_(name),
       send_id_(send_id),
       id_sent_(false),
       receive_id_(kChannelIdUnknown),
       socket_(nullptr) {
 }
 
 ChannelMultiplexer::MuxChannel::~MuxChannel() {
   // Socket must be destroyed before the channel.
   DCHECK(!socket_);
   STLDeleteElements(&pending_packets_);
 }
 
-scoped_ptr<net::StreamSocket> ChannelMultiplexer::MuxChannel::CreateSocket() {
+scoped_ptr<P2PStreamSocket> ChannelMultiplexer::MuxChannel::CreateSocket() {
   DCHECK(!socket_);  // Can't create more than one socket per channel.
   scoped_ptr<MuxSocket> result(new MuxSocket(this));
   socket_ = result.get();
   return result.Pass();
 }
 
 void ChannelMultiplexer::MuxChannel::OnIncomingPacket(
     scoped_ptr<MultiplexPacket> packet,
     const base::Closure& done_task) {
   DCHECK_EQ(packet->channel_id(), receive_id_);
@@ skipping 20 matching lines @@
     scoped_ptr<MultiplexPacket> packet,
     const base::Closure& done_task) {
   packet->set_channel_id(send_id_);
   if (!id_sent_) {
     packet->set_channel_name(name_);
     id_sent_ = true;
   }
   return multiplexer_->DoWrite(packet.Pass(), done_task);
 }
 
-int ChannelMultiplexer::MuxChannel::DoRead(net::IOBuffer* buffer,
-                                           int buffer_len) {
+int ChannelMultiplexer::MuxChannel::DoRead(
+    const scoped_refptr<net::IOBuffer>& buffer,
+    int buffer_len) {
   int pos = 0;
   while (buffer_len > 0 && !pending_packets_.empty()) {
     DCHECK(!pending_packets_.front()->is_empty());
     int result = pending_packets_.front()->Read(
         buffer->data() + pos, buffer_len);
     DCHECK_LE(result, buffer_len);
     pos += result;
     buffer_len -= pos;
     if (pending_packets_.front()->is_empty()) {
       delete pending_packets_.front();
       pending_packets_.erase(pending_packets_.begin());
     }
   }
   return pos;
 }
 
 ChannelMultiplexer::MuxSocket::MuxSocket(MuxChannel* channel)
     : channel_(channel),
       read_buffer_size_(0),
       write_pending_(false),
       write_result_(0) {
 }
 
 ChannelMultiplexer::MuxSocket::~MuxSocket() {
   channel_->OnSocketDestroyed();
 }
 
 int ChannelMultiplexer::MuxSocket::Read(
-    net::IOBuffer* buffer, int buffer_len,
+    const scoped_refptr<net::IOBuffer>& buffer, int buffer_len,
     const net::CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK(read_callback_.is_null());
 
   if (base_channel_error_ != net::OK)
     return base_channel_error_;
 
   int result = channel_->DoRead(buffer, buffer_len);
   if (result == 0) {
     read_buffer_ = buffer;
     read_buffer_size_ = buffer_len;
     read_callback_ = callback;
     return net::ERR_IO_PENDING;
   }
   return result;
 }
 
 int ChannelMultiplexer::MuxSocket::Write(
-    net::IOBuffer* buffer, int buffer_len,
+    const scoped_refptr<net::IOBuffer>& buffer, int buffer_len,
     const net::CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK(write_callback_.is_null());
 
   if (base_channel_error_ != net::OK)
     return base_channel_error_;
 
   scoped_ptr<MultiplexPacket> packet(new MultiplexPacket());
   size_t size = std::min(kMaxPacketSize, buffer_len);
   packet->mutable_data()->assign(buffer->data(), size);
@@ skipping 100 matching lines @@
   for (std::list<PendingChannel>::iterator it = pending_channels_.begin();
        it != pending_channels_.end(); ++it) {
     if (it->name == name) {
       pending_channels_.erase(it);
       return;
     }
   }
 }
 
 void ChannelMultiplexer::OnBaseChannelReady(
-    scoped_ptr<net::StreamSocket> socket) {
+    scoped_ptr<P2PStreamSocket> socket) {
   base_channel_factory_ = nullptr;
   base_channel_ = socket.Pass();
 
   if (base_channel_.get()) {
     // Initialize reader and writer.
     reader_.StartReading(base_channel_.get(),
                          base::Bind(&ChannelMultiplexer::OnBaseChannelError,
                                     base::Unretained(this)));
-    writer_.Init(base_channel_.get(),
+    writer_.Init(base::Bind(&P2PStreamSocket::Write,
+                            base::Unretained(base_channel_.get())),
                  base::Bind(&ChannelMultiplexer::OnBaseChannelError,
                             base::Unretained(this)));
   }
 
   DoCreatePendingChannels();
 }
 
 void ChannelMultiplexer::DoCreatePendingChannels() {
   if (pending_channels_.empty())
     return;
 
   // Every time this function is called it connects a single channel and posts a
   // separate task to connect other channels. This is necessary because the
   // callback may destroy the multiplexer or somehow else modify
   // |pending_channels_| list (e.g. call CancelChannelCreation()).
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE, base::Bind(&ChannelMultiplexer::DoCreatePendingChannels,
                             weak_factory_.GetWeakPtr()));
 
   PendingChannel c = pending_channels_.front();
   pending_channels_.erase(pending_channels_.begin());
-  scoped_ptr<net::StreamSocket> socket;
+  scoped_ptr<P2PStreamSocket> socket;
   if (base_channel_.get())
     socket = GetOrCreateChannel(c.name)->CreateSocket();
   c.callback.Run(socket.Pass());
 }
 
 ChannelMultiplexer::MuxChannel* ChannelMultiplexer::GetOrCreateChannel(
     const std::string& name) {
   // Check if we already have a channel with the requested name.
   std::map<std::string, MuxChannel*>::iterator it = channels_.find(name);
   if (it != channels_.end())
@@ skipping 55 matching lines @@
   channel->OnIncomingPacket(packet.Pass(), done_task);
 }
 
 bool ChannelMultiplexer::DoWrite(scoped_ptr<MultiplexPacket> packet,
                                  const base::Closure& done_task) {
   return writer_.Write(SerializeAndFrameMessage(*packet), done_task);
 }
 
 }  // namespace protocol
 }  // namespace remoting