connection_manager.cpp

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

            this->sender_context_->resolve (name
                                            ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK_EX(SENDER_CONTEXT_EXISTS);

          this->sender_ =
            AVStreams::MMDevice::_narrow (object.in () ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK_EX(SENDER_CONTEXT_EXISTS);
        }
      ACE_CATCH (CosNaming::NamingContext::NotFound, al_ex)
        {
          // No problem if the sender was not there.
        }
      ACE_ENDTRY;
      ACE_CHECK;
    }
}

void
Connection_Manager::connect_to_sender (ACE_ENV_SINGLE_ARG_DECL)
{
  if (CORBA::is_nil (this->sender_.in ()))
    return;

  ACE_CString flowname =
    this->sender_name_ +
    "_" +
    this->receiver_name_;

  // Create the forward flow specification to describe the flow.
  TAO_Forward_FlowSpec_Entry sender_entry (flowname.c_str (),
                                           "IN",
                                           "USER_DEFINED",
                                           "",
                                           "UDP",
                                           0);

  // Set the flow specification for the stream between sender and
  // receiver.
  AVStreams::flowSpec flow_spec (1);
  flow_spec.length (1);
  flow_spec [0] =
    CORBA::string_dup (sender_entry.entry_to_string ());

  // Create the stream control for this stream
  TAO_StreamCtrl* streamctrl;
  ACE_NEW (streamctrl,
           TAO_StreamCtrl);

  // Servant Reference Counting to manage lifetime
  PortableServer::ServantBase_var safe_streamctrl =
    streamctrl;

  // Register streamctrl.
  AVStreams::StreamCtrl_var streamctrl_object =
    streamctrl->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  //
  // Since senders terminate the streams, we don't need the streamctrl
  // for these.
  //
  // this->streamctrls_.bind (flowname,
  //                          streamctrl_object);

  // Initialize the  QoS
  AVStreams::streamQoS_var the_qos (new AVStreams::streamQoS);

  // Connect the sender and receiver devices.
  CORBA::Boolean result =
    streamctrl->bind_devs (this->sender_.in (),
                           this->receiver_.in (),
                           the_qos.inout (),
                           flow_spec
                           ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  if (result == 0)
    ACE_ERROR ((LM_ERROR,
                "Streamctrl::bind_devs failed\n"));

  // Start the data sending.
  AVStreams::flowSpec start_spec;
  streamctrl->start (start_spec
                     ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
Connection_Manager::add_streamctrl (const ACE_CString &flowname,
                                    TAO_StreamEndPoint *endpoint
                                    ACE_ENV_ARG_DECL)
{
  // Get the stream controller for this endpoint.
  CORBA::Any_var streamctrl_any =
    endpoint->get_property_value ("Related_StreamCtrl"
                                  ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  AVStreams::StreamCtrl_ptr streamctrl;

  if( streamctrl_any.in() >>= streamctrl )
  {
    // the CORBA::Any_var owns the pointer, so we should
    // _duplicate it before passing it around
    AVStreams::StreamCtrl::_duplicate( streamctrl );
    this->streamctrls_.unbind(flowname);
    this->streamctrls_.bind (flowname,
                             streamctrl);
  }
}


void
Connection_Manager::destroy (const ACE_CString &flowname
                             ACE_ENV_ARG_DECL)
{
  this->protocol_objects_.unbind (flowname);
  this->receivers_.unbind (flowname);

  this->streamctrls_.unbind(flowname);
  ACE_CHECK;
}

void
Connection_Manager::unbind_sender (const ACE_CString &sender_name,
                                   AVStreams::MMDevice_ptr sender_mmdevice_obj
                                   ACE_ENV_ARG_DECL)
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "Connection_Manager::unbind_sender\n"));

  CosNaming::Name name (1);
  name.length (1);

  // Try binding the sender context in the NS
  name [0].id =
    CORBA::string_dup (sender_name.c_str ());

  ACE_TRY
    {
      CORBA::Object_var object =
        this->naming_client_->resolve (name
                                       ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      CosNaming::NamingContext_var sender_context =
        CosNaming::NamingContext::_narrow (object.in ());

      if (TAO_debug_level > 0)
        ACE_DEBUG ((LM_DEBUG,
                    "Sender Context Found\n"));

      ACE_TRY_EX (Resolve_Sender)
        {
          object =
            sender_context->resolve (name
                                     ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK_EX (Resolve_Sender);

          AVStreams::MMDevice_var mmdevice =
            AVStreams::MMDevice::_narrow (object.in ());

          if (mmdevice->_is_equivalent (sender_mmdevice_obj))
          {
            sender_context->unbind (name
                                    ACE_ENV_ARG_PARAMETER);
            ACE_TRY_CHECK_EX (Resolve_Sender);

            if (TAO_debug_level > 0)
              ACE_DEBUG ((LM_DEBUG,
                          "Sender Unbound\n"));
          }
        }
      ACE_CATCH (CosNaming::NamingContext::NotFound, al_ex)
        {
          ACE_DEBUG ((LM_DEBUG,
                      "Sender object not found\n"));
          // Do not have to unbind.
        }
      ACE_ENDTRY;

    }
  ACE_CATCH (CosNaming::NamingContext::NotFound, al_ex)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "Sender Context Not Found\n"));
      // Do not have to unbind.
    }
  ACE_ENDTRY;

}

void
Connection_Manager::unbind_receiver (const ACE_CString &sender_name,
                                     const ACE_CString &receiver_name,
                                     AVStreams::MMDevice_ptr receiver_mmdevice)
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "Connection_Manager::unbind_receiver\n"));

  CosNaming::Name name (1);
  name.length (1);

  // Try binding the sender context in the NS
  name [0].id =
    CORBA::string_dup (sender_name.c_str ());

  ACE_TRY_NEW_ENV
    {
      CORBA::Object_var object =
        this->naming_client_->resolve (name
                                       ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      CosNaming::NamingContext_var sender_context =
        CosNaming::NamingContext::_narrow (object.in ());

      if (TAO_debug_level > 0)
        ACE_DEBUG ((LM_DEBUG,
                    "Sender Context Found\n"));

      ACE_TRY_EX (Resolve_Receiver)
        {
          name [0].id =
            CORBA::string_dup ("Receivers");

          object =
            sender_context->resolve (name
                                     ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK_EX (Resolve_Receiver);

          if (TAO_debug_level > 0)
            ACE_DEBUG ((LM_DEBUG,
                        "Receivers Context Found\n"));

          CosNaming::NamingContext_var receivers_context =
            CosNaming::NamingContext::_narrow (object.in ());

          name [0].id =
            CORBA::string_dup (receiver_name.c_str ());

          object =
            receivers_context->resolve (name
                                        ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK_EX (Resolve_Receiver);

          AVStreams::MMDevice_var  mmdevice =
            AVStreams::MMDevice::_narrow (object.in ());

          if (mmdevice->_is_equivalent (receiver_mmdevice))
          {
            receivers_context->unbind (name
                                       ACE_ENV_ARG_PARAMETER);
            ACE_TRY_CHECK_EX (Resolve_Receiver);

            if (TAO_debug_level > 0)
              ACE_DEBUG ((LM_DEBUG,
                          "Receiver Unbound\n"));
          }
        }
      ACE_CATCH (CosNaming::NamingContext::NotFound, al_ex)
        {
          ACE_DEBUG ((LM_DEBUG,
                      "Receiver Not found\n"));
          // Do not have to unbind.
        }
      ACE_ENDTRY;

    }
  ACE_CATCH (CosNaming::NamingContext::NotFound, al_ex)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "Sender Context Not Found\n"));
      // Do not have to unbind.
    }
  ACE_ENDTRY;

}

Connection_Manager::Receivers &
Connection_Manager::receivers (void)
{
  return this->receivers_;
}

Connection_Manager::Protocol_Objects &
Connection_Manager::protocol_objects (void)
{
  return this->protocol_objects_;
}

Connection_Manager::StreamCtrls &
Connection_Manager::streamctrls (void)
{
  return this->streamctrls_;
}

#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)

template class ACE_Hash_Map_Entry<ACE_CString, AVStreams::MMDevice_var>;
template class ACE_Hash_Map_Manager<ACE_CString, AVStreams::MMDevice_var, ACE_Null_Mutex>;
template class ACE_Hash_Map_Manager_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Base_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Reverse_Iterator_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;

template class ACE_Hash_Map_Entry<ACE_CString, TAO_AV_Protocol_Object *>;
template class ACE_Hash_Map_Manager<ACE_CString, TAO_AV_Protocol_Object *, ACE_Null_Mutex>;
template class ACE_Hash_Map_Manager_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Base_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Reverse_Iterator_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;

template class ACE_Hash_Map_Entry<ACE_CString, AVStreams::StreamCtrl_var>;
template class ACE_Hash_Map_Manager<ACE_CString, AVStreams::StreamCtrl_var, ACE_Null_Mutex>;
template class ACE_Hash_Map_Manager_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Base_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Reverse_Iterator_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>;

#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)

#pragma instantiate ACE_Hash_Map_Entry<ACE_CString, AVStreams::MMDevice_var>
#pragma instantiate ACE_Hash_Map_Manager<ACE_CString, AVStreams::MMDevice_var, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Manager_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Base_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Reverse_Iterator_Ex<ACE_CString, AVStreams::MMDevice_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>

#pragma instantiate ACE_Hash_Map_Entry<ACE_CString, TAO_AV_Protocol_Object *>
#pragma instantiate ACE_Hash_Map_Manager<ACE_CString, TAO_AV_Protocol_Object *, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Manager_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Base_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Reverse_Iterator_Ex<ACE_CString, TAO_AV_Protocol_Object *, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>

#pragma instantiate ACE_Hash_Map_Entry<ACE_CString, AVStreams::StreamCtrl_var>
#pragma instantiate ACE_Hash_Map_Manager<ACE_CString, AVStreams::StreamCtrl_var, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Manager_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Base_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Reverse_Iterator_Ex<ACE_CString, AVStreams::StreamCtrl_var, ACE_Hash<ACE_CString>, ACE_Equal_To<ACE_CString>, ACE_Null_Mutex>

#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */