remoting/protocol/quic_channel.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/protocol/quic_channel.h"

#include "base/callback_helpers.h"
#include "net/base/io_buffer.h"
#include "net/base/net_errors.h"

namespace remoting {
namespace protocol {

static const size_t kNamePrefixLength = 1;
static const size_t kMaxNameLength  = 255;

QuicChannel::QuicChannel(net::QuicP2PStream* stream,
                         const base::Closure& on_destroyed_callback)
    : stream_(stream), on_destroyed_callback_(on_destroyed_callback) {
  DCHECK(stream_);
  stream_->SetDelegate(this);
}

QuicChannel::~QuicChannel() {
  // Don't call the read callback when destroying the stream.
  read_callback_.Reset();

  on_destroyed_callback_.Run();

  // The callback must destroy the stream which must result in OnClose().
  DCHECK(!stream_);
}

int QuicChannel::Read(const scoped_refptr<net::IOBuffer>& buffer,
                      int buffer_len,
                      const net::CompletionCallback& callback) {
  DCHECK(read_callback_.is_null());

  if (error_ != net::OK)
    return error_;

  if (data_received_.total_bytes() == 0) {
    read_buffer_ = buffer;
    read_buffer_size_ = buffer_len;
    read_callback_ = callback;
    return net::ERR_IO_PENDING;
  }

  int result = std::min(buffer_len, data_received_.total_bytes());
  data_received_.CopyTo(buffer->data(), result);
  data_received_.CropFront(result);
  return result;
}

int QuicChannel::Write(const scoped_refptr<net::IOBuffer>& buffer,
                       int buffer_len,
                       const net::CompletionCallback& callback) {
  if (error_ != net::OK)
    return error_;

  return stream_->Write(base::StringPiece(buffer->data(), buffer_len),
                        callback);
}

void QuicChannel::SetName(const std::string& name) {
  DCHECK(name_.empty());

  name_ = name;
}

void QuicChannel::OnDataReceived(const char* data, int length) {
  if (read_callback_.is_null()) {
    data_received_.AppendCopyOf(data, length);
    return;
  }

  DCHECK_EQ(data_received_.total_bytes(), 0);
  int bytes_to_read = std::min(length, read_buffer_size_);
  memcpy(read_buffer_->data(), data, bytes_to_read);
  read_buffer_ = nullptr;

  // Copy leftover data to |data_received_|.
  if (length > bytes_to_read)
    data_received_.AppendCopyOf(data + bytes_to_read, length - bytes_to_read);

  base::ResetAndReturn(&read_callback_).Run(bytes_to_read);
}

void QuicChannel::OnClose(net::QuicErrorCode error) {
  error_ = (error == net::QUIC_NO_ERROR) ? net::ERR_CONNECTION_CLOSED
                                         : net::ERR_QUIC_PROTOCOL_ERROR;
  stream_ = nullptr;
  if (!read_callback_.is_null()) {
    base::ResetAndReturn(&read_callback_).Run(error_);
  }
}

QuicClientChannel::QuicClientChannel(net::QuicP2PStream* stream,
                                     const base::Closure& on_destroyed_callback,
                                     const std::string& name)
    : QuicChannel(stream, on_destroyed_callback) {
  CHECK_LE(name.size(), kMaxNameLength);

  SetName(name);

  // Send the name to the host.
  stream_->WriteHeader(
      std::string(kNamePrefixLength, static_cast<char>(name.size())) + name);
}

QuicClientChannel::~QuicClientChannel() {}

QuicServerChannel::QuicServerChannel(
    net::QuicP2PStream* stream,
    const base::Closure& on_destroyed_callback)
    : QuicChannel(stream, on_destroyed_callback) {}

void QuicServerChannel::ReceiveName(
    const base::Closure& name_received_callback) {
  name_received_callback_ = name_received_callback;

  // First read 1 byte containing name length.
  name_read_buffer_ = new net::DrainableIOBuffer(
      new net::IOBuffer(kNamePrefixLength), kNamePrefixLength);
  int result = Read(name_read_buffer_, kNamePrefixLength,
                    base::Bind(&QuicServerChannel::OnNameSizeReadResult,
                               base::Unretained(this)));
  if (result != net::ERR_IO_PENDING)
    OnNameSizeReadResult(result);
}

QuicServerChannel::~QuicServerChannel() {}

void QuicServerChannel::OnNameSizeReadResult(int result) {
  if (result < 0) {
    base::ResetAndReturn(&name_received_callback_).Run();
    return;
  }

  DCHECK_EQ(result, static_cast<int>(kNamePrefixLength));
  name_length_ = *reinterpret_cast<uint8_t*>(name_read_buffer_->data());
  name_read_buffer_ =
      new net::DrainableIOBuffer(new net::IOBuffer(name_length_), name_length_);
  ReadNameLoop(0);
}

void QuicServerChannel::ReadNameLoop(int result) {
  while (result >= 0 && name_read_buffer_->BytesRemaining() > 0) {
    result = Read(name_read_buffer_, name_read_buffer_->BytesRemaining(),
                  base::Bind(&QuicServerChannel::OnNameReadResult,
                             base::Unretained(this)));
    if (result >= 0) {
      name_read_buffer_->DidConsume(result);
    }
  }

  if (result < 0 && result != net::ERR_IO_PENDING) {
    // Failed to read name for the stream.
    base::ResetAndReturn(&name_received_callback_).Run();
    return;
  }

  if (name_read_buffer_->BytesRemaining() == 0) {
    name_read_buffer_->SetOffset(0);
    SetName(std::string(name_read_buffer_->data(),
                        name_read_buffer_->data() + name_length_));
    base::ResetAndReturn(&name_received_callback_).Run();
  }
}

void QuicServerChannel::OnNameReadResult(int result) {
  if (result > 0)
    name_read_buffer_->DidConsume(result);

  ReadNameLoop(result);
}

}  // namespace protocol
}  // namespace remoting