/**
 * Check that we send OSC messages correctly.
 */
void OSCNodeTest::testSendBlob() {
  // First up create a UDP socket to receive the messages on.
  // Port 0 means 'ANY'
  IPV4SocketAddress socket_address(IPV4Address::Loopback(), 0);
  // Bind the socket, set the callback, and register with the select server.
  OLA_ASSERT_TRUE(m_udp_socket.Bind(socket_address));
  m_udp_socket.SetOnData(NewCallback(this, &OSCNodeTest::UDPSocketReady));
  OLA_ASSERT_TRUE(m_ss.AddReadDescriptor(&m_udp_socket));
  // Store the local address of the UDP socket so we know where to tell the
  // OSCNode to send to.
  OLA_ASSERT_TRUE(m_udp_socket.GetSocketAddress(&socket_address));

  // Setup the OSCTarget pointing to the local socket address
  OSCTarget target(socket_address, TEST_OSC_ADDRESS);
  // Add the target to the node.
  m_osc_node->AddTarget(TEST_GROUP, target);
  // Send the data
  OLA_ASSERT_TRUE(m_osc_node->SendData(TEST_GROUP, OSCNode::FORMAT_BLOB,
                  m_dmx_data));

  // Run the SelectServer this will return either when UDPSocketReady
  // completes, or the abort timeout triggers.
  m_ss.Run();

  // Remove target
  OLA_ASSERT_TRUE(m_osc_node->RemoveTarget(TEST_GROUP, target));
  // Try to remove it a second time
  OLA_ASSERT_FALSE(m_osc_node->RemoveTarget(TEST_GROUP, target));

  // Try to remove the target from a group that doesn't exist
  OLA_ASSERT_FALSE(m_osc_node->RemoveTarget(TEST_GROUP + 1, target));
}

/**
 * Called when a de-registration request completes.
 */
void DeRegisterCallback(ola::io::SelectServer *ss,
                        unsigned int *registrations_active, bool ok) {
  if (ok) {
    OLA_INFO << "Registered E1.33 device";
  } else {
    OLA_WARN << "Failed to register E1.33 device";
  }
  if (--(*registrations_active) == 0)
    ss->Terminate();
}

/**
 * Called when data arrives on our UDP socket. We check that this data matches
 * the expected OSC packet
 */
void OSCNodeTest::UDPSocketReady() {
  uint8_t data[500];  // 500 bytes is more than enough
  ssize_t data_read = sizeof(data);
  // Read the received packet into 'data'.
  OLA_ASSERT_TRUE(m_udp_socket.RecvFrom(data, &data_read));
  // Verify it matches the expected packet
  ASSERT_DATA_EQUALS(__LINE__, OSC_BLOB_DATA,
                     sizeof(OSC_BLOB_DATA), data, data_read);
  // Stop the SelectServer
  m_ss.Terminate();
}

/**
 * Called when we receive DMX data via OSC. We check this matches what we
 * expect, and then stop the SelectServer.
 */
void OSCNodeTest::DMXHandler(const DmxBuffer &dmx) {
  m_received_data = dmx;
  m_ss.Terminate();
}

/**
 * Check that we receive OSC messages correctly.
 */
void OSCNodeTest::testReceive() {
  DmxBuffer expected_data;

  // Register the test OSC Address with the OSCNode using the DMXHandler as the
  // callback.
  OLA_ASSERT_TRUE(m_osc_node->RegisterAddress(
      TEST_OSC_ADDRESS, NewCallback(this, &OSCNodeTest::DMXHandler)));

  // Attempt to register the same address with a different path, this should
  // return false
  OLA_ASSERT_FALSE(m_osc_node->RegisterAddress(
      TEST_OSC_ADDRESS,
      NewCallback(this, &OSCNodeTest::DMXHandler)));

  // Using our test UDP socket, send the OSC_BLOB_DATA to the default OSC
  // port. The OSCNode should receive the packet and call DMXHandler.
  IPV4SocketAddress dest_address(IPV4Address::Loopback(),
                                 m_osc_node->ListeningPort());

  // send a single float update
  m_udp_socket.SendTo(OSC_SINGLE_FLOAT_DATA, sizeof(OSC_SINGLE_FLOAT_DATA),
                      dest_address);
  m_ss.Run();
  OLA_ASSERT_EQ(512u, m_received_data.Size());
  expected_data.SetChannel(0, 127);
  OLA_ASSERT_EQ(expected_data, m_received_data);

  // now send a blob update
  m_udp_socket.SendTo(OSC_BLOB_DATA, sizeof(OSC_BLOB_DATA),
                      dest_address);
  // Run the SelectServer, this will return either when DMXHandler
  // completes, or the abort timeout triggers.
  m_ss.Run();

  OLA_ASSERT_EQ(11u, m_received_data.Size());
  expected_data.SetFromString("0,1,2,3,4,5,6,7,8,9,10");
  OLA_ASSERT_EQ(expected_data, m_received_data);

  // Now try sending a float update.
  m_udp_socket.SendTo(OSC_SINGLE_FLOAT_DATA, sizeof(OSC_SINGLE_FLOAT_DATA),
                      dest_address);
  m_ss.Run();
  OLA_ASSERT_EQ(11u, m_received_data.Size());
  expected_data.SetChannel(0, 127);
  OLA_ASSERT_EQ(expected_data, m_received_data);

  // Now try sending an int update.
  m_udp_socket.SendTo(OSC_SINGLE_INT_DATA, sizeof(OSC_SINGLE_INT_DATA),
                      dest_address);
  m_ss.Run();
  OLA_ASSERT_EQ(11u, m_received_data.Size());
  expected_data.SetChannel(5, 140);
  OLA_ASSERT_EQ(expected_data, m_received_data);

  // An 'ii' update
  m_udp_socket.SendTo(OSC_INT_TUPLE_DATA, sizeof(OSC_INT_TUPLE_DATA),
                      dest_address);
  m_ss.Run();
  OLA_ASSERT_EQ(11u, m_received_data.Size());
  expected_data.SetChannel(7, 90);
  OLA_ASSERT_EQ(expected_data, m_received_data);

  // An 'if' update
  m_udp_socket.SendTo(OSC_FLOAT_TUPLE_DATA, sizeof(OSC_FLOAT_TUPLE_DATA),
                      dest_address);
  m_ss.Run();
  OLA_ASSERT_EQ(11u, m_received_data.Size());
  expected_data.SetChannel(8, 127);
  OLA_ASSERT_EQ(expected_data, m_received_data);

  // De-regsiter
  OLA_ASSERT_TRUE(m_osc_node->RegisterAddress(TEST_OSC_ADDRESS, NULL));
  // De-register a second time
  OLA_ASSERT_TRUE(m_osc_node->RegisterAddress(TEST_OSC_ADDRESS, NULL));
}

/**
 * Send a GET request
 */
void SimpleE133Controller::SendGetRequest(const UID &dst_uid,
                                          uint16_t endpoint,
                                          uint16_t pid,
                                          const uint8_t *data,
                                          unsigned int length) {
  // send a second one
  ola::rdm::RDMGetRequest *command = new ola::rdm::RDMGetRequest(
      m_src_uid,
      dst_uid,
      0,  // transaction #
      1,  // port id
      0,  // message count
      ola::rdm::ROOT_RDM_DEVICE,  // sub device
      pid,  // param id
      data,  // data
      length);  // data length

  if (!SendRequest(dst_uid, endpoint, command)) {
    OLA_FATAL << "Failed to send request";
    m_ss.Terminate();
  } else if (dst_uid.IsBroadcast()) {
    OLA_INFO << "Request broadcast";
    m_ss.Terminate();
  } else {
    OLA_INFO << "Request sent, waiting for response";
  }
}

void RootSenderTest::testRootSenderWithCIDs(const CID &root_cid,
                                            const CID &send_cid) {
  std::auto_ptr<Callback0<void> > stop_closure(
      NewCallback(this, &RootSenderTest::Stop));

  // inflators
  MockInflator inflator(send_cid, stop_closure.get());
  RootInflator root_inflator;
  OLA_ASSERT(root_inflator.AddInflator(&inflator));

  // sender
  RootSender root_sender(root_cid);

  // setup the socket
  ola::network::UDPSocket socket;
  OLA_ASSERT(socket.Init());
  OLA_ASSERT(
      socket.Bind(IPV4SocketAddress(IPV4Address::Loopback(), 0)));
  OLA_ASSERT(socket.EnableBroadcast());

  IncomingUDPTransport incoming_udp_transport(&socket, &root_inflator);
  socket.SetOnData(NewCallback(&incoming_udp_transport,
                               &IncomingUDPTransport::Receive));
  OLA_ASSERT(m_ss->AddReadDescriptor(&socket));

  // Get the port we bound to.
  IPV4SocketAddress local_address;
  OLA_ASSERT(socket.GetSocketAddress(&local_address));

  OutgoingUDPTransportImpl udp_transport_impl(&socket);
  OutgoingUDPTransport outgoing_udp_transport(&udp_transport_impl,
      IPV4Address::Loopback(), local_address.Port());

  // now actually send some data
  MockPDU mock_pdu(4, 8);

  if (root_cid == send_cid)
    OLA_ASSERT(root_sender.SendPDU(MockPDU::TEST_VECTOR,
                                       mock_pdu,
                                       &outgoing_udp_transport));
  else
    OLA_ASSERT(root_sender.SendPDU(MockPDU::TEST_VECTOR,
                                       mock_pdu,
                                       send_cid,
                                       &outgoing_udp_transport));

  SingleUseCallback0<void> *closure =
    NewSingleCallback(this, &RootSenderTest::FatalStop);
  m_ss->RegisterSingleTimeout(ABORT_TIMEOUT_IN_MS, closure);
  m_ss->Run();
}

/**
 * main
 */
int main(int argc, char *argv[]) {
  ola::AppInit(&argc, argv, "[options] [services]",
               "Register one or more E1.33 services with SLP. [services] is\n"
               "a list of IP, UIDs in the form: uid[@ip], e.g. \n"
               "7a70:00000001 (default ip) or 7a70:[email protected]\n");

  if (argc < 2) {
    ola::DisplayUsage();
    exit(ola::EXIT_USAGE);
  }

  multimap<IPV4Address, UID> services;
  for (int i = 1; i < argc; i++)
    ProcessService(argv[i], &services);

  // signal handler
  ola::InstallSignal(SIGINT, InteruptSignal);

  auto_ptr<ola::e133::BaseSLPThread> slp_thread(
    ola::e133::SLPThreadFactory::NewSLPThread(&ss));

  if (!slp_thread->Init()) {
    OLA_WARN << "SLPThread Init() failed";
    exit(ola::EXIT_UNAVAILABLE);
  }

  if (!slp_thread->Start()) {
    OLA_WARN << "SLPThread Start() failed";
    exit(ola::EXIT_UNAVAILABLE);
  }
  multimap<IPV4Address, UID>::const_iterator iter;
  for (iter = services.begin(); iter != services.end(); ++iter) {
    slp_thread->RegisterDevice(
      ola::NewSingleCallback(&RegisterCallback),
      iter->first, iter->second, FLAGS_lifetime);
  }

  ss.Run();

  // start the de-registration process
  unsigned int registrations_active = services.size();
  for (iter = services.begin(); iter != services.end(); ++iter) {
    slp_thread->DeRegisterDevice(
        ola::NewSingleCallback(&DeRegisterCallback, &ss, &registrations_active),
        iter->first, iter->second);
  }
  ss.Run();

  slp_thread->Join(NULL);
}

void SimpleE133Node::Run() {
  m_slp_thread.Register(
    ola::NewSingleCallback(this, &SimpleE133Node::RegisterCallback),
    m_service_name,
    m_lifetime);

  m_ss.Run();
  OLA_INFO << "Starting shutdown process";

  m_slp_thread.DeRegister(
    ola::NewSingleCallback(this, &SimpleE133Node::DeRegisterCallback),
    m_service_name);
  m_ss.Run();
}

void RootSenderTest::testRootSenderWithCIDs(const CID &root_cid,
                                            const CID &send_cid) {
  std::auto_ptr<Callback0<void> > stop_closure(
      NewCallback(this, &RootSenderTest::Stop));

  // inflators
  MockInflator inflator(send_cid, stop_closure.get());
  RootInflator root_inflator;
  CPPUNIT_ASSERT(root_inflator.AddInflator(&inflator));

  // sender
  RootSender root_sender(root_cid);

  // setup the socket
  ola::network::UdpSocket socket;
  CPPUNIT_ASSERT(socket.Init());
  CPPUNIT_ASSERT(socket.Bind(ACN_PORT));
  CPPUNIT_ASSERT(socket.EnableBroadcast());

  IncomingUDPTransport incoming_udp_transport(&socket, &root_inflator);
  socket.SetOnData(NewCallback(&incoming_udp_transport,
                               &IncomingUDPTransport::Receive));
  CPPUNIT_ASSERT(m_ss->AddReadDescriptor(&socket));

  // outgoing transport
  IPV4Address addr;
  CPPUNIT_ASSERT(IPV4Address::FromString("255.255.255.255", &addr));

  OutgoingUDPTransportImpl udp_transport_impl(&socket);
  OutgoingUDPTransport outgoing_udp_transport(&udp_transport_impl, addr);

  // now actually send some data
  MockPDU mock_pdu(4, 8);

  if (root_cid == send_cid)
    CPPUNIT_ASSERT(root_sender.SendPDU(MockPDU::TEST_VECTOR,
                                       mock_pdu,
                                       &outgoing_udp_transport));
  else
    CPPUNIT_ASSERT(root_sender.SendPDU(MockPDU::TEST_VECTOR,
                                       mock_pdu,
                                       send_cid,
                                       &outgoing_udp_transport));

  SingleUseCallback0<void> *closure =
    NewSingleCallback(this, &RootSenderTest::FatalStop);
  m_ss->RegisterSingleTimeout(ABORT_TIMEOUT_IN_MS, closure);
  m_ss->Run();
}

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;
  }
}

/**
 * Main
 */
int main(int argc, char *argv[]) {
  ola::AppInit(argc, argv);
  ola::SetHelpString("[options]", "Locate E1.33 SLP services.");
  ola::ParseFlags(&argc, argv);
  ola::InitLoggingFromFlags();

  ola::InstallSignal(SIGINT, InteruptSignal);

  auto_ptr<ola::e133::BaseSLPThread> slp_thread(
    ola::e133::SLPThreadFactory::NewSLPThread(&ss));
  slp_thread->SetNewDeviceCallback(ola::NewCallback(&DiscoveryDone));

  if (!slp_thread->Init()) {
    OLA_WARN << "Init failed";
    exit(ola::EXIT_UNAVAILABLE);
  }

  if (!slp_thread->Start()) {
    OLA_WARN << "SLPThread Start() failed";
    exit(ola::EXIT_UNAVAILABLE);
  }

  ss.Run();
  slp_thread->Join(NULL);
}

/**
 * Init this node
 */
bool SimpleE133Node::Init() {
  // setup notifications for stdin & turn off buffering
  m_stdin_descriptor.SetOnData(ola::NewCallback(this, &SimpleE133Node::Input));
  m_ss.AddReadDescriptor(&m_stdin_descriptor);
  tcgetattr(STDIN_FILENO, &m_old_tc);
  termios new_tc = m_old_tc;
  new_tc.c_lflag &= static_cast<tcflag_t>(~ICANON & ~ECHO);
  tcsetattr(STDIN_FILENO, TCSANOW, &new_tc);

  if (!m_e133_device.Init())
    return false;

  // register the root endpoint
  m_e133_device.SetRootEndpoint(&m_root_endpoint);
  // add a single endpoint
  m_endpoint_manager.RegisterEndpoint(1, &m_first_endpoint);

  // register in SLP
  OLA_INFO << "service is " << m_service_name;
  if (!m_slp_thread.Init()) {
    OLA_WARN << "SlpThread Init() failed";
    return false;
  }

  m_slp_thread.Start();
  return true;
}

void OPCServerTest::testUnknownCommand() {
  uint8_t data[] = {1, 1, 0, 2, 3, 4};
  m_client_socket->Send(data, arraysize(data));
  m_ss.Run();

  DmxBuffer buffer;
  buffer.SetFromString("3,4");
  OLA_ASSERT_EQ(static_cast<uint8_t>(1), m_command);
  OLA_ASSERT_EQ(m_received_data, buffer);
}

void SimpleE133Device::SocketClosed() {
  OLA_INFO << "controller connection was closed";

  m_health_checked_connection.reset();
  m_message_queue.reset();
  m_in_transport.reset();
  m_ss.RemoveReadDescriptor(m_socket.get());
  m_socket.reset();
  m_connector.Disconnect(m_controller);
}

/**
 * Handle a RDM message.
 */
void SimpleE133Controller::HandlePacket(
    const ola::e133::E133RDMMessage &rdm_message) {
  OLA_INFO << "RDM callback executed with code: " <<
    ola::rdm::ResponseCodeToString(rdm_message.response_code);

  m_ss.Terminate();

  if (rdm_message.response_code != ola::rdm::RDM_COMPLETED_OK)
    return;

  switch (rdm_message.response->ResponseType()) {
    case ola::rdm::RDM_NACK_REASON:
      HandleNack(rdm_message.response);
      return;
    default:
      break;
  }

  const RDMResponse *response = rdm_message.response;

  const ola::rdm::PidDescriptor *pid_descriptor = m_pid_helper->GetDescriptor(
      response->ParamId(),
      response->SourceUID().ManufacturerId());
  const ola::messaging::Descriptor *descriptor = NULL;

  if (pid_descriptor) {
    switch (response->CommandClass()) {
      case ola::rdm::RDMCommand::GET_COMMAND_RESPONSE:
        descriptor = pid_descriptor->GetResponse();
        break;
      case ola::rdm::RDMCommand::SET_COMMAND_RESPONSE:
        descriptor = pid_descriptor->SetResponse();
        break;
      default:
        OLA_WARN << "Unknown command class " << response->CommandClass();
    }
  }

  auto_ptr<const ola::messaging::Message> message;
  if (descriptor)
    message.reset(m_pid_helper->DeserializeMessage(descriptor,
                                                   response->ParamData(),
                                                   response->ParamDataSize()));

  if (message.get())
    cout << m_pid_helper->PrettyPrintMessage(
        response->SourceUID().ManufacturerId(),
        response->CommandClass() == ola::rdm::RDMCommand::SET_COMMAND_RESPONSE,
        response->ParamId(),
        message.get());
  else
    m_command_printer.DisplayResponse(response, true);
}

void OPCServerTest::SendDataAndCheck(uint8_t channel,
                                     const DmxBuffer &buffer) {
  unsigned int dmx_size = buffer.Size();
  uint8_t data[dmx_size + 4];
  data[0] = channel;
  data[1] = SET_PIXELS_COMMAND;
  ola::utils::SplitUInt16(static_cast<uint16_t>(dmx_size), &data[2], &data[3]);
  buffer.Get(data + 4, &dmx_size);
  m_client_socket->Send(data, dmx_size + 4);

  m_ss.Run();
  OLA_ASSERT_EQ(m_received_data, buffer);
}

/*
 * Test the UDPTransport
 */
void UDPTransportTest::testUDPTransport() {
  CID cid;
  std::auto_ptr<Callback0<void> > stop_closure(
      NewCallback(this, &UDPTransportTest::Stop));
  MockInflator inflator(cid, stop_closure.get());

  // setup the socket
  ola::network::UDPSocket socket;
  OLA_ASSERT(socket.Init());
  OLA_ASSERT(socket.Bind(IPV4SocketAddress(IPV4Address::Loopback(), 0)));
  OLA_ASSERT(socket.EnableBroadcast());

  // Get the port we bound to.
  IPV4SocketAddress local_address;
  OLA_ASSERT(socket.GetSocketAddress(&local_address));

  IncomingUDPTransport incoming_udp_transport(&socket, &inflator);
  socket.SetOnData(NewCallback(&incoming_udp_transport,
                               &IncomingUDPTransport::Receive));
  OLA_ASSERT(m_ss->AddReadDescriptor(&socket));

  // outgoing transport
  OutgoingUDPTransportImpl udp_transport_impl(&socket);
  OutgoingUDPTransport outgoing_udp_transport(&udp_transport_impl,
      IPV4Address::Loopback(), local_address.Port());

  // now actually send some data
  PDUBlock<PDU> pdu_block;
  MockPDU mock_pdu(4, 8);
  pdu_block.AddPDU(&mock_pdu);
  OLA_ASSERT(outgoing_udp_transport.Send(pdu_block));

  SingleUseCallback0<void> *closure =
    NewSingleCallback(this, &UDPTransportTest::FatalStop);
  m_ss->RegisterSingleTimeout(ABORT_TIMEOUT_IN_MS, closure);
  m_ss->Run();
}

void OPCServerTest::testLargeFrame() {
  uint8_t data[1028];
  data[0] = 1;
  data[1] = 0;
  data[2] = 4;
  data[3] = 0;
  for (unsigned int i = 0; i < 1024; i++) {
    data[i + 4] = i;
  }

  m_client_socket->Send(data, arraysize(data));
  m_ss.Run();

  DmxBuffer buffer(data + 4, ola::DMX_UNIVERSE_SIZE);
  OLA_ASSERT_EQ(m_received_data, buffer);
}

/**
 * Called when there is data on stdin.
 */
void SimpleE133Node::Input() {
  switch (getchar()) {
    case 'c':
      m_e133_device.CloseTCPConnection();
      break;
    case 'q':
      m_ss.Terminate();
      break;
    case 's':
      SendUnsolicited();
      break;
    case 't':
      DumpTCPStats();
      break;
    default:
      break;
  }
}