StageProfiDetector::StageProfiDetector(SelectServerInterface *ss,
                                       const vector<string> &widget_paths,
                                       WidgetCallback *callback)
    : m_ss(ss),
      m_callback(callback),
      m_timeout_id(INVALID_TIMEOUT),
      m_backoff(TimeInterval(1, 0), TimeInterval(300, 0)),
      m_socket_factory(NewCallback(this, &StageProfiDetector::SocketConnected)),
      m_tcp_connector(ss, &m_socket_factory, TimeInterval(3, 0)) {
  if (!callback) {
    OLA_FATAL << "No WidgetCallback provided";
    return;
  }

  set<string> paths(widget_paths.begin(), widget_paths.end());
  set<string>::const_iterator iter = paths.begin();
  for (; iter != paths.end(); ++iter) {
    if (iter->empty()) {
      continue;
    }

    if (iter->at(0) == ola::file::PATH_SEPARATOR) {
      STLReplace(&m_usb_widgets, *iter, NULL);
    } else {
      IPV4SocketAddress socket_addr;
      if (EndpointFromString(*iter, &socket_addr)) {
        m_tcp_connector.AddEndpoint(socket_addr, &m_backoff);
      }
    }
  }
}

/**
 * Create a new DesignatedControllerConnection.
 * This listens for connections from the controllers, and will ensure that if
 * any controllers try to connect, at least one will be picked as the
 * designated controller.
 */
DesignatedControllerConnection::DesignatedControllerConnection(
    ola::io::SelectServerInterface *ss,
    const IPV4Address &ip_address,
    ola::e133::MessageBuilder *message_builder,
    TCPConnectionStats *tcp_stats,
    unsigned int max_queue_size)
    : m_ip_address(ip_address),
      m_max_queue_size(max_queue_size),
      m_ss(ss),
      m_message_builder(message_builder),
      m_tcp_stats(tcp_stats),
      m_tcp_socket(NULL),
      m_health_checked_connection(NULL),
      m_message_queue(NULL),
      m_incoming_tcp_transport(NULL),
      m_tcp_socket_factory(
          NewCallback(this, &DesignatedControllerConnection::NewTCPConnection)),
      m_listening_tcp_socket(&m_tcp_socket_factory),
      m_root_inflator(
          NewCallback(this, &DesignatedControllerConnection::RLPDataReceived)),
      m_unsent_messages(false) {
  m_root_inflator.AddInflator(&m_e133_inflator);
  m_e133_inflator.AddInflator(&m_e133_status_inflator);

  m_e133_status_inflator.SetStatusHandler(
      NewCallback(this, &DesignatedControllerConnection::HandleStatusMessage));
}

/**
 * Create a new E133Receiver.
 * @param socket the UDP socket to read from
 * @param status_callback the callback to run when status messages are
 * received.
 * @param rdm_callback the callback to run when RDM messages are received.
 */
E133Receiver::E133Receiver(ola::network::UDPSocket *socket,
                           StatusCallback *status_callback,
                           RDMCallback *rdm_callback)
    : m_udp_socket(socket),
      m_status_callback(status_callback),
      m_rdm_callback(rdm_callback),
      m_root_inflator(new plugin::e131::RootInflator()),
      m_e133_inflator(new plugin::e131::E133Inflator()),
      m_rdm_inflator(new plugin::e131::RDMInflator()),
      m_e133_status_inflator(new plugin::e131::E133StatusInflator()),
      m_incoming_udp_transport(
          new plugin::e131::IncomingUDPTransport(
            m_udp_socket,
            m_root_inflator.get())) {
  m_root_inflator->AddInflator(m_e133_inflator.get());
  m_e133_inflator->AddInflator(m_rdm_inflator.get());
  m_e133_inflator->AddInflator(m_e133_status_inflator.get());

  m_rdm_inflator->SetRDMHandler(
      NewCallback(this, &E133Receiver::HandlePacket));
  m_e133_status_inflator->SetStatusHandler(
      NewCallback(this, &E133Receiver::HandleStatusMessage));

  m_udp_socket->SetOnData(
        NewCallback(m_incoming_udp_transport.get(),
                    &ola::plugin::e131::IncomingUDPTransport::Receive));
}

/*
 * Fetch the state of a plugin.
 */
int FetchPluginState(OlaClientWrapper *wrapper, const options &opts) {
  SelectServer *ss = wrapper->GetSelectServer();
  OlaClient *client = wrapper->GetClient();
  if (opts.plugin_id == 0) {
    DisplayPluginStateHelp(opts);
    exit(1);
  }
  if (!opts.state.empty()) {
    bool state;
    if (ola::StringToBoolTolerant(opts.state, &state)) {
      cout << "Setting state to " << (state ? "enabled" : "disabled") << endl;
      client->SetPluginState(
          (ola::ola_plugin_id) opts.plugin_id,
          state,
          NewSingleCallback(&HandleAck, ss));
    } else {
      cerr << "Invalid state: " << opts.state << endl;
      DisplayPluginStateHelp(opts);
      exit(1);
    }
  } else {
    client->FetchPluginState((ola::ola_plugin_id) opts.plugin_id,
                             NewSingleCallback(&DisplayPluginState, ss));
  }
  return 0;
}

/*
 * Check that queued messages work.
 */
void DmxTriWidgetTest::testQueuedMessages() {
  UID source(1, 2);
  UID destination(0x707a, 0xffffff00);
  vector<string> packets;
  PopulateTod();

  // first try a response which is too short
  const RDMRequest *request = NewQueuedMessageRequest(source, destination, 1);
  uint8_t expected_rdm_command[] = {0x3a, 0x02, 0x01};
  uint8_t small_response[] = {0x3a, 0x04};
  m_endpoint->AddExpectedUsbProDataAndReturn(
      EXTENDED_LABEL,
      expected_rdm_command,
      sizeof(expected_rdm_command),
      EXTENDED_LABEL,
      small_response,
      sizeof(small_response));

  m_widget->SendRDMRequest(
      request,
      NewSingleCallback(this,
                        &DmxTriWidgetTest::ValidateStatus,
                        ola::rdm::RDM_INVALID_RESPONSE,
                        packets));
  m_ss.Run();
  m_endpoint->Verify();

  // now try a proper response
  request = NewQueuedMessageRequest(source, destination, 1);
  uint8_t queued_response[] = {0x3a, 0x00, 0x00, 0x60, 0x52};
  m_endpoint->AddExpectedUsbProDataAndReturn(
      EXTENDED_LABEL,
      expected_rdm_command,
      sizeof(expected_rdm_command),
      EXTENDED_LABEL,
      queued_response,
      sizeof(queued_response));

  uint8_t return_data = 0x52;
  ola::rdm::RDMGetResponse response(
      destination,
      source,
      0,  // transaction #
      ola::rdm::RDM_ACK,
      0,  // message count
      10,  // sub device
      0x0060,  // param id
      &return_data,
      sizeof(return_data));  // data length

  m_widget->SendRDMRequest(
      request,
      NewSingleCallback(this,
                        &DmxTriWidgetTest::ValidateResponse,
                        ola::rdm::RDM_COMPLETED_OK,
                        static_cast<const RDMResponse*>(&response),
                        packets));
  m_ss.Run();
  m_endpoint->Verify();
}

void ArtNetInputPort::PostSetUniverse(Universe *old_universe,
                                      Universe *new_universe) {
  if (new_universe)
    m_node->SetOutputPortUniverse(PortId(), new_universe->UniverseId() % 0xf);
  else
    m_node->DisableOutputPort(PortId());

  if (new_universe && !old_universe) {
    m_node->SetDMXHandler(
        PortId(),
        &m_buffer,
        NewCallback(static_cast<ola::BasicInputPort*>(this),
                    &ArtNetInputPort::DmxChanged));
    m_node->SetOutputPortRDMHandlers(
        PortId(),
        NewCallback(
            this,
            &ArtNetInputPort::RespondWithTod),
        NewCallback(
            this,
            &ArtNetInputPort::TriggerDiscovery),
        ola::NewCallback(
            static_cast<ola::BasicInputPort*>(this),
            &ArtNetInputPort::HandleRDMRequest));

  } else if (!new_universe) {
    m_node->SetDMXHandler(PortId(), NULL, NULL);
    m_node->SetOutputPortRDMHandlers(PortId(), NULL, NULL, NULL);
  }

  if (new_universe)
    TriggerRDMDiscovery(
        NewSingleCallback(this, &ArtNetInputPort::SendTODWithUIDs));
}

bool ArtNetDevice::StartHook() {
  unsigned int subnet = 0;
  StringToInt(m_preferences->GetValue(K_SUBNET_KEY), &subnet);

  unsigned int net;
  StringToInt(m_preferences->GetValue(K_NET_KEY), &net);

  ola::network::Interface interface;
  auto_ptr<ola::network::InterfacePicker> picker(
      ola::network::InterfacePicker::NewPicker());
  ola::network::InterfacePicker::Options options;
  options.include_loopback = m_preferences->GetValueAsBool(K_LOOPBACK_KEY);
  if (!picker->ChooseInterface(
          &interface,
          m_preferences->GetValue(K_IP_KEY),
          options)) {
    OLA_INFO << "Failed to find an interface";
    return false;
  }

  ArtNetNodeOptions node_options;
  node_options.always_broadcast = m_preferences->GetValueAsBool(
      K_ALWAYS_BROADCAST_KEY);
  node_options.use_limited_broadcast_address = m_preferences->GetValueAsBool(
      K_LIMITED_BROADCAST_KEY);
  // OLA Output ports are ArtNet input ports
  StringToInt(m_preferences->GetValue(K_OUTPUT_PORT_KEY),
              &node_options.input_port_count);

  m_node = new ArtNetNode(interface, m_plugin_adaptor, node_options);
  m_node->SetNetAddress(net);
  m_node->SetSubnetAddress(subnet);
  m_node->SetShortName(m_preferences->GetValue(K_SHORT_NAME_KEY));
  m_node->SetLongName(m_preferences->GetValue(K_LONG_NAME_KEY));

  for (unsigned int i = 0; i < node_options.input_port_count; i++) {
    AddPort(new ArtNetOutputPort(this, i, m_node));
  }

  for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++) {
    AddPort(new ArtNetInputPort(this, i, m_plugin_adaptor, m_node));
  }

  if (!m_node->Start()) {
    DeleteAllPorts();
    delete m_node;
    m_node = NULL;
    return false;
  }

  ostringstream str;
  str << K_DEVICE_NAME << " [" << interface.ip_address << "]";
  SetName(str.str());

  m_timeout_id = m_plugin_adaptor->RegisterRepeatingTimeout(
      POLL_INTERVAL,
      NewCallback(m_node, &ArtNetNode::SendPoll));
  return true;
}

/*
 * Check that nacks work as expected.
 */
void DmxTriWidgetTest::testNack() {
  UID source(1, 2);
  UID destination(0x707a, 0xffffff00);

  PopulateTod();

  RDMRequest *request = NewRequest(source, destination, NULL, 0);

  uint8_t expected_rdm_command[] = {0x38, 0x02, 0x00, 0x0a, 0x01, 0x28};
  uint8_t nack_pid_response[] = {0x38, 0x20};  // unknown pid
  m_endpoint->AddExpectedUsbProDataAndReturn(
      EXTENDED_LABEL,
      expected_rdm_command,
      sizeof(expected_rdm_command),
      EXTENDED_LABEL,
      nack_pid_response,
      sizeof(nack_pid_response));

  RDMResponse *response = ola::rdm::NackWithReason(
      request,
      ola::rdm::NR_UNKNOWN_PID);

  m_widget->SendRDMRequest(
      request,
      NewSingleCallback(this,
                        &DmxTriWidgetTest::ValidateResponse,
                        ola::rdm::RDM_COMPLETED_OK,
                        static_cast<const RDMResponse*>(response)));
  m_ss.Run();
  m_endpoint->Verify();
  delete response;

  // try a proxy buffer full
  request = NewRequest(source, destination, NULL, 0);

  uint8_t nack_proxy_response[] = {0x38, 0x2a};  // bad proxy
  m_endpoint->AddExpectedUsbProDataAndReturn(
      EXTENDED_LABEL,
      expected_rdm_command,
      sizeof(expected_rdm_command),
      EXTENDED_LABEL,
      nack_proxy_response,
      sizeof(nack_proxy_response));

  response = ola::rdm::NackWithReason(
      request,
      ola::rdm::NR_PROXY_BUFFER_FULL);

  m_widget->SendRDMRequest(
      request,
      NewSingleCallback(this,
                        &DmxTriWidgetTest::ValidateResponse,
                        ola::rdm::RDM_COMPLETED_OK,
                        static_cast<const RDMResponse*>(response)));
  m_ss.Run();
  m_endpoint->Verify();
  delete response;
}

/*
 * Check that attempting a discovery while another is running fails.
 */
void QueueingRDMControllerTest::testMultipleDiscovery() {
  MockRDMController mock_controller;
  auto_ptr<ola::rdm::DiscoverableQueueingRDMController> controller(
      new ola::rdm::DiscoverableQueueingRDMController(&mock_controller, 1));

  UIDSet uids, uids2;
  UID uid1(2, 3);
  UID uid2(10, 11);
  UID uid3(20, 22);
  UID uid4(65, 45);
  uids.AddUID(uid1);
  uids.AddUID(uid2);
  uids2.AddUID(uid3);
  uids2.AddUID(uid4);

  // trigger discovery, this doesn't run the callback immediately
  mock_controller.AddExpectedDiscoveryCall(true, NULL);
  controller->RunFullDiscovery(
      NewSingleCallback(
          this,
          &QueueingRDMControllerTest::VerifyDiscoveryComplete,
          &uids));
  mock_controller.Verify();
  OLA_ASSERT_FALSE(m_discovery_complete_count);

  // trigger discovery again, this should queue the discovery request
  controller->RunIncrementalDiscovery(
      NewSingleCallback(
          this,
          &QueueingRDMControllerTest::VerifyDiscoveryComplete,
          &uids2));
  mock_controller.Verify();

  // and again
  controller->RunIncrementalDiscovery(
      NewSingleCallback(
          this,
          &QueueingRDMControllerTest::VerifyDiscoveryComplete,
          &uids2));
  mock_controller.Verify();

  // return from the first one, this will trigger the second discovery call
  mock_controller.AddExpectedDiscoveryCall(false, NULL);
  mock_controller.RunDiscoveryCallback(uids);
  OLA_ASSERT_TRUE(m_discovery_complete_count);
  m_discovery_complete_count = 0;
  mock_controller.Verify();

  // now return from the second one, which should complete the 2nd and 3rd
  // requests
  mock_controller.RunDiscoveryCallback(uids2);
  OLA_ASSERT_EQ(2, m_discovery_complete_count);
  m_discovery_complete_count = 0;
  mock_controller.Verify();
}

SimpleE133Device::SimpleE133Device(const Options &options)
    : m_controller(options.controller),
      m_message_builder(ola::acn::CID::Generate(), "E1.33 Device"),
      m_tcp_socket_factory(NewCallback(this, &SimpleE133Device::OnTCPConnect)),
      m_connector(&m_ss, &m_tcp_socket_factory,
                  TimeInterval(FLAGS_tcp_connect_timeout_ms / 1000,
                               (FLAGS_tcp_connect_timeout_ms % 1000) * 1000)),
      m_backoff_policy(TimeInterval(
            FLAGS_tcp_retry_interval_ms / 1000,
            (FLAGS_tcp_retry_interval_ms % 1000) * 1000)),
      m_root_inflator(NewCallback(this, &SimpleE133Device::RLPDataReceived)) {
  m_connector.AddEndpoint(options.controller, &m_backoff_policy);
}

/*
 * Fetch information on plugins.
 */
int FetchPluginInfo(OlaCallbackClientWrapper *wrapper, const options &opts) {
  SelectServer *ss = wrapper->GetSelectServer();
  OlaCallbackClient *client = wrapper->GetClient();
  if (opts.plugin_id > 0) {
    client->FetchPluginDescription(
        (ola::ola_plugin_id) opts.plugin_id,
        NewSingleCallback(&DisplayPluginDescription, ss));
  } else {
    client->FetchPluginList(
        NewSingleCallback(&DisplayPlugins, ss, opts.list_plugin_ids));
  }
  return 0;
}

/**
 * Test that pausing a connection works.
 */
void AdvancedTCPConnectorTest::testPause() {
  ola::network::TCPSocketFactory socket_factory(
      ola::NewCallback(this, &AdvancedTCPConnectorTest::AcceptedConnection));
  TCPAcceptingSocket listening_socket(&socket_factory);
  SetupListeningSocket(&listening_socket);

  AdvancedTCPConnector connector(
      m_ss,
      m_tcp_socket_factory.get(),
      TimeInterval(0, CONNECT_TIMEOUT_IN_MS * 1000));

  // 5 per attempt, up to a max of 30
  LinearBackoffPolicy policy(TimeInterval(5, 0), TimeInterval(30, 0));
  // add endpoint, but make sure it's paused
  connector.AddEndpoint(m_server_address, &policy, true);
  OLA_ASSERT_EQ(1u, connector.EndpointCount());

  ConfirmState(__LINE__, connector, m_server_address,
               AdvancedTCPConnector::PAUSED, 0);

  m_ss->RunOnce(TimeInterval(0, 500000));

  // now unpause
  connector.Resume(m_server_address);
  // The socket may be connected immediately depending on the platform.
  AdvancedTCPConnector::ConnectionState state;
  unsigned int failed_attempts;
  connector.GetEndpointState(m_server_address, &state, &failed_attempts);
  if (state == AdvancedTCPConnector::DISCONNECTED) {
    m_ss->Run();
  }
  OLA_ASSERT_EQ(1u, connector.EndpointCount());
  ConfirmState(__LINE__, connector, m_server_address,
               AdvancedTCPConnector::CONNECTED, 0);

  // check our socket exists
  OLA_ASSERT_TRUE(m_connected_socket);
  m_connected_socket->Close();
  delete m_connected_socket;
  connector.Disconnect(m_server_address, true);

  // state should be updated
  ConfirmState(__LINE__, connector, m_server_address,
               AdvancedTCPConnector::PAUSED, 0);

  // clean up
  connector.RemoveEndpoint(m_server_address);
  OLA_ASSERT_EQ(0u, connector.EndpointCount());

  m_ss->RemoveReadDescriptor(&listening_socket);
}

void SimpleE133Device::OnTCPConnect(TCPSocket *socket) {
  OLA_INFO << "Opened new TCP connection: " << socket;

  m_socket.reset(socket);
  m_in_transport.reset(new IncomingTCPTransport(&m_root_inflator, socket));

  m_message_queue.reset(
      new MessageQueue(m_socket.get(), &m_ss, m_message_builder.pool()));

  m_health_checked_connection.reset(new E133HealthCheckedConnection(
      &m_message_builder,
      m_message_queue.get(),
      NewSingleCallback(this, &SimpleE133Device::SocketClosed),
      &m_ss));

  socket->SetOnData(NewCallback(this, &SimpleE133Device::ReceiveTCPData));
  socket->SetOnClose(NewSingleCallback(this, &SimpleE133Device::SocketClosed));
  m_ss.AddReadDescriptor(socket);


  if (!m_health_checked_connection->Setup()) {
    OLA_WARN << "Failed to setup heartbeat controller for " << m_controller;
    SocketClosed();
    return;
  }
}

/*
 * Check that RemoveWriteDescriptor is reentrant.
 * Similar to the case above, but this removes & deletes the descriptor from
 * within the OnRead callback of the same descriptor.
 */
void SelectServerTest::testRemoveWriteWhenReadable() {
  // Ownership is transferred to the SelectServer.
  LoopbackDescriptor *loopback = new LoopbackDescriptor();
  loopback->Init();

  loopback->SetOnData(NewCallback(
      this, &SelectServerTest::ReadDataAndRemove,
      static_cast<ConnectedDescriptor*>(loopback)));
  loopback->SetOnWritable(NewCallback(this, &SelectServerTest::NullHandler));

  OLA_ASSERT_TRUE(m_ss->AddReadDescriptor(loopback));
  OLA_ASSERT_TRUE(m_ss->AddWriteDescriptor(loopback));
  OLA_ASSERT_EQ(2, connected_read_descriptor_count->Get());
  OLA_ASSERT_EQ(1, write_descriptor_count->Get());
  OLA_ASSERT_EQ(0, read_descriptor_count->Get());

  // Send some data to make this descriptor readable.
  uint8_t data[] = {'a'};
  loopback->Send(data, arraysize(data));

  m_ss->Run();
  OLA_ASSERT_EQ(0, write_descriptor_count->Get());
  OLA_ASSERT_EQ(1, connected_read_descriptor_count->Get());
  OLA_ASSERT_EQ(0, read_descriptor_count->Get());
}

/*
 * Test the Method Callbacks
 */
void CallbackTest::testMethodCallbacks4() {
  // test 2 arg callbacks that return unsigned ints
  BaseCallback4<void, unsigned int, int, char, const string&> *c1 =
    NewSingleCallback(
      this,
      &CallbackTest::Method4);
  c1->Run(TEST_INT_VALUE, TEST_INT_VALUE2, TEST_CHAR_VALUE, TEST_STRING_VALUE);
  BaseCallback4<void, unsigned int, int, char, const string&> *c2 = NewCallback(
      this,
      &CallbackTest::Method4);
  c2->Run(TEST_INT_VALUE, TEST_INT_VALUE2, TEST_CHAR_VALUE, TEST_STRING_VALUE);
  c2->Run(TEST_INT_VALUE, TEST_INT_VALUE2, TEST_CHAR_VALUE, TEST_STRING_VALUE);
  delete c2;

  // test 2 arg callbacks that return bools
  BaseCallback4<bool, unsigned int, int, char, const string&> *c3 =
    NewSingleCallback(
      this,
      &CallbackTest::BoolMethod4);
  OLA_ASSERT_TRUE(c3->Run(TEST_INT_VALUE, TEST_INT_VALUE2, TEST_CHAR_VALUE,
                         TEST_STRING_VALUE));
  BaseCallback4<bool, unsigned int, int, char, const string&> *c4 =
    NewCallback(
      this,
      &CallbackTest::BoolMethod4);
  OLA_ASSERT_TRUE(c4->Run(TEST_INT_VALUE, TEST_INT_VALUE2, TEST_CHAR_VALUE,
                         TEST_STRING_VALUE));
  OLA_ASSERT_TRUE(c4->Run(TEST_INT_VALUE, TEST_INT_VALUE2, TEST_CHAR_VALUE,
                         TEST_STRING_VALUE));
  delete c4;
}

/**
 * Check pausing doesn't mark the channel as bad.
 */
void HealthCheckedConnectionTest::testPauseAndResume() {
  MockHealthCheckedConnection connection(&socket,
                                         &m_ss,
                                         heartbeat_interval,
                                         options,
                                         &m_clock);
  socket.SetOnData(
      NewCallback(&connection, &MockHealthCheckedConnection::ReadData));
  connection.Setup();
  m_ss.AddReadDescriptor(&socket);
  connection.Setup();

  m_ss.RegisterSingleTimeout(
      TimeInterval(1, 0),
      NewSingleCallback(this,
                        &HealthCheckedConnectionTest::PauseReading,
                        &connection));
  m_ss.RegisterSingleTimeout(
      TimeInterval(3, 0),
      NewSingleCallback(this,
                        &HealthCheckedConnectionTest::ResumeReading,
                        &connection));

  m_ss.Run();
  OLA_ASSERT_TRUE(connection.ChannelOk());
}

/*
 * test the IntToString function.
 */
void StringUtilsTest::testIntToString() {
  OLA_ASSERT_EQ(string("0"), IntToString(0));
  OLA_ASSERT_EQ(string("1234"), IntToString(1234));
  OLA_ASSERT_EQ(string("-1234"), IntToString(-1234));
  unsigned int i = 42;
  OLA_ASSERT_EQ(string("42"), IntToString(i));
}

void StringUtilsTest::testFormatData() {
  uint8_t data[] = {0, 'H', 'e', 'l', 'l', 'o', ' ', 'W', 'o', 'r', 'l', 'd',
                    1, 2};
  std::stringstream str;
  FormatData(&str, data, sizeof(data));
  OLA_ASSERT_EQ(
      string("00 48 65 6c 6c 6f 20 57  .Hello W\n"
             "6f 72 6c 64 01 02        orld..\n"),
      str.str());

  // try 4 bytes per line with a 2 space indent
  str.str("");
  FormatData(&str, data, sizeof(data), 2, 4);
  OLA_ASSERT_EQ(
      string("  00 48 65 6c  .Hel\n"
             "  6c 6f 20 57  lo W\n"
             "  6f 72 6c 64  orld\n"
             "  01 02        ..\n"),
      str.str());

  str.str("");
  // try ending on the block boundary
  uint8_t data1[] = {'H', 'e', 'l', 'l', 'o', ' ', 'W', 'o'};
  FormatData(&str, data1, sizeof(data1), 0, 4);
  OLA_ASSERT_EQ(
      string("48 65 6c 6c  Hell\n"
             "6f 20 57 6f  o Wo\n"),
      str.str());
}

/*
 * Check that we don't invalid iterators by removing descriptors during an
 * on_Write callback.
 */
void SelectServerTest::testRemoveOthersWhenWriteable() {
  Descriptors read_set, write_set, delete_set;
  LoopbackDescriptor loopback1, loopback2, loopback3;
  loopback1.Init();
  loopback2.Init();
  loopback3.Init();

  OLA_ASSERT_TRUE(m_ss->AddWriteDescriptor(&loopback1));
  OLA_ASSERT_TRUE(m_ss->AddWriteDescriptor(&loopback2));
  OLA_ASSERT_TRUE(m_ss->AddWriteDescriptor(&loopback3));

  write_set.insert(&loopback1);
  write_set.insert(&loopback2);
  write_set.insert(&loopback3);

  loopback1.SetOnWritable(NewCallback(
      this, &SelectServerTest::NullHandler));
  loopback2.SetOnWritable(NewCallback(
      this,
      &SelectServerTest::RemoveAndDeleteDescriptors,
      read_set, write_set, delete_set));

  loopback3.SetOnWritable(NewCallback(
      this, &SelectServerTest::NullHandler));

  OLA_ASSERT_EQ(3, write_descriptor_count->Get());
  OLA_ASSERT_EQ(1, connected_read_descriptor_count->Get());

  m_ss->Run();

  OLA_ASSERT_EQ(0, write_descriptor_count->Get());
  OLA_ASSERT_EQ(1, connected_read_descriptor_count->Get());
  OLA_ASSERT_EQ(0, read_descriptor_count->Get());
}

/*
 * Test the StringEndsWith function.
 */
void StringUtilsTest::testEndsWith() {
  string input = "foo bar baz";
  OLA_ASSERT_TRUE(StringEndsWith(input, "baz"));
  OLA_ASSERT_TRUE(StringEndsWith(input, " baz"));
  OLA_ASSERT_TRUE(StringEndsWith(input, "bar baz"));
  OLA_ASSERT_TRUE(StringEndsWith(input, ""));
  OLA_ASSERT_FALSE(StringEndsWith(input, "foo"));
}