remoting/host/native_messaging/native_messaging_reader.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "remoting/host/native_messaging/native_messaging_reader.h"
   6
   7 #include <string>
   8
   9 #include "base/bind.h"
  10 #include "base/files/file.h"
  11 #include "base/json/json_reader.h"
  12 #include "base/location.h"
  13 #include "base/sequenced_task_runner.h"
  14 #include "base/single_thread_task_runner.h"
  15 #include "base/stl_util.h"
  16 #include "base/thread_task_runner_handle.h"
  17 #include "base/threading/sequenced_worker_pool.h"
  18 #include "base/values.h"
  19
  20 namespace {
  21
  22 // uint32 is specified in the protocol as the type for the message header.
  23 typedef uint32 MessageLengthType;
  24
  25 const int kMessageHeaderSize = sizeof(MessageLengthType);
  26
  27 // Limit the size of received messages, to avoid excessive memory-allocation in
  28 // this process, and potential overflow issues when casting to a signed 32-bit
  29 // int.
  30 const MessageLengthType kMaximumMessageSize = 1024 * 1024;
  31
  32 }  // namespace
  33
  34 namespace remoting {
  35
  36 class NativeMessagingReader::Core {
  37  public:
  38   Core(base::File file,
  39        scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner,
  40        scoped_refptr<base::SequencedTaskRunner> read_task_runner,
  41        base::WeakPtr<NativeMessagingReader> reader_);
  42   ~Core();
  43
  44   // Reads a message from the Native Messaging client and passes it to
  45   // |message_callback_| on the originating thread. Called on the reader thread.
  46   void ReadMessage();
  47
  48  private:
  49   // Notify the reader's EOF callback when an error occurs or EOF is reached.
  50   void NotifyEof();
  51
  52   base::File read_stream_;
  53
  54   base::WeakPtr<NativeMessagingReader> reader_;
  55
  56   // Used to post the caller-supplied reader callbacks on the caller thread.
  57   scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner_;
  58
  59   // Used to DCHECK that the reader code executes on the correct thread.
  60   scoped_refptr<base::SequencedTaskRunner> read_task_runner_;
  61
  62   DISALLOW_COPY_AND_ASSIGN(Core);
  63 };
  64
  65 NativeMessagingReader::Core::Core(
  66     base::File file,
  67     scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner,
  68     scoped_refptr<base::SequencedTaskRunner> read_task_runner,
  69     base::WeakPtr<NativeMessagingReader> reader)
  70     : read_stream_(file.Pass()),
  71       reader_(reader),
  72       caller_task_runner_(caller_task_runner),
  73       read_task_runner_(read_task_runner) {
  74 }
  75
  76 NativeMessagingReader::Core::~Core() {}
  77
  78 void NativeMessagingReader::Core::ReadMessage() {
  79   DCHECK(read_task_runner_->RunsTasksOnCurrentThread());
  80
  81   // Keep reading messages until the stream is closed or an error occurs.
  82   while (true) {
  83     MessageLengthType message_length;
  84     int read_result = read_stream_.ReadAtCurrentPos(
  85         reinterpret_cast<char*>(&message_length), kMessageHeaderSize);
  86     if (read_result != kMessageHeaderSize) {
  87       // 0 means EOF which is normal and should not be logged as an error.
  88       if (read_result != 0) {
  89         LOG(ERROR) << "Failed to read message header, read returned "
  90                    << read_result;
  91       }
  92       NotifyEof();
  93       return;
  94     }
  95
  96     if (message_length > kMaximumMessageSize) {
  97       LOG(ERROR) << "Message size too large: " << message_length;
  98       NotifyEof();
  99       return;
 100     }
 101
 102     std::string message_json(message_length, '\0');
 103     read_result = read_stream_.ReadAtCurrentPos(string_as_array(&message_json),
 104                                                 message_length);
 105     if (read_result != static_cast<int>(message_length)) {
 106       LOG(ERROR) << "Failed to read message body, read returned "
 107                  << read_result;
 108       NotifyEof();
 109       return;
 110     }
 111
 112     scoped_ptr<base::Value> message = base::JSONReader::Read(message_json);
 113     if (!message) {
 114       LOG(ERROR) << "Failed to parse JSON message: " << message;
 115       NotifyEof();
 116       return;
 117     }
 118
 119     // Notify callback of new message.
 120     caller_task_runner_->PostTask(
 121         FROM_HERE, base::Bind(&NativeMessagingReader::InvokeMessageCallback,
 122                               reader_, base::Passed(&message)));
 123   }
 124 }
 125
 126 void NativeMessagingReader::Core::NotifyEof() {
 127   DCHECK(read_task_runner_->RunsTasksOnCurrentThread());
 128   caller_task_runner_->PostTask(
 129       FROM_HERE,
 130       base::Bind(&NativeMessagingReader::InvokeEofCallback, reader_));
 131 }
 132
 133 NativeMessagingReader::NativeMessagingReader(base::File file)
 134     : reader_thread_("Reader"),
 135       weak_factory_(this) {
 136   reader_thread_.Start();
 137   read_task_runner_ = reader_thread_.task_runner();
 138   core_.reset(new Core(file.Pass(), base::ThreadTaskRunnerHandle::Get(),
 139                        read_task_runner_, weak_factory_.GetWeakPtr()));
 140 }
 141
 142 NativeMessagingReader::~NativeMessagingReader() {
 143   read_task_runner_->DeleteSoon(FROM_HERE, core_.release());
 144 }
 145
 146 void NativeMessagingReader::Start(MessageCallback message_callback,
 147                                   base::Closure eof_callback) {
 148   message_callback_ = message_callback;
 149   eof_callback_ = eof_callback;
 150
 151   // base::Unretained is safe since |core_| is only deleted via the
 152   // DeleteSoon task which is posted from this class's dtor.
 153   read_task_runner_->PostTask(
 154       FROM_HERE, base::Bind(&NativeMessagingReader::Core::ReadMessage,
 155                             base::Unretained(core_.get())));
 156 }
 157
 158 void NativeMessagingReader::InvokeMessageCallback(
 159     scoped_ptr<base::Value> message) {
 160   message_callback_.Run(message.Pass());
 161 }
 162
 163 void NativeMessagingReader::InvokeEofCallback() {
 164   eof_callback_.Run();
 165 }
 166
 167 }  // namespace remoting