request_interceptor.cpp

来自「这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用」· C++ 代码 · 共 506 行 · 第 1/2 页

//Request_Interceptor.cpp,v 1.14 2003/11/04 05:21:38 dhinton Exp

#include "Request_Interceptor.h"
#include "Current.h"
#include "Distributable_Thread.h"
#include "tao/TSS_Resources.h"
#include "tao/debug.h"
#include "tao/Any.h"
#include "tao/Typecode.h"
#include "tao/ORB_Constants.h"
#include "ace/OS_NS_string.h"

ACE_RCSID (RTScheduling,
           Request_Interceptor,
           "Request_Interceptor.cpp,v 1.14 2003/11/04 05:21:38 dhinton Exp")


const IOP::ServiceId
Client_Interceptor::SchedulingInfo = 30;

Client_Interceptor::Client_Interceptor (void)
{
}

void
Client_Interceptor::send_request (PortableInterceptor::ClientRequestInfo_ptr ri
                                  ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   PortableInterceptor::ForwardRequest))
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "Client_Interceptor::send_request\n"));

  // Temporary current.
  TAO_RTScheduler_Current_i *new_current = 0;
  TAO_RTScheduler_Current_i *current = 0;

  TAO_TSS_Resources *tss = TAO_TSS_RESOURCES::instance ();

  current = ACE_static_cast (TAO_RTScheduler_Current_i *,
                                  tss->rtscheduler_current_impl_);

  if (current != 0)
    {
      // If this is a one way request
      if (!ri->response_expected ())
        {

          // Generate GUID.
          RTScheduling::Current::IdType guid;
          guid.length (sizeof(long));

          long temp = ++TAO_RTScheduler_Current::guid_counter;
          ACE_OS::memcpy (guid.get_buffer (),
                          &temp,
                          sizeof(long));

          int id;
          ACE_OS::memcpy (&id,
                          guid.get_buffer (),
                          guid.length ());

          if (TAO_debug_level > 0)
            ACE_DEBUG ((LM_DEBUG,
                        "The Guid is %d %d\n",
                        id,
                        TAO_RTScheduler_Current::guid_counter.value_i ()));

          // Create new DT.
          RTScheduling::DistributableThread_var dt =
            TAO_DistributableThread_Factory::create_DT ();

          // Add new DT to map.
          int result = current->dt_hash ()->bind (guid, dt);
          if (result != 0)
            {
              ACE_DEBUG ((LM_DEBUG,
                          "No Scheduling Segment Context\n"));
              ACE_THROW (CORBA::INTERNAL ());

            }

          // @@ Store implicit_sched_param in a var

          // Create new temporary current. Note that
          // the new <sched_param> is the current
          // <implicit_sched_param> and there is no
          // segment name.
          ACE_NEW (new_current,
                   TAO_RTScheduler_Current_i (current->orb (),
                                              current->dt_hash (),
                                              guid,
                                              0,
                                              current->implicit_scheduling_parameter (ACE_ENV_SINGLE_ARG_PARAMETER),
                                              0,
                                              dt.in (),
                                              current));

          // Install new current in the ORB.
          //current->implementation (new_current);
          tss->rtscheduler_current_impl_ = new_current;

        }

      // Scheduler populates the service context with
      // scheduling parameters.
      current->scheduler ()->send_request (ri);

      // If this is a one way request
      if (!ri->response_expected ())
        {
          // Cleanup temporary DT.
          new_current->cleanup_DT ();

          //Restore old current
          new_current->cleanup_current ();
        }
    }
}

void
Client_Interceptor::send_poll (PortableInterceptor::ClientRequestInfo_ptr ri
                               ACE_ENV_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "Client_Interceptor::send_poll\n"));

    TAO_RTScheduler_Current_i *current = 0;

    TAO_TSS_Resources *tss = TAO_TSS_RESOURCES::instance ();

    current = ACE_static_cast (TAO_RTScheduler_Current_i *,
                               tss->rtscheduler_current_impl_);
    if (current != 0)
      current->scheduler ()->send_poll (ri);

}

void
Client_Interceptor::receive_reply (PortableInterceptor::ClientRequestInfo_ptr ri
                                   ACE_ENV_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "Client_Interceptor::receive_reply\n"));

  TAO_RTScheduler_Current_i *current = 0;

  TAO_TSS_Resources *tss = TAO_TSS_RESOURCES::instance ();

  current = ACE_static_cast (TAO_RTScheduler_Current_i *,
                                  tss->rtscheduler_current_impl_);
  if (current != 0)
    current->scheduler ()->receive_reply (ri);
}

void
Client_Interceptor::receive_exception (PortableInterceptor::ClientRequestInfo_ptr ri
                                       ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   PortableInterceptor::ForwardRequest))
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "Client_Interceptor::receive_exception\n"));


  TAO_RTScheduler_Current_i *current = 0;

  TAO_TSS_Resources *tss = TAO_TSS_RESOURCES::instance ();

  current = ACE_static_cast (TAO_RTScheduler_Current_i *,
                             tss->rtscheduler_current_impl_);

  if (current != 0)
    {
      if (ri == 0)
        {
          ACE_ERROR ((LM_ERROR,
                      "ri = 0\n"));
          return;
        }

      CORBA::Any_var ex =
        ri->received_exception (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;
      CORBA::TypeCode_var type = ex->type ();

      if (CORBA::is_nil (type.in ()))
        {
          ACE_ERROR ((LM_ERROR,
                      "type = 0 \n"));
          return;
        }
      const char * id = type->id ();

      if (TAO_debug_level > 0)
        ACE_DEBUG ((LM_DEBUG,
                    "Received Exception %s\n",
                    id));


      // If the remote host threw a THREAD_CANCELLED
      // exception, make sure to take the appropriate
      // local action.
      if (ACE_OS::strstr (id, "CORBA::THREAD_CANCELLED") == 0)
        {
          // Perform the necessary cleanup as the
          // thread was cancelled.
          current->cancel_thread (ACE_ENV_SINGLE_ARG_PARAMETER);
          ACE_CHECK;
        }
      else
        {
          // Inform scheduler that exception was
          // received.
          current->scheduler ()->receive_exception (ri);
        }
    }
}

void
Client_Interceptor::receive_other (PortableInterceptor::ClientRequestInfo_ptr ri
                                   ACE_ENV_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   PortableInterceptor::ForwardRequest))
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "Client_Interceptor::receive_other\n"));

  TAO_RTScheduler_Current_i *current = 0;

  TAO_TSS_Resources *tss = TAO_TSS_RESOURCES::instance ();

  current = ACE_static_cast (TAO_RTScheduler_Current_i *,
                             tss->rtscheduler_current_impl_);
  if (current != 0)
    current->scheduler ()->receive_other (ri);

}

char*
Client_Interceptor::name (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  return  CORBA::string_dup ("RTSchdeuler_Client_Interceptor");
}