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.
+
+  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