Add P2PDatagramSocket and P2PStreamSocket interfaces.

remoting/base/buffered_socket_writer.cc

 // Copyright 2015 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/base/buffered_socket_writer.h"
 
 #include "base/bind.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/io_buffer.h"
 #include "net/base/net_errors.h"
+#include "net/socket/socket.h"
 
 namespace remoting {
 
-struct BufferedSocketWriterBase::PendingPacket {
-  PendingPacket(scoped_refptr<net::IOBufferWithSize> data,
+namespace {
+
+int WriteNetSocket(net::Socket* socket,
+                   const scoped_refptr<net::IOBuffer>& buf,
+                   int buf_len,
+                   const net::CompletionCallback& callback) {
+  return socket->Write(buf.get(), buf_len, callback);
+}
+
+}  // namespace
+
+struct BufferedSocketWriter::PendingPacket {
+  PendingPacket(scoped_refptr<net::DrainableIOBuffer> data,
                 const base::Closure& done_task)
       : data(data),
         done_task(done_task) {
   }
 
-  scoped_refptr<net::IOBufferWithSize> data;
+  scoped_refptr<net::DrainableIOBuffer> data;
   base::Closure done_task;
 };
 
-BufferedSocketWriterBase::BufferedSocketWriterBase()
-    : socket_(nullptr),
-      write_pending_(false),
-      closed_(false),
-      destroyed_flag_(nullptr) {
+// static
+scoped_ptr<BufferedSocketWriter> BufferedSocketWriter::CreateForSocket(
+    net::Socket* socket, const WriteFailedCallback& write_failed_callback) {
+  scoped_ptr<BufferedSocketWriter> result(new BufferedSocketWriter());
+  result->Init(base::Bind(&WriteNetSocket, socket), write_failed_callback);
+  return result.Pass();
 }
 
-void BufferedSocketWriterBase::Init(net::Socket* socket,
-                                    const WriteFailedCallback& callback) {
-  DCHECK(CalledOnValidThread());
-  DCHECK(socket);
-  socket_ = socket;
-  write_failed_callback_ = callback;
+BufferedSocketWriter::BufferedSocketWriter() {}
+
+BufferedSocketWriter::~BufferedSocketWriter() {
+  if (destroyed_flag_)
+    *destroyed_flag_ = true;
+
+  STLDeleteElements(&queue_);
 }
 
-bool BufferedSocketWriterBase::Write(
-    scoped_refptr<net::IOBufferWithSize> data, const base::Closure& done_task) {
-  DCHECK(CalledOnValidThread());
-  DCHECK(socket_);
+void BufferedSocketWriter::Init(
+    const WriteCallback& write_callback,
+    const WriteFailedCallback& write_failed_callback) {
+  write_callback_ = write_callback;
+  write_failed_callback_ = write_failed_callback;
+}
+
+bool BufferedSocketWriter::Write(
+    const scoped_refptr<net::IOBufferWithSize>& data,
+    const base::Closure& done_task) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!write_callback_.is_null());
   DCHECK(data.get());
 
-  // Don't write after Close().
+  // Don't write after error.
   if (closed_)
     return false;
 
-  queue_.push_back(new PendingPacket(data, done_task));
+  queue_.push_back(new PendingPacket(
+      new net::DrainableIOBuffer(data.get(), data->size()), done_task));
 
   DoWrite();
 
-  // DoWrite() may trigger OnWriteError() to be called.
   return !closed_;
 }
 
-void BufferedSocketWriterBase::DoWrite() {
-  DCHECK(CalledOnValidThread());
-  DCHECK(socket_);
+void BufferedSocketWriter::DoWrite() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!write_callback_.is_null());
 
   // Don't try to write if there is another write pending.
   if (write_pending_)
     return;
 
-  // Don't write after Close().
+  // Don't write after error.
   if (closed_)
     return;
 
   while (true) {
-    net::IOBuffer* current_packet;
-    int current_packet_size;
-    GetNextPacket(&current_packet, &current_packet_size);
+    if (queue_.empty())
+      break;
 
-    // Return if the queue is empty.
-    if (!current_packet)
-      return;
-
-    int result = socket_->Write(
-        current_packet, current_packet_size,
-        base::Bind(&BufferedSocketWriterBase::OnWritten,
+    int result = write_callback_.Run(
+        queue_.front()->data.get(), queue_.front()->data->size(),
+        base::Bind(&BufferedSocketWriter::OnWritten,
                    base::Unretained(this)));

[Wez, 2015/07/14 18:28:09]

You're using base::Unretained() here, but this class doesn't own the supplied
|socket|, so there's no explicit guarantee that a queued Write() will be dropped
when necessary.

[Sergey Ulanov, 2015/07/14 21:07:54]

Good point. Fixed by using weak pointer here.

     bool write_again = false;
     HandleWriteResult(result, &write_again);
     if (!write_again)
       return;
   }
 }
 
-void BufferedSocketWriterBase::HandleWriteResult(int result,
-                                                 bool* write_again) {
+void BufferedSocketWriter::HandleWriteResult(int result, bool* write_again) {
   *write_again = false;
   if (result < 0) {
     if (result == net::ERR_IO_PENDING) {
       write_pending_ = true;
     } else {
-      HandleError(result);
+      closed_ = true;

[Wez, 2015/07/14 18:28:09]

nit: Could you instead implement a private getter is_closed() that checks
|write_callback_|, to ensure consistent internal state?

[Sergey Ulanov, 2015/07/14 21:07:54]

Done.

+      STLDeleteElements(&queue_);

[Wez, 2015/07/14 18:28:09]

Why do we need to delete the elements here? Surely we're not going to write any
more anyway?

[Sergey Ulanov, 2015/07/14 21:07:54]

Done.

       if (!write_failed_callback_.is_null())
         write_failed_callback_.Run(result);
     }
     return;
   }
 
-  base::Closure done_task = AdvanceBufferPosition(result);
-  if (!done_task.is_null()) {
+  DCHECK(!queue_.empty());
+
+  queue_.front()->data->DidConsume(result);
+
+  if (queue_.front()->data->BytesRemaining() == 0) {
+    base::Closure done_task = queue_.front()->done_task;
+    delete queue_.front();
+    queue_.pop_front();
     bool destroyed = false;
     destroyed_flag_ = &destroyed;

[Wez, 2015/07/14 18:28:09]

I realise that this was in the original implementation, but it seems super ugly
- do we really need the caller to be able to delete the buffered writer in
response to the write-completion callback? Do we really need a write-completion
callback per-buffer, even, or could we batch it?

If so, given that write completion is already asynchronous, could we defer the
callback, or the continuation of writes, via PostTask?

If that sounds evil, then consider turning DoWrite/HandleWriteResult into a
tail-recursive cycle, and using a WeakPtr to do the HandleWriteResult->DoWrite
call; that is equivalent to the pointer-to-bool but more familiar to folks
reading this code.

[Sergey Ulanov, 2015/07/14 21:07:54]

I'm not sure we need it right now, but I think we needed it in the past and in
general I prefer when callback allow to delete the calling object. E.g. it
allows to create a writer to write a single message and delete it right after
that.
Not sure what you mean by batching. 
Yes. Changed this code to call done_task and post async task for DoWrite().

-    done_task.Run();
+    if (!done_task.is_null())
+      done_task.Run();
     if (destroyed) {
       // Stop doing anything if we've been destroyed by the callback.
       return;
     }
     destroyed_flag_ = nullptr;
   }
 
   *write_again = true;
 }
 
-void BufferedSocketWriterBase::OnWritten(int result) {
-  DCHECK(CalledOnValidThread());
+void BufferedSocketWriter::OnWritten(int result) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(write_pending_);
   write_pending_ = false;
 
   bool write_again;
   HandleWriteResult(result, &write_again);
   if (write_again)
     DoWrite();
 }
 
-void BufferedSocketWriterBase::HandleError(int result) {
-  DCHECK(CalledOnValidThread());
-
-  closed_ = true;
-
-  STLDeleteElements(&queue_);
-
-  // Notify subclass that an error is received.
-  OnError(result);
-}
-
-void BufferedSocketWriterBase::Close() {
-  DCHECK(CalledOnValidThread());
-  closed_ = true;
-}
-
-BufferedSocketWriterBase::~BufferedSocketWriterBase() {
-  if (destroyed_flag_)
-    *destroyed_flag_ = true;
-
-  STLDeleteElements(&queue_);
-}
-
-base::Closure BufferedSocketWriterBase::PopQueue() {
-  base::Closure result = queue_.front()->done_task;
-  delete queue_.front();
-  queue_.pop_front();
-  return result;
-}
-
-BufferedSocketWriter::BufferedSocketWriter() {
-}
-
-void BufferedSocketWriter::GetNextPacket(
-    net::IOBuffer** buffer, int* size) {
-  if (!current_buf_.get()) {
-    if (queue_.empty()) {
-      *buffer = nullptr;
-      return;  // Nothing to write.
-    }
-    current_buf_ = new net::DrainableIOBuffer(queue_.front()->data.get(),
-                                              queue_.front()->data->size());
-  }
-
-  *buffer = current_buf_.get();
-  *size = current_buf_->BytesRemaining();
-}
-
-base::Closure BufferedSocketWriter::AdvanceBufferPosition(int written) {
-  current_buf_->DidConsume(written);
-
-  if (current_buf_->BytesRemaining() == 0) {
-    current_buf_ = nullptr;
-    return PopQueue();
-  }
-  return base::Closure();
-}
-
-void BufferedSocketWriter::OnError(int result) {
-  current_buf_ = nullptr;
-}
-
-BufferedSocketWriter::~BufferedSocketWriter() {
-}
-
-BufferedDatagramWriter::BufferedDatagramWriter() {
-}
-
-void BufferedDatagramWriter::GetNextPacket(
-    net::IOBuffer** buffer, int* size) {
-  if (queue_.empty()) {
-    *buffer = nullptr;
-    return;  // Nothing to write.
-  }
-  *buffer = queue_.front()->data.get();
-  *size = queue_.front()->data->size();
-}
-
-base::Closure BufferedDatagramWriter::AdvanceBufferPosition(int written) {
-  DCHECK_EQ(written, queue_.front()->data->size());
-  return PopQueue();
-}
-
-void BufferedDatagramWriter::OnError(int result) {
-  // Nothing to do here.
-}
-
-BufferedDatagramWriter::~BufferedDatagramWriter() {
-}
-
 }  // namespace remoting