remoting/base/protocol_decoder.cc

   1 // Copyright (c) 2010 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/base/protocol_decoder.h"
   6
   7 #include "remoting/base/multiple_array_input_stream.h"
   8 #include "talk/base/byteorder.h"
   9
  10 namespace remoting {
  11
  12 ProtocolDecoder::ProtocolDecoder()
  13     : last_read_position_(0),
  14       available_bytes_(0),
  15       next_payload_(0),
  16       next_payload_known_(false) {
  17 }
  18
  19 void ProtocolDecoder::ParseClientMessages(scoped_refptr<media::DataBuffer> data,
  20                                           ClientMessageList* messages) {
  21   ParseMessages<ClientMessage>(data, messages);
  22 }
  23
  24 void ProtocolDecoder::ParseHostMessages(scoped_refptr<media::DataBuffer> data,
  25                                         HostMessageList* messages) {
  26   ParseMessages<HostMessage>(data, messages);
  27 }
  28
  29 template <typename T>
  30 void ProtocolDecoder::ParseMessages(scoped_refptr<media::DataBuffer> data,
  31                                     std::vector<T*>* messages) {
  32   // If this is the first data in the processing queue, then set the
  33   // last read position to 0.
  34   if (data_list_.empty())
  35     last_read_position_ = 0;
  36
  37   // First enqueue the data received.
  38   data_list_.push_back(data);
  39   available_bytes_ += data->GetDataSize();
  40
  41   // Then try to parse one message until we can't parse anymore.
  42   T* message;
  43   while (ParseOneMessage<T>(&message)) {
  44     messages->push_back(message);
  45   }
  46 }
  47
  48 template <typename T>
  49 bool ProtocolDecoder::ParseOneMessage(T** message) {
  50   // Determine the payload size. If we already know it, then skip this
  51   // part.
  52   // We have the value set to -1 for checking later.
  53   int next_payload = -1;
  54   if (!next_payload_known_ && GetPayloadSize(&next_payload)) {
  55     DCHECK_NE(-1, next_payload);
  56     next_payload_ = next_payload;
  57     next_payload_known_ = true;
  58   }
  59
  60   // If the next payload size is still not known or we don't have enough
  61   // data for parsing then exit.
  62   if (!next_payload_known_ || available_bytes_ < next_payload_)
  63     return false;
  64   next_payload_known_ = false;
  65
  66   // Extract data from |data_list_| used to form a full protocol buffer.
  67   DataList buffers;
  68   std::deque<const uint8*> buffer_pointers;
  69   std::deque<int> buffer_sizes;
  70   while (next_payload_ > 0 && !data_list_.empty()) {
  71     scoped_refptr<media::DataBuffer> buffer = data_list_.front();
  72     size_t read_bytes = std::min(buffer->GetDataSize() - last_read_position_,
  73                                  next_payload_);
  74
  75     buffers.push_back(buffer);
  76     buffer_pointers.push_back(buffer->GetData() + last_read_position_);
  77     buffer_sizes.push_back(read_bytes);
  78
  79     // Adjust counters.
  80     last_read_position_ += read_bytes;
  81     next_payload_ -= read_bytes;
  82     available_bytes_ -= read_bytes;
  83
  84     // If the front buffer is fully read, remove it from the queue.
  85     if (buffer->GetDataSize() == last_read_position_) {
  86       data_list_.pop_front();
  87       last_read_position_ = 0;
  88     }
  89   }
  90   DCHECK_EQ(0UL, next_payload_);
  91   DCHECK_EQ(buffers.size(), buffer_pointers.size());
  92   DCHECK_EQ(buffers.size(), buffer_sizes.size());
  93
  94   // Create a MultipleArrayInputStream for parsing.
  95   MultipleArrayInputStream stream(buffers.size());
  96   for (size_t i = 0; i < buffers.size(); ++i) {
  97     stream.SetBuffer(i, buffer_pointers[i], buffer_sizes[i]);
  98   }
  99
 100   // And finally it is parsing.
 101   *message = new T();
 102   bool ret = (*message)->ParseFromZeroCopyStream(&stream);
 103   if (!ret)
 104     delete *message;
 105   return ret;
 106 }
 107
 108 bool ProtocolDecoder::GetPayloadSize(int* size) {
 109   // The header has a size of 4 bytes.
 110   const size_t kHeaderSize = sizeof(int32);
 111
 112   if (available_bytes_ < kHeaderSize)
 113     return false;
 114
 115   std::string header;
 116   while (header.length() < kHeaderSize && !data_list_.empty()) {
 117     scoped_refptr<media::DataBuffer> buffer = data_list_.front();
 118
 119     // Find out how many bytes we need and how many bytes are available in this
 120     // buffer.
 121     int needed_bytes = kHeaderSize - header.length();
 122     int available_bytes = buffer->GetDataSize() - last_read_position_;
 123
 124     // Then append the required bytes into the header and advance the last
 125     // read position.
 126     int read_bytes = std::min(needed_bytes, available_bytes);
 127     header.append(
 128         reinterpret_cast<const char*>(buffer->GetData()) + last_read_position_,
 129         read_bytes);
 130     last_read_position_ += read_bytes;
 131     available_bytes_ -= read_bytes;
 132
 133     // If the buffer is depleted then remove it from the queue.
 134     if (last_read_position_ == buffer->GetDataSize()) {
 135       last_read_position_ = 0;
 136       data_list_.pop_front();
 137     }
 138   }
 139
 140   if (header.length() == kHeaderSize) {
 141     *size = talk_base::GetBE32(header.c_str());
 142     return true;
 143   }
 144   NOTREACHED() << "Unable to extract payload size";
 145   return false;
 146 }
 147
 148 }  // namespace remoting