Fix MessageReader to pass errors to the channel

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"
 
 namespace remoting {
 namespace protocol {
@@ skipping 51 matching lines @@
   ~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();
   void OnIncomingPacket(scoped_ptr<MultiplexPacket> packet,
                         const base::Closure& done_task);
-  void OnWriteFailed();
+  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);
 
  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,
                                       public base::NonThreadSafe,
                                       public base::SupportsWeakPtr<MuxSocket> {
  public:
   MuxSocket(MuxChannel* channel);
   ~MuxSocket() override;
 
   void OnWriteComplete();
-  void OnWriteFailed();
+  void OnBaseChannelError(int error);
   void OnPacketReceived();
 
   // net::StreamSocket interface.
   int Read(net::IOBuffer* buffer,
            int buffer_len,
            const net::CompletionCallback& callback) override;
   int Write(net::IOBuffer* buffer,
             int buffer_len,
             const net::CompletionCallback& callback) override;
 
@@ skipping 40 matching lines @@
     return net::kProtoUnknown;
   }
   bool GetSSLInfo(net::SSLInfo* ssl_info) override {
     NOTIMPLEMENTED();
     return false;
   }
 
  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_;
 
@@ skipping 32 matching lines @@
   DCHECK_EQ(packet->channel_id(), receive_id_);
   if (packet->data().size() > 0) {
     pending_packets_.push_back(new PendingPacket(packet.Pass(), done_task));
     if (socket_) {
       // Notify the socket that we have more data.
       socket_->OnPacketReceived();
     }
   }
 }
 
-void ChannelMultiplexer::MuxChannel::OnWriteFailed() {
+void ChannelMultiplexer::MuxChannel::OnBaseChannelError(int error) {
   if (socket_)
-    socket_->OnWriteFailed();
+    socket_->OnBaseChannelError(error);
 }
 
 void ChannelMultiplexer::MuxChannel::OnSocketDestroyed() {
   DCHECK(socket_);
   socket_ = nullptr;
 }
 
 bool ChannelMultiplexer::MuxChannel::DoWrite(
     scoped_ptr<MultiplexPacket> packet,
     const base::Closure& done_task) {
@@ skipping 33 matching lines @@
 ChannelMultiplexer::MuxSocket::~MuxSocket() {
   channel_->OnSocketDestroyed();
 }
 
 int ChannelMultiplexer::MuxSocket::Read(
     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 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);
 
   write_pending_ = true;
   bool result = channel_->DoWrite(packet.Pass(), base::Bind(
       &ChannelMultiplexer::MuxSocket::OnWriteComplete, AsWeakPtr()));
 
   if (!result) {
     // Cannot complete the write, e.g. if the connection has been terminated.
     return net::ERR_FAILED;
   }
 
   // OnWriteComplete() might be called above synchronously.
   if (write_pending_) {
     DCHECK(write_callback_.is_null());
     write_callback_ = callback;
     write_result_ = size;
     return net::ERR_IO_PENDING;
   }
 
   return size;
 }
 
 void ChannelMultiplexer::MuxSocket::OnWriteComplete() {
   write_pending_ = false;
-  if (!write_callback_.is_null()) {
-    net::CompletionCallback cb;
-    std::swap(cb, write_callback_);
-    cb.Run(write_result_);
-  }
+  if (!write_callback_.is_null())
+    base::ResetAndReturn(&write_callback_).Run(write_result_);
+
 }
 
-void ChannelMultiplexer::MuxSocket::OnWriteFailed() {
-  if (!write_callback_.is_null()) {
-    net::CompletionCallback cb;
-    std::swap(cb, write_callback_);
-    cb.Run(net::ERR_FAILED);
+void ChannelMultiplexer::MuxSocket::OnBaseChannelError(int error) {
+  base_channel_error_ = error;
+
+  // Here only one of the read and write callbacks is called if both of them are
+  // pending. Ideally both of them should be called in that case, but that would
+  // require the second one to be called asynchronously which would complicate
+  // this code. Channels handle read and write errors the same way (see
+  // ChannelDispatcherBase::OnReadWriteFailed) so calling only one of the
+  // callbacks is enough.

[Jamie, 2015/05/13 18:25:17]

But you're only resetting one of the callbacks in this case. Is that what you
want? It seems that if the right thing to do would be to invoke (and reset) both
callbacks, then the next-best thing (given that they are expected to do the same
thing) would be to invoke only one but reset them both.

[Sergey Ulanov, 2015/05/14 00:15:21]

It's better to leave write_callback_ not null in that case because Write() call
is still pending.

+
+  if (!read_callback_.is_null()) {
+    base::ResetAndReturn(&read_callback_).Run(error);
+    return;
   }
+
+  if (!write_callback_.is_null())
+    base::ResetAndReturn(&write_callback_).Run(error);
 }
 
 void ChannelMultiplexer::MuxSocket::OnPacketReceived() {
   if (!read_callback_.is_null()) {
     int result = channel_->DoRead(read_buffer_.get(), read_buffer_size_);
     read_buffer_ = nullptr;
     DCHECK_GT(result, 0);
-    net::CompletionCallback cb;
-    std::swap(cb, read_callback_);
-    cb.Run(result);
+    base::ResetAndReturn(&read_callback_).Run(result);
   }
 }
 
 ChannelMultiplexer::ChannelMultiplexer(StreamChannelFactory* factory,
                                        const std::string& base_channel_name)
     : base_channel_factory_(factory),
       base_channel_name_(base_channel_name),
       next_channel_id_(0),
       parser_(base::Bind(&ChannelMultiplexer::OnIncomingPacket,
                          base::Unretained(this)),
@@ skipping 43 matching lines @@
   }
 }
 
 void ChannelMultiplexer::OnBaseChannelReady(
     scoped_ptr<net::StreamSocket> socket) {
   base_channel_factory_ = nullptr;
   base_channel_ = socket.Pass();
 
   if (base_channel_.get()) {
     // Initialize reader and writer.
-    reader_.StartReading(base_channel_.get());
+    reader_.StartReading(base_channel_.get(),
+                         base::Bind(&ChannelMultiplexer::OnBaseChannelError,
+                                    base::Unretained(this)));
     writer_.Init(base_channel_.get(),
-                 base::Bind(&ChannelMultiplexer::OnWriteFailed,
+                 base::Bind(&ChannelMultiplexer::OnBaseChannelError,
                             base::Unretained(this)));
   }
 
   DoCreatePendingChannels();
 }
 
 void ChannelMultiplexer::DoCreatePendingChannels() {
   if (pending_channels_.empty())
     return;
 
@@ skipping 21 matching lines @@
     return it->second;
 
   // Create a new channel if we haven't found existing one.
   MuxChannel* channel = new MuxChannel(this, name, next_channel_id_);
   ++next_channel_id_;
   channels_[channel->name()] = channel;
   return channel;
 }
 
 
-void ChannelMultiplexer::OnWriteFailed(int error) {
+void ChannelMultiplexer::OnBaseChannelError(int error) {
   for (std::map<std::string, MuxChannel*>::iterator it = channels_.begin();
        it != channels_.end(); ++it) {
     base::ThreadTaskRunnerHandle::Get()->PostTask(
-        FROM_HERE, base::Bind(&ChannelMultiplexer::NotifyWriteFailed,
-                              weak_factory_.GetWeakPtr(), it->second->name()));
+        FROM_HERE,
+        base::Bind(&ChannelMultiplexer::NotifyBaseChannelError,
+                   weak_factory_.GetWeakPtr(), it->second->name(), error));
   }
 }
 
-void ChannelMultiplexer::NotifyWriteFailed(const std::string& name) {
+void ChannelMultiplexer::NotifyBaseChannelError(const std::string& name,
+                                                int error) {
   std::map<std::string, MuxChannel*>::iterator it = channels_.find(name);
-  if (it != channels_.end()) {
-    it->second->OnWriteFailed();
-  }
+  if (it != channels_.end())
+    it->second->OnBaseChannelError(error);
 }
 
 void ChannelMultiplexer::OnIncomingPacket(scoped_ptr<MultiplexPacket> packet,
                                           const base::Closure& done_task) {
   DCHECK(packet->has_channel_id());
   if (!packet->has_channel_id()) {
     LOG(ERROR) << "Received packet without channel_id.";
     done_task.Run();
     return;
   }
@@ skipping 20 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