summaryrefslogtreecommitdiff
path: root/TAO/orbsvcs/orbsvcs/Notify/CosNotify_Service.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'TAO/orbsvcs/orbsvcs/Notify/CosNotify_Service.cpp')
-rw-r--r--TAO/orbsvcs/orbsvcs/Notify/CosNotify_Service.cpp402
1 files changed, 402 insertions, 0 deletions
diff --git a/TAO/orbsvcs/orbsvcs/Notify/CosNotify_Service.cpp b/TAO/orbsvcs/orbsvcs/Notify/CosNotify_Service.cpp
new file mode 100644
index 00000000000..3e095e32f73
--- /dev/null
+++ b/TAO/orbsvcs/orbsvcs/Notify/CosNotify_Service.cpp
@@ -0,0 +1,402 @@
+// $Id$
+
+#include "orbsvcs/Notify/CosNotify_Service.h"
+#include "orbsvcs/Notify/Properties.h"
+#include "orbsvcs/Notify/Default_Factory.h"
+#include "orbsvcs/Notify/Builder.h"
+#include "ace/Sched_Params.h"
+#include "ace/Arg_Shifter.h"
+#include "ace/Dynamic_Service.h"
+#include "tao/ORB_Core.h"
+#include "orbsvcs/NotifyExtC.h"
+#include "tao/debug.h"
+
+ACE_RCSID (Notify,
+ TAO_CosNotify_Service,
+ "$Id$")
+
+TAO_BEGIN_VERSIONED_NAMESPACE_DECL
+
+TAO_CosNotify_Service::TAO_CosNotify_Service (void)
+{
+}
+
+TAO_CosNotify_Service::~TAO_CosNotify_Service ()
+{
+}
+
+int
+TAO_CosNotify_Service::init (int argc, ACE_TCHAR *argv[])
+{
+ ACE_Arg_Shifter arg_shifter (argc, argv);
+
+ const ACE_TCHAR *current_arg = 0;
+
+ // Default to an all reactive system.
+ int ec_threads = 0;
+ int consumer_threads = 0;
+ int supplier_threads = 0;
+
+ int task_per_proxy = 0;
+
+ TAO_Notify_Properties *properties = TAO_Notify_PROPERTIES::instance();
+
+ while (arg_shifter.is_anything_left ())
+ {
+ if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-MTDispatching")) == 0)
+ {
+ arg_shifter.consume_arg ();
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%P|%t) The -MTDispatching option has been deprecated, use -DispatchingThreads \n")));
+ }
+ else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-DispatchingThreads"))))
+ {
+ consumer_threads += ACE_OS::atoi (current_arg);
+ arg_shifter.consume_arg ();
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-MTSourceEval")) == 0)
+ {
+ arg_shifter.consume_arg ();
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%P|%t) The -MTSourceEval option has been deprecated, use -SourceThreads \n")));
+ }
+ else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-SourceThreads"))))
+ {
+ supplier_threads += ACE_OS::atoi (current_arg);
+ arg_shifter.consume_arg ();
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-MTLookup")) == 0)
+ {
+ arg_shifter.consume_arg ();
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%P|%t) The -MTLookup option has been deprecated, use -SourceThreads \n")));
+ }
+ else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-LookupThreads"))))
+ {
+ supplier_threads += ACE_OS::atoi (current_arg);
+ arg_shifter.consume_arg ();
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%P|%t) The -LookupThreads option has been deprecated, use -SourceThreads \n")));
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-MTListenerEval")) == 0)
+ {
+ arg_shifter.consume_arg ();
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%P|%t) The -MTListenerEval option has been deprecated, use -DispatchingThreads \n")));
+ }
+ else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-ListenerThreads"))))
+ {
+ // Since this option is always added to consumer_threads, we'll
+ // deprecate it in favor of that option.
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%P|%t) The -ListenerThreads option has been deprecated, use -DispatchingThreads \n")));
+ consumer_threads += ACE_OS::atoi (current_arg);
+ arg_shifter.consume_arg ();
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-AsynchUpdates")) == 0)
+ {
+ arg_shifter.consume_arg ();
+
+ properties->asynch_updates (1);
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-NoUpdates")) == 0)
+ {
+ arg_shifter.consume_arg ();
+
+ properties->updates (0);
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-AllocateTaskperProxy")) == 0)
+ {
+ task_per_proxy = 1;
+ arg_shifter.consume_arg ();
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-UseSeparateDispatchingORB")) == 0)
+ {
+ current_arg = arg_shifter.get_the_parameter
+ (ACE_TEXT("-UseSeparateDispatchingORB"));
+ if (current_arg != 0 &&
+ (ACE_OS::strcmp(ACE_TEXT ("0"), current_arg) == 0 ||
+ ACE_OS::strcmp(ACE_TEXT ("1"), current_arg) == 0))
+ {
+ properties->separate_dispatching_orb (
+ static_cast<bool> (ACE_OS::atoi(current_arg)));
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("Using separate Dispatching ORB\n")));
+ }
+ else
+ {
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%P|%t) WARNING: Unrecognized ")
+ ACE_TEXT ("argument (%s). Ignoring invalid ")
+ ACE_TEXT ("-UseSeparateDispatchingORB usage.\n"),
+ (current_arg == 0 ? ACE_TEXT ("''") : current_arg)));
+ }
+ if (current_arg != 0)
+ arg_shifter.consume_arg ();
+ }
+ else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-AllowReconnect")) == 0)
+ {
+ arg_shifter.consume_arg ();
+ TAO_Notify_PROPERTIES::instance()->allow_reconnect (true);
+ }
+ else
+ {
+ ACE_ERROR ((LM_ERROR,
+ ACE_TEXT ("(%P|%t) Ignoring unknown option for Notify Factory: %s\n"),
+ arg_shifter.get_current()
+ ));
+ arg_shifter.consume_arg ();
+ }
+ }
+
+ // Init the EC QoS
+ {
+ CosNotification::QoSProperties qos;
+ this->set_threads (qos, ec_threads);
+ properties->default_event_channel_qos_properties (qos);
+ }
+
+ if (task_per_proxy == 0)
+ {
+ // Set the per ConsumerAdmin QoS
+ {
+ if (consumer_threads > 0)
+ ACE_DEBUG((LM_DEBUG, "Using %d threads for each ConsumerAdmin.\n", consumer_threads));
+ CosNotification::QoSProperties qos;
+ this->set_threads (qos, consumer_threads);
+ properties->default_consumer_admin_qos_properties (qos);
+ }
+
+ // Set the per SupplierAdmin QoS
+ {
+ if (supplier_threads > 0)
+ ACE_DEBUG((LM_DEBUG, "Using %d threads for each SupplierAdmin.\n", supplier_threads));
+ CosNotification::QoSProperties qos;
+ this->set_threads (qos, supplier_threads);
+ properties->default_supplier_admin_qos_properties (qos);
+ }
+ }
+ else
+ {
+ // Set the per ProxyConsumer QoS
+ {
+ if (supplier_threads > 0)
+ ACE_DEBUG((LM_DEBUG, "Using %d threads for each Supplier.\n", supplier_threads));
+ CosNotification::QoSProperties qos;
+ this->set_threads (qos, supplier_threads); // lookup thread per proxy doesn't make sense.
+ properties->default_proxy_consumer_qos_properties (qos);
+ }
+
+ // Set the per ProxySupplier QoS
+ {
+ if (consumer_threads > 0)
+ ACE_DEBUG((LM_DEBUG, "Using %d threads for each Consumer.\n", consumer_threads));
+ CosNotification::QoSProperties qos;
+ this->set_threads (qos, consumer_threads);
+ properties->default_proxy_supplier_qos_properties (qos);
+ }
+ }
+
+ return 0;
+}
+
+void
+TAO_CosNotify_Service::set_threads (CosNotification::QoSProperties &qos, int threads)
+{
+ NotifyExt::ThreadPoolParams tp_params =
+ {NotifyExt::CLIENT_PROPAGATED, 0, 0, (unsigned)threads, 0, 0, 0, 0, 0 };
+
+ qos.length (1);
+ qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
+ qos[0].value <<= tp_params;
+}
+
+int
+TAO_CosNotify_Service::fini (void)
+{
+ return 0;
+}
+
+void
+TAO_CosNotify_Service::init_service (CORBA::ORB_ptr orb ACE_ENV_ARG_DECL)
+{
+ ACE_DEBUG ((LM_DEBUG, "Loading the Cos Notification Service...\n"));
+
+ if (TAO_Notify_PROPERTIES::instance()->separate_dispatching_orb())
+ {
+ // got here by way of svc.conf. no second orb supplied so create one
+ if (NULL == TAO_Notify_PROPERTIES::instance()->dispatching_orb())
+ {
+ ACE_DEBUG ((LM_DEBUG, "No dispatching orb supplied. Creating default one.\n"));
+
+ int argc = 0;
+ char *argv0 = 0;
+ char **argv = &argv0; // ansi requires argv be null terminated.
+ CORBA::ORB_var dispatcher = CORBA::ORB_init (argc, argv,
+ "default_dispatcher" ACE_ENV_ARG_PARAMETER);
+ //ACE_CHECK_RETURN (-1);
+
+ TAO_Notify_PROPERTIES::instance()->dispatching_orb(dispatcher.in());
+ }
+
+ this->init_i2 (orb, TAO_Notify_PROPERTIES::instance()->dispatching_orb() ACE_ENV_ARG_PARAMETER);
+
+ }
+ else
+ {
+ this->init_i (orb ACE_ENV_ARG_PARAMETER);
+ }
+}
+
+void
+TAO_CosNotify_Service::init_service2 (CORBA::ORB_ptr orb, CORBA::ORB_ptr dispatching_orb ACE_ENV_ARG_DECL)
+{
+ this->init_i2 (orb, dispatching_orb ACE_ENV_ARG_PARAMETER);
+ ACE_CHECK;
+}
+
+void
+TAO_CosNotify_Service::init_i (CORBA::ORB_ptr orb ACE_ENV_ARG_DECL)
+{
+ // Obtain the Root POA
+ CORBA::Object_var object =
+ orb->resolve_initial_references("RootPOA" ACE_ENV_ARG_PARAMETER);
+ ACE_CHECK;
+
+ if (CORBA::is_nil (object.in ()))
+ ACE_ERROR ((LM_ERROR,
+ " (%P|%t) Unable to resolve the RootPOA.\n"));
+
+ PortableServer::POA_var default_poa = PortableServer::POA::_narrow (object.in () ACE_ENV_ARG_PARAMETER);
+ ACE_CHECK;
+
+ // Set the properties
+ TAO_Notify_Properties* properties = TAO_Notify_PROPERTIES::instance();
+
+ properties->orb (orb);
+ properties->default_poa (default_poa.in ());
+
+ // Init the factory
+ this->factory_.reset (this->create_factory (ACE_ENV_SINGLE_ARG_PARAMETER));
+ ACE_CHECK;
+ ACE_ASSERT( this->factory_.get() != 0 );
+ TAO_Notify_PROPERTIES::instance()->factory (this->factory_.get());
+
+ this->builder_.reset (this->create_builder (ACE_ENV_SINGLE_ARG_PARAMETER));
+ ACE_CHECK;
+ ACE_ASSERT( this->builder_.get() != 0 );
+ TAO_Notify_PROPERTIES::instance()->builder (this->builder_.get());
+}
+
+void
+TAO_CosNotify_Service::init_i2 (CORBA::ORB_ptr orb, CORBA::ORB_ptr dispatching_orb ACE_ENV_ARG_DECL)
+{
+ // Obtain the Root POA
+ CORBA::Object_var object =
+ orb->resolve_initial_references("RootPOA" ACE_ENV_ARG_PARAMETER);
+ ACE_CHECK;
+
+ if (CORBA::is_nil (object.in ()))
+ ACE_ERROR ((LM_ERROR, " (%P|%t) Unable to resolve the RootPOA.\n"));
+
+ PortableServer::POA_var default_poa = PortableServer::POA::_narrow (object.in () ACE_ENV_ARG_PARAMETER);
+ ACE_CHECK;
+
+ // Set the properties
+ TAO_Notify_Properties* properties = TAO_Notify_PROPERTIES::instance();
+
+ properties->orb (orb);
+ properties->dispatching_orb (dispatching_orb);
+ properties->separate_dispatching_orb (true);
+
+ properties->default_poa (default_poa.in ());
+
+ // Init the factory and builder
+ this->factory_.reset (this->create_factory (ACE_ENV_SINGLE_ARG_PARAMETER));
+ ACE_CHECK;
+ ACE_ASSERT( this->factory_.get() != 0 );
+ TAO_Notify_PROPERTIES::instance()->factory (this->factory_.get());
+
+ this->builder_.reset (this->create_builder (ACE_ENV_SINGLE_ARG_PARAMETER));
+ ACE_CHECK;
+ ACE_ASSERT( this->builder_.get() != 0 );
+ TAO_Notify_PROPERTIES::instance()->builder (this->builder_.get());
+}
+
+TAO_Notify_Factory*
+TAO_CosNotify_Service::create_factory (ACE_ENV_SINGLE_ARG_DECL)
+{
+ TAO_Notify_Factory* factory = ACE_Dynamic_Service<TAO_Notify_Factory>::instance ("TAO_Notify_Factory");
+ if (factory == 0)
+ {
+ ACE_NEW_THROW_EX (factory,
+ TAO_Notify_Default_Factory (),
+ CORBA::NO_MEMORY ());
+ ACE_CHECK_RETURN (0);
+ }
+ return factory;
+}
+
+TAO_Notify_Builder*
+TAO_CosNotify_Service::create_builder (ACE_ENV_SINGLE_ARG_DECL)
+{
+ TAO_Notify_Builder* builder = 0;
+ ACE_NEW_THROW_EX (builder,
+ TAO_Notify_Builder (),
+ CORBA::NO_MEMORY ());
+ ACE_CHECK_RETURN (0);
+
+ return builder;
+}
+
+CosNotifyChannelAdmin::EventChannelFactory_ptr
+TAO_CosNotify_Service::create (PortableServer::POA_ptr poa ACE_ENV_ARG_DECL)
+{
+ return this->builder().build_event_channel_factory (poa ACE_ENV_ARG_PARAMETER);
+}
+
+void
+TAO_CosNotify_Service::remove (TAO_Notify_EventChannelFactory* /*ecf*/ ACE_ENV_ARG_DECL_NOT_USED)
+{
+ // NOP.
+}
+
+TAO_Notify_Factory&
+TAO_CosNotify_Service::factory (void)
+{
+ ACE_ASSERT( this->factory_.get() != 0 );
+ return *this->factory_;
+}
+
+TAO_Notify_Builder&
+TAO_CosNotify_Service::builder (void)
+{
+ ACE_ASSERT( this->builder_.get() != 0 );
+ return *this->builder_;
+}
+
+TAO_END_VERSIONED_NAMESPACE_DECL
+
+
+/*********************************************************************************************************************/
+
+ACE_STATIC_SVC_DEFINE (TAO_Notify_Default_EMO_Factory_OLD,
+ ACE_TEXT (TAO_NOTIFY_DEF_EMO_FACTORY_NAME),
+ ACE_SVC_OBJ_T,
+ &ACE_SVC_NAME (TAO_CosNotify_Service),
+ ACE_Service_Type::DELETE_THIS | ACE_Service_Type::DELETE_OBJ,
+ 0)
+
+/*********************************************************************************************************************/
+
+ACE_STATIC_SVC_DEFINE (TAO_CosNotify_Service,
+ ACE_TEXT (TAO_COS_NOTIFICATION_SERVICE_NAME),
+ ACE_SVC_OBJ_T,
+ &ACE_SVC_NAME (TAO_CosNotify_Service),
+ ACE_Service_Type::DELETE_THIS | ACE_Service_Type::DELETE_OBJ,
+ 0)
+
+
+ACE_FACTORY_DEFINE (TAO_Notify_Serv, TAO_CosNotify_Service)
+
+/*********************************************************************************************************************/