// $Id$

#include "Client_Task.h"
#include "ace/Get_Opt.h"
#include "ace/Argv_Type_Converter.h"
#include "tao/Messaging/Messaging.h"

ACE_RCSID(Muxing, client, "$Id$")

const char *ior = "file://test.ior";

// 3 clients with 2 threads each send this many messages.
// so the server should expect NUM_MSGS * 6 total.
static const int NUM_MSGS = 100;
static const int NUM_THRDS = 2;
static const int MSG_SIZE = 4096;

int
parse_args (int argc, char *argv[])
{
  ACE_Get_Arg_Opt<char> get_opts (argc, argv, "k:");
  int c;

  while ((c = get_opts ()) != -1)
    switch (c)
  {
    case 'k':
      ior = get_opts.opt_arg ();
      break;

    case '?':
    default:
      ACE_ERROR_RETURN ((LM_ERROR,
        "usage:  %s "
        "-k <ior>"
        "\n",
        argv [0]),
        -1);
  }
  // Indicates sucessful parsing of the command line
  return 0;
}

int
ACE_TMAIN (int argc, ACE_TCHAR *argv[])
{
  ACE_Argv_Type_Converter convert (argc, argv);

  ACE_DEBUG ((LM_DEBUG, "Starting client\n"));

  ACE_TRY_NEW_ENV
  {
    CORBA::ORB_var orb =
      CORBA::ORB_init (convert.get_argc(), convert.get_ASCII_argv(), "" ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;

    if (parse_args (convert.get_argc(), convert.get_ASCII_argv()) != 0)
      return 1;

    CORBA::Object_var tmp =
      orb->string_to_object(ior ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;

    Test::Payload_Receiver_var payload_receiver =
      Test::Payload_Receiver::_narrow(tmp.in () ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;

    if (CORBA::is_nil (payload_receiver.in ()))
    {
      ACE_ERROR_RETURN ((LM_DEBUG,
        "Nil coordinator reference <%s>\n",
        ior),
        1);
    }

    Client_Task task0 (ACE_Thread_Manager::instance (),
      payload_receiver.in (),
      NUM_MSGS,
      MSG_SIZE,
      orb.in (),
      Messaging::SYNC_WITH_TARGET);
    Client_Task task1 (ACE_Thread_Manager::instance (),
      payload_receiver.in (),
      NUM_MSGS,
      MSG_SIZE,
      orb.in (),
      Messaging::SYNC_WITH_TRANSPORT);
    Client_Task task2 (ACE_Thread_Manager::instance (),
      payload_receiver.in (),
      NUM_MSGS,
      MSG_SIZE,
      orb.in (),
      Messaging::SYNC_NONE);

    ACE_DEBUG ((LM_DEBUG, "(%P) Activating threads in client\n"));
    if (task0.activate (THR_NEW_LWP | THR_JOINABLE, NUM_THRDS, 1) == -1)
    {
      ACE_ERROR ((LM_ERROR, "Error activating client task\n"));
    }
    if (task1.activate (THR_NEW_LWP | THR_JOINABLE, NUM_THRDS, 1) == -1)
    {
      ACE_ERROR ((LM_ERROR, "Error activating client task\n"));
    }
    if (task2.activate (THR_NEW_LWP | THR_JOINABLE, NUM_THRDS, 1) == -1)
    {
      ACE_ERROR ((LM_ERROR, "Error activating client task\n"));
    }

    ACE_Time_Value end_time = ACE_OS::gettimeofday() + ACE_Time_Value(10);
    while (ACE_OS::gettimeofday() < end_time)
    {
      ACE_Time_Value tv (0, 100 * 1000);
      orb->run (tv ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      if (task0.done() && task1.done() && task2.done())
        break;
    }

    ACE_Thread_Manager::instance ()->wait ();
    ACE_DEBUG ((LM_DEBUG, "(%P) Threads finished\n"));

    while (orb->work_pending())
    {
      ACE_Time_Value tv(0, 100 * 1000);
      orb->run(tv ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }

    orb->destroy (ACE_ENV_SINGLE_ARG_PARAMETER);
    ACE_TRY_CHECK;
  }
  ACE_CATCHANY
  {
    ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,
      "Exception caught:");
    return 1;
  }
  ACE_ENDTRY;

  ACE_DEBUG ((LM_DEBUG, "Ending client\n"));

  return 0;
}