client.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

// client.cpp,v 1.8 2003/04/21 18:28:11 irfan Exp

// ================================================================
//
//
// = FILENAME
//     client.cpp
//
// = DESCRIPTION
//     This is a client implementation.
//
// = AUTHOR
//     Irfan Pyarali
//
// ================================================================

#include "ace/Get_Opt.h"
#include "ace/Task.h"
#include "tao/ORB_Core.h"
#include "tao/RTCORBA/RTCORBA.h"
#include "tao/Strategies/advanced_resource.h"
#include "../check_supported_priorities.cpp"
#include "../common_args.cpp"
#include "testC.h"

ACE_RCSID(Profile_And_Endpoint_Selection, client, "client.cpp,v 1.8 2003/04/21 18:28:11 irfan Exp")

static int iterations = 1;
static int debug = 1;
static int shutdown_server = 0;
static const char *ior = "file://ior";
static const char *invocation_priorities_file = "empty_file";
static const char *protocols_file = "empty_file";
static const char *bands_file = "empty_file";

static int
parse_args (int argc, char **argv)
{
  ACE_Get_Opt get_opts (argc, argv, "b:d:i:k:p:t:x");
  int c;

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

      case 'd':
        debug = ::atoi (get_opts.opt_arg ());
        break;

      case 'i':
        iterations = ::atoi (get_opts.opt_arg ());
        break;

      case 'k':
        ior = get_opts.opt_arg ();
        break;

      case 'p':
        protocols_file = get_opts.opt_arg ();
        break;

      case 't':
        invocation_priorities_file = get_opts.opt_arg ();
        break;

      case 'x':
        shutdown_server = 1;
        break;

      case '?':
      default:
        ACE_ERROR_RETURN ((LM_ERROR,
                           "%s usage: \n"
                           "\t-b <bands file> (defaults to %s)\n"
                           "\t-i <iterations> (defaults to %d)\n"
                           "\t-k <ior> (defaults to %s)\n"
                           "\t-p <protocol file> (defaults to %s)\n"
                           "\t-t <invocation priorities file> (defaults to %s)\n"
                           "\t-x <shutdown server> (defaults to %d)\n"
                           "\n",
                           argv[0],
                           bands_file,
                           iterations,
                           ior,
                           protocols_file,
                           invocation_priorities_file,
                           shutdown_server),
                          -1);
      }

  return 0;
}

class Client
{

public:

  Client (test_ptr test,
          CORBA::ORB_ptr orb,
          RTCORBA::Current_ptr current,
          RTCORBA::RTORB_ptr rt_orb,
          CORBA::PolicyManager_ptr policy_manager);

  void vanilla_invocations (ACE_ENV_SINGLE_ARG_DECL);

  void priority_invocations (int debug
                             ACE_ENV_ARG_DECL);

  void set_client_protocols_policies (int debug
                                      ACE_ENV_ARG_DECL);

  void set_priority_bands (int debug
                           ACE_ENV_ARG_DECL);

  void set_private_connection_policies (ACE_ENV_SINGLE_ARG_DECL);

  void reset_policies (ACE_ENV_SINGLE_ARG_DECL);


private:

  test_var test_;
  CORBA::ORB_var orb_;
  RTCORBA::Current_var current_;
  RTCORBA::RTORB_var rt_orb_;
  CORBA::PolicyManager_var policy_manager_;

};

class Worker_Thread : public ACE_Task_Base
{
public:
  Worker_Thread (ACE_Thread_Manager &thread_manager,
                 Client &client,
                 test_ptr test,
                 RTCORBA::Current_ptr current,
                 CORBA::Short priority);

  int svc (void);

  void validate_connection (ACE_ENV_SINGLE_ARG_DECL);

private:
  Client client_;
  test_var test_;
  RTCORBA::Current_var current_;
  CORBA::Short priority_;
};

Worker_Thread::Worker_Thread (ACE_Thread_Manager &thread_manager,
                              Client &client,
                              test_ptr test,
                              RTCORBA::Current_ptr current,
                              CORBA::Short priority)
  : ACE_Task_Base (&thread_manager),
    client_ (client),
    test_ (test::_duplicate (test)),
    current_ (RTCORBA::Current::_duplicate (current)),
    priority_ (priority)
{
}

void
Worker_Thread::validate_connection (ACE_ENV_SINGLE_ARG_DECL)
{
  // Try to validate the connection several times, ignoring transient
  // exceptions.  If the connection can still not be setup, return
  // failure.
  CORBA::PolicyList_var inconsistent_policies;
  int max_attempts = 10;
  int current_attempt = 0;
  for (;;)
    {
      ACE_TRY
        {
          ++current_attempt;
          this->test_->_validate_connection (inconsistent_policies.out ()
                                             ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;

          // If successful, we are done.
          return;
        }
      ACE_CATCH (CORBA::TRANSIENT, exception)
        {
          // If we have reach our maximum number of tries, throw exception.
          if (current_attempt == max_attempts)
            ACE_RE_THROW;
          // Otherwise, ignore...
        }
      ACE_CATCHANY
        {
          // Rethrow any other exceptions.
          ACE_RE_THROW;
        }
      ACE_ENDTRY;
      ACE_CHECK;
    }
}

int
Worker_Thread::svc (void)
{
  ACE_TRY_NEW_ENV
    {
      this->current_->the_priority (this->priority_
                                    ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      this->validate_connection (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      this->client_.vanilla_invocations (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,
                           "Worker Thread exception:");
    }
  ACE_ENDTRY;
  return 0;
}

Client::Client (test_ptr test,
                CORBA::ORB_ptr orb,
                RTCORBA::Current_ptr current,
                RTCORBA::RTORB_ptr rt_orb,
                CORBA::PolicyManager_ptr policy_manager)
  : test_ (test::_duplicate (test)),
    orb_ (CORBA::ORB::_duplicate (orb)),
    current_ (RTCORBA::Current::_duplicate (current)),
    rt_orb_ (RTCORBA::RTORB::_duplicate (rt_orb)),
    policy_manager_ (CORBA::PolicyManager::_duplicate (policy_manager))
{
}

void
Client::vanilla_invocations (ACE_ENV_SINGLE_ARG_DECL)
{
  for (int i = 0; i < iterations; i++)
    {
      this->test_->method (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
Client::priority_invocations (int debug
                              ACE_ENV_ARG_DECL)
{
  ULong_Array priorities;
  int result =
    get_values ("client",
                invocation_priorities_file,
                "invocation priorities",
                priorities,
                debug);
  if (result != 0)
    {
      ACE_ERROR ((LM_ERROR,
                  "Error in parsing invocation priorities data file: %s\n",
                  invocation_priorities_file));
      return;
    }

  u_long i = 0;

  Worker_Thread **workers = 0;

  ACE_NEW_THROW_EX (workers,
                    Worker_Thread *[priorities.size ()],
                    CORBA::NO_MEMORY ());

  ACE_Thread_Manager thread_manager;

  for (i = 0;
       i < priorities.size ();
       ++i)
    {
      ACE_NEW_THROW_EX (workers[i],
                        Worker_Thread (thread_manager,
                                       *this,
                                       this->test_.in (),
                                       this->current_.in (),
                                       priorities[i]),
                        CORBA::NO_MEMORY ());

      long flags =
        THR_NEW_LWP |
        THR_JOINABLE |
        this->orb_->orb_core ()->orb_params ()->thread_creation_flags ();

      result =
        workers[i]->activate (flags);
      if (result != 0)
        {
          ACE_ERROR ((LM_ERROR,
                      "Cannot activate thread\n"));
          return;
        }
    }

  thread_manager.wait ();

  for (i = 0;
       i < priorities.size ();
       ++i)
    {
      delete workers[i];
    }
  delete[] workers;