media/cast/test/transport/transport.cc

// Copyright 2013 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 "media/cast/test/transport/transport.h"

#include <string>

#include "base/bind.h"
#include "base/logging.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_ptr.h"
#include "base/message_loop/message_loop.h"
#include "base/rand_util.h"
#include "net/base/io_buffer.h"
#include "net/base/rand_callback.h"
#include "net/base/test_completion_callback.h"

namespace media {
namespace cast {
namespace test {

const int kMaxPacketSize = 1500;

class LocalUdpTransportData;

void CreateUDPAddress(std::string ip_str, int port, net::IPEndPoint* address) {
  net::IPAddressNumber ip_number;
  bool rv = net::ParseIPLiteralToNumber(ip_str, &ip_number);
  if (!rv)
    return;
  *address = net::IPEndPoint(ip_number, port);
}

class LocalUdpTransportData
    : public base::RefCountedThreadSafe<LocalUdpTransportData> {
 public:
  LocalUdpTransportData(net::UDPServerSocket* udp_socket,
                        scoped_refptr<base::TaskRunner> io_thread_proxy)
    : udp_socket_(udp_socket),
      buffer_(new net::IOBufferWithSize(kMaxPacketSize)),
      io_thread_proxy_(io_thread_proxy) {
  }

  void ListenTo(net::IPEndPoint bind_address) {
    DCHECK(io_thread_proxy_->RunsTasksOnCurrentThread());

    bind_address_ = bind_address;
    io_thread_proxy_->PostTask(FROM_HERE,
        base::Bind(&LocalUdpTransportData::RecvFromSocketLoop, this));
  }

  void DeletePacket(uint8* data) {
    // Should be called from the receiver (not on the transport thread).
    DCHECK(!(io_thread_proxy_->RunsTasksOnCurrentThread()));
    delete [] data;
  }

  void PacketReceived(int size) {
    DCHECK(io_thread_proxy_->RunsTasksOnCurrentThread());
    // Got a packet with length result.
    uint8* data = new uint8[size];
    memcpy(data, buffer_->data(), size);
    packet_receiver_->ReceivedPacket(data, size,
        base::Bind(&LocalUdpTransportData::DeletePacket, this, data));
    RecvFromSocketLoop();

  }

  void RecvFromSocketLoop() {
    DCHECK(io_thread_proxy_->RunsTasksOnCurrentThread());
    // Callback should always trigger with a packet.
    int res = udp_socket_->RecvFrom(buffer_.get(), kMaxPacketSize,
        &bind_address_, base::Bind(&LocalUdpTransportData::PacketReceived,
        this));
    DCHECK(res >= net::ERR_IO_PENDING);
    if (res > 0) {
      PacketReceived(res);
    }
  }

  void set_packet_receiver(PacketReceiver* packet_receiver) {
    packet_receiver_ = packet_receiver;
  }

  void Close() {
    udp_socket_->Close();
  }

 protected:
  virtual ~LocalUdpTransportData() {}

 private:
  friend class base::RefCountedThreadSafe<LocalUdpTransportData>;

  net::UDPServerSocket* udp_socket_;
  net::IPEndPoint bind_address_;
  PacketReceiver* packet_receiver_;
  scoped_refptr<net::IOBufferWithSize> buffer_;
  scoped_refptr<base::TaskRunner> io_thread_proxy_;

  DISALLOW_COPY_AND_ASSIGN(LocalUdpTransportData);
};

class LocalPacketSender : public PacketSender,
                          public base::RefCountedThreadSafe<LocalPacketSender> {
 public:
  LocalPacketSender(net::UDPServerSocket* udp_socket,
                    scoped_refptr<base::TaskRunner> io_thread_proxy)
      : udp_socket_(udp_socket),
        send_address_(),
        loss_limit_(0),
        io_thread_proxy_(io_thread_proxy) {}

  virtual bool SendPacket(const Packet& packet) OVERRIDE {
    io_thread_proxy_->PostTask(FROM_HERE,
    base::Bind(&LocalPacketSender::SendPacketToNetwork, this, packet));
    return true;
  }

  virtual void SendPacketToNetwork(const Packet& packet) {
    DCHECK(io_thread_proxy_->RunsTasksOnCurrentThread());
    const uint8* data = packet.data();
    if (loss_limit_ > 0) {
      int r = base::RandInt(0, 100);
      if (r < loss_limit_) {
        VLOG(1) << "Drop packet f:" << static_cast<int>(data[12 + 1])
                << " p:" << static_cast<int>(data[12 + 3])
                << " m:" << static_cast<int>(data[12 + 5]);
        return;
      }
    }
    net::TestCompletionCallback callback;
    scoped_refptr<net::WrappedIOBuffer> buffer(
        new net::WrappedIOBuffer(reinterpret_cast<const char*>(data)));
    udp_socket_->SendTo(buffer.get(), static_cast<int>(packet.size()),
                        send_address_, callback.callback());
  }

  virtual bool SendPackets(const PacketList& packets) OVERRIDE {
    bool out_val = true;
    for (size_t i = 0; i < packets.size(); ++i) {
      const Packet& packet = packets[i];
      out_val |= SendPacket(packet);
    }
    return out_val;
  }

  void SetPacketLoss(int percentage) {
    DCHECK_GE(percentage, 0);
    DCHECK_LT(percentage, 100);
    loss_limit_ = percentage;
  }

  void SetSendAddress(const net::IPEndPoint& send_address) {
    send_address_ = send_address;
  }

 protected:
  virtual ~LocalPacketSender() {}

 private:
  friend class base::RefCountedThreadSafe<LocalPacketSender>;

  net::UDPServerSocket* udp_socket_;  // Not owned by this class.
  net::IPEndPoint send_address_;
  int loss_limit_;
  scoped_refptr<base::TaskRunner> io_thread_proxy_;
};

Transport::Transport(
    scoped_refptr<base::TaskRunner> io_thread_proxy)
    : udp_socket_(new net::UDPServerSocket(NULL, net::NetLog::Source())),
      local_udp_transport_data_(new LocalUdpTransportData(udp_socket_.get(),
                                                          io_thread_proxy)),
      packet_sender_(new LocalPacketSender(udp_socket_.get(), io_thread_proxy)),
      io_thread_proxy_(io_thread_proxy) {}

Transport::~Transport() {}

PacketSender* Transport::packet_sender() {
  return static_cast<PacketSender*>(packet_sender_.get());
}

void Transport::SetSendSidePacketLoss(int percentage) {
  packet_sender_->SetPacketLoss(percentage);
}

void Transport::StopReceiving() {
  local_udp_transport_data_->Close();
}

void Transport::SetLocalReceiver(PacketReceiver* packet_receiver,
                                 std::string ip_address,
                                 std::string local_ip_address,
                                 int port) {
  DCHECK(io_thread_proxy_->RunsTasksOnCurrentThread());
  net::IPEndPoint bind_address, local_bind_address;
  CreateUDPAddress(ip_address, port, &bind_address);
  CreateUDPAddress(local_ip_address, port, &local_bind_address);
  local_udp_transport_data_->set_packet_receiver(packet_receiver);
  udp_socket_->AllowAddressReuse();
  udp_socket_->SetMulticastLoopbackMode(true);
  udp_socket_->Listen(local_bind_address);

  // Start listening once receiver has been set.
  local_udp_transport_data_->ListenTo(bind_address);
}

void Transport::SetSendDestination(std::string ip_address, int port) {
  net::IPEndPoint send_address;
  CreateUDPAddress(ip_address, port, &send_address);
  packet_sender_->SetSendAddress(send_address);
}

}  // namespace test
}  // namespace cast
}  // namespace media