summaryrefslogtreecommitdiff
path: root/ACE/examples/ASX/UPIPE_Event_Server
diff options
context:
space:
mode:
Diffstat (limited to 'ACE/examples/ASX/UPIPE_Event_Server')
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp138
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.h53
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.cpp73
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h37
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Makefile.am50
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Options.cpp209
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Options.h88
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Options.inl166
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.cpp283
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.h127
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp137
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.h57
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/UPIPE_Event.mpc15
-rw-r--r--ACE/examples/ASX/UPIPE_Event_Server/event_server.cpp271
14 files changed, 1704 insertions, 0 deletions
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp b/ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp
new file mode 100644
index 00000000000..b9c9c0cf2bd
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp
@@ -0,0 +1,138 @@
+// $Id$
+
+#include "ace/OS_NS_stdio.h"
+#include "ace/OS_NS_string.h"
+#include "Consumer_Router.h"
+#include "Options.h"
+
+ACE_RCSID(UPIPE_Event_Server, Consumer_Router, "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+
+typedef Acceptor_Factory<Consumer_Handler, CONSUMER_KEY> CONSUMER_FACTORY;
+
+int
+Consumer_Handler::open (void *a)
+{
+ CONSUMER_FACTORY *af = (CONSUMER_FACTORY *) a;
+ this->router_task_ = af->router ();
+ return this->Peer_Handler<CONSUMER_ROUTER, CONSUMER_KEY>::open (a);
+}
+
+Consumer_Handler::Consumer_Handler (ACE_Thread_Manager *tm)
+ : Peer_Handler<CONSUMER_ROUTER, CONSUMER_KEY> (tm)
+{
+}
+
+// Create a new handler that will interact with a consumer and point
+// its ROUTER_TASK_ data member to the CONSUMER_ROUTER.
+
+Consumer_Router::Consumer_Router (ACE_Thread_Manager *tm)
+ : CONSUMER_ROUTER (tm)
+{
+}
+
+// Initialize the Router..
+
+int
+Consumer_Router::open (void *)
+{
+ ACE_ASSERT (this->is_reader ());
+ ACE_TCHAR *argv[3];
+
+ argv[0] = (ACE_TCHAR *) this->name ();
+ argv[1] = (ACE_TCHAR *) options.consumer_file ();
+ argv[2] = 0;
+
+ if (this->init (1, &argv[1]) == -1)
+ return -1;
+
+ // Make this an active object.
+ // return this->activate (options.t_flags ());
+
+ // Until that's done, return 1 to indicate that the object wasn't activated.
+ return 1;
+}
+
+int
+Consumer_Router::close (u_long)
+{
+ ACE_ASSERT (this->is_reader ());
+ this->peer_map_.close ();
+ this->msg_queue ()->deactivate();
+ return 0;
+}
+
+
+// Handle incoming messages in a separate thread..
+
+int
+Consumer_Router::svc (void)
+{
+ ACE_Message_Block *mb = 0;
+
+ ACE_ASSERT (this->is_reader ());
+
+ if (options.debug ())
+ ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) starting svc in %s\n"),
+ this->name ()));
+
+ while (this->getq (mb) > 0)
+ if (this->put_next (mb) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("(%t) put_next failed in %s\n"),
+ this->name ()), -1);
+
+ return 0;
+ // Note the implicit ACE_OS::thr_exit() via destructor.
+}
+
+// Send a MESSAGE_BLOCK to the supplier(s)..
+
+int
+Consumer_Router::put (ACE_Message_Block *mb, ACE_Time_Value *)
+{
+ ACE_ASSERT (this->is_reader ());
+
+ if (mb->msg_type () == ACE_Message_Block::MB_IOCTL)
+ {
+ this->control (mb);
+ return this->put_next (mb);
+ }
+ else
+{
+//printf("consumer-Router is routing : send_peers\n");
+ return this->send_peers (mb);
+}
+}
+
+// Return information about the Client_Router ACE_Module..
+
+int
+Consumer_Router::info (ACE_TCHAR **strp, size_t length) const
+{
+ ACE_TCHAR buf[BUFSIZ];
+ ACE_UPIPE_Addr addr;
+ const ACE_TCHAR *mod_name = this->name ();
+ ACE_UPIPE_Acceptor &sa = (ACE_UPIPE_Acceptor &) *this->acceptor_;
+
+ if (sa.get_local_addr (addr) == -1)
+ return -1;
+
+#if !defined (ACE_WIN32) && defined (ACE_USES_WCHAR)
+# define FMTSTR ACE_TEXT ("%ls\t %ls/ %ls")
+#else
+# define FMTSTR ACE_TEXT ("%s\t %s/ %s")
+#endif
+
+ ACE_OS::sprintf (buf, FMTSTR,
+ mod_name, ACE_TEXT ("upipe"),
+ ACE_TEXT ("# consumer router\n"));
+
+ if (*strp == 0 && (*strp = ACE_OS::strdup (mod_name)) == 0)
+ return -1;
+ else
+ ACE_OS::strncpy (*strp, mod_name, length);
+ return ACE_OS::strlen (mod_name);
+}
+
+#endif /* ACE_HAS_THREADS */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.h b/ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.h
new file mode 100644
index 00000000000..93a1220dc11
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.h
@@ -0,0 +1,53 @@
+/* -*- C++ -*- */
+// $Id$
+
+// The interface between one or more consumers and an Event Server
+// ACE_Stream.
+
+#ifndef _CONSUMER_ROUTER_H
+#define _CONSUMER_ROUTER_H
+
+#include "ace/Thread_Manager.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "ace/UPIPE_Acceptor.h"
+#include "ace/UPIPE_Addr.h"
+#include "ace/Svc_Handler.h"
+#include "Peer_Router.h"
+
+#if defined (ACE_HAS_THREADS)
+
+class Consumer_Handler; // Forward declaration....
+
+typedef ACE_HANDLE CONSUMER_KEY;
+
+typedef Peer_Router<Consumer_Handler, CONSUMER_KEY> CONSUMER_ROUTER;
+
+class Consumer_Handler
+ : public Peer_Handler<CONSUMER_ROUTER, CONSUMER_KEY>
+{
+public:
+ Consumer_Handler (ACE_Thread_Manager *tm = 0);
+ virtual int open (void *);
+};
+
+class Consumer_Router : public CONSUMER_ROUTER
+{
+public:
+ Consumer_Router (ACE_Thread_Manager *thr_manager);
+
+protected:
+ // ACE_Task hooks..
+ virtual int open (void *a = 0);
+ virtual int close (u_long flags = 0);
+ virtual int put (ACE_Message_Block *msg, ACE_Time_Value * = 0);
+ virtual int svc (void);
+
+ // Dynamic linking hooks.
+ virtual int info (ACE_TCHAR **info_string, size_t length) const;
+};
+#endif /* ACE_HAS_THREADS */
+#endif /* _CONSUMER_ROUTER_H */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.cpp b/ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.cpp
new file mode 100644
index 00000000000..689a9280766
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.cpp
@@ -0,0 +1,73 @@
+// $Id$
+
+#include "ace/OS_NS_string.h"
+#include "Event_Analyzer.h"
+
+ACE_RCSID(UPIPE_Event_Server, Event_Analyzer, "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+
+int
+Event_Analyzer::open (void *)
+{
+ return 0;
+}
+
+int
+Event_Analyzer::close (u_long)
+{
+ return 0;
+}
+
+int
+Event_Analyzer::control (ACE_Message_Block *mb)
+{
+ ACE_IO_Cntl_Msg *ioc = (ACE_IO_Cntl_Msg *) mb->rd_ptr ();
+ ACE_IO_Cntl_Msg::ACE_IO_Cntl_Cmds cmd;
+
+ switch (cmd = ioc->cmd ())
+ {
+ case ACE_IO_Cntl_Msg::SET_LWM:
+ case ACE_IO_Cntl_Msg::SET_HWM:
+ this->water_marks (cmd, *(size_t *) mb->cont ()->rd_ptr ());
+ break;
+ default:
+ break;
+ }
+ return 0;
+}
+
+int
+Event_Analyzer::put (ACE_Message_Block *mb, ACE_Time_Value *)
+{
+ if (mb->msg_type () == ACE_Message_Block::MB_IOCTL)
+ this->control (mb);
+
+ return this->put_next (mb);
+}
+
+int
+Event_Analyzer::init (int, ACE_TCHAR *[])
+{
+ return 0;
+}
+
+int
+Event_Analyzer::fini (void)
+{
+ return 0;
+}
+
+int
+Event_Analyzer::info (ACE_TCHAR **strp, size_t length) const
+{
+ const ACE_TCHAR *mod_name = this->name ();
+
+ if (*strp == 0 && (*strp = ACE_OS::strdup (mod_name)) == 0)
+ return -1;
+ else
+ ACE_OS::strncpy (*strp, mod_name, length);
+ return ACE_OS::strlen (mod_name);
+}
+
+#endif /* ACE_HAS_THREADS */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h b/ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h
new file mode 100644
index 00000000000..01bc3028964
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h
@@ -0,0 +1,37 @@
+/* -*- C++ -*- */
+// $Id$
+
+// Signal router.
+
+#ifndef _EVENT_ANALYZER_H
+#define _EVENT_ANALYZER_H
+
+#include "ace/Stream.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "ace/Module.h"
+#include "ace/Task.h"
+
+#if defined (ACE_HAS_THREADS)
+
+class Event_Analyzer : public ACE_Task<ACE_MT_SYNCH>
+{
+public:
+ virtual int open (void *a = 0);
+ virtual int close (u_long flags = 0);
+ virtual int put (ACE_Message_Block *msg, ACE_Time_Value * = 0);
+
+ // Dynamic linking hooks.
+ virtual int init (int argc, ACE_TCHAR *argv[]);
+ virtual int fini (void);
+ virtual int info (ACE_TCHAR **info_string, size_t length) const;
+
+private:
+ virtual int control (ACE_Message_Block *);
+};
+
+#endif /* ACE_HAS_THREADS */
+#endif /* _EVENT_ANALYZER_H */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Makefile.am b/ACE/examples/ASX/UPIPE_Event_Server/Makefile.am
new file mode 100644
index 00000000000..f8ac89a7479
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Makefile.am
@@ -0,0 +1,50 @@
+## Process this file with automake to create Makefile.in
+##
+## $Id$
+##
+## This file was generated by MPC. Any changes made directly to
+## this file will be lost the next time it is generated.
+##
+## MPC Command:
+## ./bin/mwc.pl -type automake -noreldefs ACE.mwc
+
+ACE_BUILDDIR = $(top_builddir)
+ACE_ROOT = $(top_srcdir)
+
+
+## Makefile.UPIPE_Event_Server.am
+
+if !BUILD_ACE_FOR_TAO
+
+noinst_PROGRAMS = UPIPE_Event_Server
+
+UPIPE_Event_Server_CPPFLAGS = \
+ -I$(ACE_ROOT) \
+ -I$(ACE_BUILDDIR)
+
+UPIPE_Event_Server_SOURCES = \
+ Consumer_Router.cpp \
+ Event_Analyzer.cpp \
+ Options.cpp \
+ Peer_Router.cpp \
+ Supplier_Router.cpp \
+ event_server.cpp \
+ Consumer_Router.h \
+ Event_Analyzer.h \
+ Options.h \
+ Options.inl \
+ Peer_Router.h \
+ Supplier_Router.h
+
+UPIPE_Event_Server_LDADD = \
+ $(ACE_BUILDDIR)/ace/libACE.la
+
+endif !BUILD_ACE_FOR_TAO
+
+## Clean up template repositories, etc.
+clean-local:
+ -rm -f *~ *.bak *.rpo *.sym lib*.*_pure_* core core.*
+ -rm -f gcctemp.c gcctemp so_locations *.ics
+ -rm -rf cxx_repository ptrepository ti_files
+ -rm -rf templateregistry ir.out
+ -rm -rf ptrepository SunWS_cache Templates.DB
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Options.cpp b/ACE/examples/ASX/UPIPE_Event_Server/Options.cpp
new file mode 100644
index 00000000000..033132610aa
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Options.cpp
@@ -0,0 +1,209 @@
+// $Id$
+
+#include "ace/Get_Opt.h"
+#include "ace/Log_Msg.h"
+#include "ace/OS_NS_Thread.h"
+#include "ace/OS_NS_stdio.h"
+#if defined (ACE_HAS_TRACE)
+# include "ace/OS_NS_strings.h"
+#endif /* ACE_HAS_TRACE */
+
+#include "Options.h"
+
+ACE_RCSID(UPIPE_Event_Server, Options, "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+
+Options::Options (void)
+ : thr_count_ (4),
+ t_flags_ (THR_DETACHED),
+ high_water_mark_ (8 * 1024),
+ low_water_mark_ (1024),
+ message_size_ (128),
+ initial_queue_length_ (0),
+ iterations_ (100000),
+ debugging_ (0),
+ verbosity_ (0),
+ consumer_port_ (ACE_TEXT ("-p 10000")),
+ supplier_port_ (ACE_TEXT ("-p 10001")),
+ consumer_file_ (ACE_TEXT ("-f/tmp/conupipe")),
+ supplier_file_ (ACE_TEXT ("-f/tmp/supupipe"))
+{
+}
+
+Options::~Options (void)
+{
+}
+
+void Options::print_results (void)
+{
+ ACE_Profile_Timer::ACE_Elapsed_Time et;
+ this->itimer_.elapsed_time (et);
+
+#if defined (ACE_HAS_PRUSAGE_T)
+ prusage_t rusage;
+ this->itimer_.get_rusage (rusage);
+
+ if (options.verbose ())
+ {
+ ACE_OS::printf ("final concurrency hint = %d\n", ACE_OS::thr_getconcurrency ());
+ ACE_OS::printf ("%8d = lwpid\n"
+ "%8d = lwp count\n"
+ "%8d = minor page faults\n"
+ "%8d = major page faults\n"
+ "%8d = input blocks\n"
+ "%8d = output blocks\n"
+ "%8d = messages sent\n"
+ "%8d = messages received\n"
+ "%8d = signals received\n"
+ "%8ds, %dms = wait-cpu (latency) time\n"
+ "%8ds, %dms = user lock wait sleep time\n"
+ "%8ds, %dms = all other sleep time\n"
+ "%8d = voluntary context switches\n"
+ "%8d = involuntary context switches\n"
+ "%8d = system calls\n"
+ "%8d = chars read/written\n",
+ (int) rusage.pr_lwpid,
+ (int) rusage.pr_count,
+ (int) rusage.pr_minf,
+ (int) rusage.pr_majf,
+ (int) rusage.pr_inblk,
+ (int) rusage.pr_oublk,
+ (int) rusage.pr_msnd,
+ (int) rusage.pr_mrcv,
+ (int) rusage.pr_sigs,
+ (int) rusage.pr_wtime.tv_sec, (int) rusage.pr_wtime.tv_nsec / 1000000,
+ (int) rusage.pr_ltime.tv_sec, (int) rusage.pr_ltime.tv_nsec / 1000000,
+ (int) rusage.pr_slptime.tv_sec, (int) rusage.pr_slptime.tv_nsec / 1000000,
+ (int) rusage.pr_vctx,
+ (int) rusage.pr_ictx,
+ (int) rusage.pr_sysc,
+ (int) rusage.pr_ioch);
+ }
+#endif /* ACE_HAS_PRUSAGE_T */
+
+ ACE_OS::printf ("---------------------\n"
+ "real time = %.3f\n"
+ "user time = %.3f\n"
+ "system time = %.3f\n"
+ "---------------------\n",
+ et.real_time, et.user_time, et.system_time);
+}
+
+// Manages the options.
+Options options;
+
+void
+Options::parse_args (int argc, ACE_TCHAR *argv[])
+{
+ ACE_LOG_MSG->open (argv[0]);
+
+ //FUZZ: disable check_for_lack_ACE_OS
+ ACE_Get_Opt getopt (argc, argv, ACE_TEXT ("C:c:bdH:i:L:l:M:nS:s:t:T:v"));
+ int c;
+
+ while ((c = getopt ()) != -1)
+ //FUZZ: enable check_for_lack_ACE_OS
+ switch (c)
+ {
+ case 'b':
+ this->t_flags (THR_BOUND);
+ break;
+ case 'C':
+ this->consumer_file (getopt.opt_arg ());
+ break;
+ case 'c':
+ this->consumer_port (getopt.opt_arg ());
+ break;
+ case 'd':
+ this->debugging_ = 1;
+ break;
+ case 'H':
+ this->high_water_mark (ACE_OS::atoi (getopt.opt_arg ()));
+ break;
+ case 'i':
+ this->iterations (ACE_OS::atoi (getopt.opt_arg ()));
+ break;
+ case 'L':
+ this->low_water_mark (ACE_OS::atoi (getopt.opt_arg ()));
+ break;
+ case 'l':
+ this->initial_queue_length (ACE_OS::atoi (getopt.opt_arg ()));
+ break;
+ case 'M':
+ this->message_size (ACE_OS::atoi (getopt.opt_arg ()));
+ break;
+ case 'n':
+ this->t_flags (THR_NEW_LWP);
+ break;
+ case 'S':
+ this->supplier_file (getopt.opt_arg ());
+ break;
+ case 's':
+ this->supplier_port (getopt.opt_arg ());
+ break;
+ case 'T':
+ #if defined (ACE_HAS_TRACE)
+ if (ACE_OS::strcasecmp (getopt.opt_arg (), ACE_TEXT ("ON")) == 0)
+ ACE_Trace::start_tracing ();
+ else if (ACE_OS::strcasecmp (getopt.opt_arg (), ACE_TEXT ("OFF")) == 0)
+ ACE_Trace::stop_tracing ();
+ #endif /* ACE_HAS_TRACE */
+ break;
+ case 't':
+ this->thr_count (ACE_OS::atoi (getopt.opt_arg ()));
+ break;
+ case 'v':
+ this->verbosity_ = 1;
+ break;
+ default:
+ ACE_OS::fprintf (stderr, "%s\n"
+ "\t[-b] (THR_BOUND)\n"
+ "\t[-C consumer file]\n"
+ "\t[-c consumer port]\n"
+ "\t[-d] (enable debugging)\n"
+ "\t[-H high water mark]\n"
+ "\t[-i number of test iterations]\n"
+ "\t[-L low water mark]\n"
+ "\t[-M] message size \n"
+ "\t[-n] (THR_NEW_LWP)\n"
+ "\t[-q max queue size]\n"
+ "\t[-S supplier file]\n"
+ "\t[-s supplier port]\n"
+ "\t[-t number of threads]\n"
+ "\t[-v] (verbose) \n",
+ ACE_TEXT_ALWAYS_CHAR (argv[0]));
+ ACE_OS::exit (1);
+ /* NOTREACHED */
+ break;
+ }
+
+ // HACK! This needs to be done to avoid the mismatch from ACE_TEXT
+ // in ACE_SIZE_T_FORMAT_SPECIFIER to narrow-char on wide-char builds.
+ // It only works because it's at the end of the file.
+# if defined (ACE_TEXT)
+# undef ACE_TEXT
+# endif
+# define ACE_TEXT(X) X
+ if (this->verbose ())
+ ACE_OS::printf ("%8d = initial concurrency hint\n"
+ ACE_SIZE_T_FORMAT_SPECIFIER " = total iterations\n"
+ ACE_SIZE_T_FORMAT_SPECIFIER " = thread count\n"
+ ACE_SIZE_T_FORMAT_SPECIFIER " = low water mark\n"
+ ACE_SIZE_T_FORMAT_SPECIFIER " = high water mark\n"
+ ACE_SIZE_T_FORMAT_SPECIFIER " = message_size\n"
+ ACE_SIZE_T_FORMAT_SPECIFIER " = initial queue length\n"
+ "%8d = THR_BOUND\n"
+ "%8d = THR_NEW_LWP\n",
+ ACE_OS::thr_getconcurrency (),
+ this->iterations (),
+ this->thr_count (),
+ this->low_water_mark (),
+ this->high_water_mark (),
+ this->message_size (),
+ this->initial_queue_length (),
+ (this->t_flags () & THR_BOUND) != 0,
+ (this->t_flags () & THR_NEW_LWP) != 0);
+}
+
+#endif /* ACE_HAS_THREADS */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Options.h b/ACE/examples/ASX/UPIPE_Event_Server/Options.h
new file mode 100644
index 00000000000..d775a49a928
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Options.h
@@ -0,0 +1,88 @@
+/* -*- C++ -*- */
+// $Id$
+
+// Option manager for Event Server.
+
+#ifndef DEVICE_OPTIONS_H
+#define DEVICE_OPTIONS_H
+
+#include "ace/config-all.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "ace/Profile_Timer.h"
+
+#if defined (ACE_HAS_THREADS)
+
+class Options
+{
+public:
+ Options (void);
+ ~Options (void);
+ void parse_args (int argc, ACE_TCHAR *argv[]);
+
+ void stop_timer (void);
+ void start_timer (void);
+
+ void thr_count (size_t count);
+ size_t thr_count (void);
+
+ void initial_queue_length (size_t length);
+ size_t initial_queue_length (void);
+
+ void high_water_mark (size_t size);
+ size_t high_water_mark (void);
+
+ void low_water_mark (size_t size);
+ size_t low_water_mark (void);
+
+ void message_size (size_t size);
+ size_t message_size (void);
+
+ void iterations (size_t n);
+ size_t iterations (void);
+
+ void t_flags (long flag);
+ long t_flags (void);
+
+ void supplier_port (const ACE_TCHAR *port);
+ const ACE_TCHAR *supplier_port (void);
+
+ void consumer_port (const ACE_TCHAR *port);
+ const ACE_TCHAR *consumer_port (void);
+
+ void supplier_file (const ACE_TCHAR *file);
+ const ACE_TCHAR *supplier_file (void);
+
+ void consumer_file (const ACE_TCHAR *file);
+ const ACE_TCHAR *consumer_file (void);
+
+ int debug (void);
+ int verbose (void);
+
+ void print_results (void);
+
+private:
+ ACE_Profile_Timer itimer_; // Time the process.
+ size_t thr_count_; // Number of threads to spawn.
+ long t_flags_; // Flags to thr_create().
+ size_t high_water_mark_; // ACE_Task high water mark.
+ size_t low_water_mark_; // ACE_Task low water mark.
+ size_t message_size_; // Size of a message.
+ size_t initial_queue_length_; // Initial number of items in the queue.
+ size_t iterations_; // Number of iterations to run the test program.
+ int debugging_; // Extra debugging info.
+ int verbosity_; // Extra verbose messages.
+ const ACE_TCHAR *consumer_port_; // Port that the Consumer_Router is using.
+ const ACE_TCHAR *supplier_port_; // Port that the Supplier_Router is using.
+ const ACE_TCHAR *consumer_file_; // file that the Consumer_Router is using.
+ const ACE_TCHAR *supplier_file_; // file that the Supplier_Router is using.
+};
+
+extern Options options;
+
+#include "Options.inl"
+#endif /* ACE_HAS_THREADS */
+#endif /* DEVICE_OPTIONS_H */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Options.inl b/ACE/examples/ASX/UPIPE_Event_Server/Options.inl
new file mode 100644
index 00000000000..af04f73eb26
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Options.inl
@@ -0,0 +1,166 @@
+/* -*- C++ -*- */
+// $Id$
+
+// Option manager for ustreams.
+
+// Since this is only included in Options.h these should stay
+// inline, not ACE_INLINE.
+// FUZZ: disable check_for_inline
+
+
+inline void
+Options::supplier_port (const ACE_TCHAR *port)
+{
+ this->supplier_port_ = port;
+}
+
+inline const ACE_TCHAR *
+Options::supplier_port (void)
+{
+ return this->supplier_port_;
+}
+
+inline void
+Options::supplier_file (const ACE_TCHAR *file)
+{
+ this->supplier_file_ = file;
+}
+
+inline const ACE_TCHAR *
+Options::supplier_file (void)
+{
+ return this->supplier_file_;
+}
+
+inline void
+Options::consumer_file (const ACE_TCHAR *file)
+{
+ this->consumer_file_ = file;
+}
+
+inline const ACE_TCHAR *
+Options::consumer_file (void)
+{
+ return this->consumer_file_;
+}
+
+inline void
+Options::consumer_port (const ACE_TCHAR *port)
+{
+ this->consumer_port_ = port;
+}
+
+inline const ACE_TCHAR *
+Options::consumer_port (void)
+{
+ return this->consumer_port_;
+}
+
+inline void
+Options::start_timer (void)
+{
+ this->itimer_.start ();
+}
+
+inline void
+Options::stop_timer (void)
+{
+ this->itimer_.stop ();
+}
+
+inline void
+Options::thr_count (size_t count)
+{
+ this->thr_count_ = count;
+}
+
+inline size_t
+Options::thr_count (void)
+{
+ return this->thr_count_;
+}
+
+inline void
+Options::initial_queue_length (size_t length)
+{
+ this->initial_queue_length_ = length;
+}
+
+inline size_t
+Options::initial_queue_length (void)
+{
+ return this->initial_queue_length_;
+}
+
+inline void
+Options::high_water_mark (size_t size)
+{
+ this->high_water_mark_ = size;
+}
+
+inline size_t
+Options::high_water_mark (void)
+{
+ return this->high_water_mark_;
+}
+
+inline void
+Options::low_water_mark (size_t size)
+{
+ this->low_water_mark_ = size;
+}
+
+inline size_t
+Options::low_water_mark (void)
+{
+ return this->low_water_mark_;
+}
+
+inline void
+Options::message_size (size_t size)
+{
+ this->message_size_ = size;
+}
+
+inline size_t
+Options::message_size (void)
+{
+ return this->message_size_;
+}
+
+inline void
+Options::iterations (size_t n)
+{
+ this->iterations_ = n;
+}
+
+inline size_t
+Options::iterations (void)
+{
+ return this->iterations_;
+}
+
+inline void
+Options::t_flags (long flag)
+{
+ this->t_flags_ |= flag;
+}
+
+inline long
+Options::t_flags (void)
+{
+ return this->t_flags_;
+}
+
+inline int
+Options::debug (void)
+{
+ return this->debugging_;
+}
+
+inline int
+Options::verbose (void)
+{
+ return this->verbosity_;
+}
+
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.cpp b/ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.cpp
new file mode 100644
index 00000000000..757eecedc33
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.cpp
@@ -0,0 +1,283 @@
+// $Id$
+
+#if !defined (_PEER_ROUTER_C)
+
+#define _PEER_ROUTER_C
+
+#include "ace/Get_Opt.h"
+#include "ace/Service_Config.h"
+
+#include "Peer_Router.h"
+#include "Options.h"
+
+ACE_RCSID(UPIPE_Event_Server, Peer_Router, "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+
+// Define some short-hand macros to deal with long templates
+// names...
+
+#define PH PEER_HANDLER
+#define PA PEER_ACCEPTOR
+#define PAD PEER_ADDR
+#define PK PEER_KEY
+#define PM PEER_MAP
+
+template <class PH, class PK> int
+Acceptor_Factory<PH, PK>::init (int argc, ACE_TCHAR *argv[])
+{
+ ACE_Get_Opt get_opt (argc, argv, ACE_TEXT ("df:"), 0);
+ ACE_UPIPE_Addr addr;
+
+ for (int c; (c = get_opt ()) != -1; )
+ switch (c)
+ {
+ case 'f':
+ addr.set (get_opt.opt_arg ());
+ break;
+ case 'd':
+ break;
+ default:
+ break;
+ }
+
+ if (this->open (addr, ACE_Reactor::instance ()) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("open")), -1);
+ return 0;
+}
+
+template <class PH, class PK>
+Acceptor_Factory<PH, PK>::Acceptor_Factory (Peer_Router<PH, PK> *pr)
+ : pr_ (pr)
+{
+}
+
+template <class PH, class PK> Peer_Router<PH, PK> *
+Acceptor_Factory<PH, PK>::router (void)
+{
+ return this->pr_;
+}
+
+template <class ROUTER, class KEY>
+Peer_Handler<ROUTER, KEY>::Peer_Handler (ACE_Thread_Manager *tm)
+ : ACE_Svc_Handler<ACE_UPIPE_STREAM, ACE_MT_SYNCH> (tm)
+{
+}
+
+template <class ROUTER, class KEY> int
+Peer_Handler<ROUTER, KEY>::svc (void)
+{
+ // Just a try !! we're just reading from our ACE_Message_Queue.
+ ACE_Message_Block *db, *hb;
+ int n;
+ // do an endless loop
+ for (;;)
+ {
+ db = new ACE_Message_Block (BUFSIZ);
+ hb = new ACE_Message_Block (sizeof (KEY), ACE_Message_Block::MB_PROTO, db);
+
+ if ((n = this->peer ().recv (db->rd_ptr (), db->size ())) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("recv failed")), -1);
+ else if (n == 0) // Client has closed down the connection.
+ {
+
+ if (this->router_task_->unbind_peer (this->get_handle ()) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("unbind failed")), -1);
+ ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) shutting down \n")));
+ return -1; // We do not need to be deregistered by reactor
+ // as we were not registered at all
+ }
+ else // Transform incoming buffer into a Message and pass downstream.
+ {
+ db->wr_ptr (n);
+ *(ACE_HANDLE *) hb->rd_ptr () = this->get_handle (); // structure assignment.
+ hb->wr_ptr (sizeof (long));
+ if (this->router_task_->reply (hb) == -1)
+ {
+ ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Peer_Handler.svc : router_task->reply failed\n")));
+ return -1;
+ }
+
+ // return this->router_task_->reply (hb) == -1 ? -1 : 0;
+ }
+ }
+ ACE_NOTREACHED(return 0);
+}
+
+template <class ROUTER, class KEY> int
+Peer_Handler<ROUTER, KEY>::put (ACE_Message_Block *mb, ACE_Time_Value *)
+{
+ return this->peer ().send_n (mb->rd_ptr (), mb->length ());
+}
+
+// Create a new handler and point its ROUTER_TASK_ data member to the
+// corresponding router. Note that this router is extracted out of
+// the Acceptor_Factory * that is passed in via the
+// ACE_Acceptor::handle_input() method.
+
+template <class ROUTER, class KEY> int
+Peer_Handler<ROUTER, KEY>::open (void *a)
+{
+ ACE_TCHAR buf[BUFSIZ], *p = buf;
+
+ if (this->router_task_->info (&p, sizeof buf) != -1)
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%t) creating handler for %s, fd = %d, this = %@\n"),
+ buf, this->get_handle (), a));
+ else
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("info")), -1);
+
+ if ( this->activate (options.t_flags ()) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("activation of thread failed")), -1);
+ else if (this->router_task_->bind_peer (this->get_handle (), this) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("bind_peer")), -1);
+ return 0;
+}
+
+// Receive a message from a supplier..
+
+template <class ROUTER, class KEY> int
+Peer_Handler<ROUTER, KEY>::handle_input (ACE_HANDLE h)
+{
+
+ ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) input arrived on sd %d\n"), h));
+// ACE_Reactor::instance ()->remove_handler(h,
+// ACE_Event_Handler::ALL_EVENTS_MASK
+// |ACE_Event_Handler::DONT_CALL);
+// this method should be called only if the peer shuts down
+// so we deactivate our ACE_Message_Queue to awake our svc thread
+
+ return 0;
+
+#if 0
+ ACE_Message_Block *db = new ACE_Message_Block (BUFSIZ);
+ ACE_Message_Block *hb = new ACE_Message_Block (sizeof (KEY), ACE_Message_Block::MB_PROTO, db);
+ int n;
+
+ if ((n = this->peer ().recv (db->rd_ptr (), db->size ())) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("recv failed")), -1);
+ else if (n == 0) // Client has closed down the connection.
+ {
+ if (this->router_task_->unbind_peer (this->get_handle ()) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("unbind failed")), -1);
+ ACE_DEBUG ((LM_DEBUG, "(%t) shutting down %d\n", h));
+ return -1; // Instruct the ACE_Reactor to deregister us by returning -1.
+ }
+ else // Transform incoming buffer into a Message and pass downstream.
+ {
+ db->wr_ptr (n);
+ *(long *) hb->rd_ptr () = this->get_handle (); // structure assignment.
+ hb->wr_ptr (sizeof (long));
+ return this->router_task_->reply (hb) == -1 ? -1 : 0;
+ }
+#endif
+}
+
+template <class PH, class PK>
+Peer_Router<PH, PK>::Peer_Router (ACE_Thread_Manager *tm)
+ : ACE_Task<ACE_MT_SYNCH> (tm)
+{
+}
+
+template <class PH, class PK> int
+Peer_Router<PH, PK>::send_peers (ACE_Message_Block *mb)
+{
+ ACE_Map_Iterator<PK, PH *, ACE_RW_Mutex> map_iter = this->peer_map_;
+ int bytes = 0;
+ int iterations = 0;
+ ACE_Message_Block *data_block = mb->cont ();
+ for (ACE_Map_Entry<PK, PH *> *ss = 0;
+ map_iter.next (ss) != 0;
+ map_iter.advance ())
+ {
+ if (options.debug ())
+ ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) sending to peer via sd %d\n"),
+ ss->ext_id_));
+
+ iterations++;
+ bytes += ss->int_id_->put (data_block);
+ }
+
+ mb->release ();
+ return bytes == 0 ? 0 : bytes / iterations;
+}
+
+template <class PH, class PK>
+Peer_Router<PH, PK>::~Peer_Router (void)
+{
+}
+
+template <class PH, class PK> int
+Peer_Router<PH, PK>::fini (void)
+{
+ delete this->acceptor_;
+ return 0;
+}
+
+template <class PH, class PK> int
+Peer_Router<PH, PK>::control (ACE_Message_Block *mb)
+{
+ ACE_IO_Cntl_Msg *ioc = (ACE_IO_Cntl_Msg *) mb->rd_ptr ();
+ ACE_IO_Cntl_Msg::ACE_IO_Cntl_Cmds command;
+
+ switch (command = ioc->cmd ())
+ {
+ case ACE_IO_Cntl_Msg::SET_LWM:
+ case ACE_IO_Cntl_Msg::SET_HWM:
+ this->water_marks (command, *(size_t *) mb->cont ()->rd_ptr ());
+ break;
+ default:
+ return -1;
+ }
+ return 0;
+}
+
+template <class PH, class PK> int
+Peer_Router<PH, PK>::unbind_peer (PK key)
+{
+ return this->peer_map_.unbind (key);
+}
+
+template <class PH, class PK> int
+Peer_Router<PH, PK>::bind_peer (PK key, Peer_Handler<Peer_Router<PH, PK>, PK> *ph)
+{
+ PH *peer_handler = (PH *) ph;
+ return this->peer_map_.bind (key, peer_handler);
+}
+
+template <class PH, class PK> int
+Peer_Router<PH, PK>::init (int argc, ACE_TCHAR *argv[])
+{
+ this->acceptor_ = new Acceptor_Factory <PH, PK> (this);
+
+ if (this->acceptor_->init (argc, argv) == -1
+ || this->peer_map_.open () == -1)
+ return -1;
+ else
+ {
+ ACE_UPIPE_Addr addr;
+ ACE_UPIPE_Acceptor &pa = this->acceptor_->acceptor ();
+
+ if (pa.get_local_addr (addr) != -1)
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("(%t) initializing %s, file = %s, fd = %d, this = %@\n"),
+ this->name (), addr.get_path_name (), pa.get_handle (), this));
+ else
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("get_local_addr")), -1);
+ }
+ return 0;
+}
+
+#undef PH
+#undef PA
+#undef PAD
+#undef PK
+#undef PM
+#endif /* ACE_HAS_THREADS */
+#endif /* _PEER_ROUTER_C */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.h b/ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.h
new file mode 100644
index 00000000000..3962d371ae0
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Peer_Router.h
@@ -0,0 +1,127 @@
+/* -*- C++ -*- */
+// $Id$
+
+// The interface between one or more peers and a stream. A peer
+// typically runs remotely on another machine.
+
+#ifndef _PEER_ROUTER_H
+#define _PEER_ROUTER_H
+
+#include "ace/Acceptor.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "ace/Svc_Handler.h"
+#include "ace/UPIPE_Acceptor.h"
+#include "ace/UPIPE_Addr.h"
+#include "ace/Thread_Manager.h"
+#include "ace/Map_Manager.h"
+
+#if defined (ACE_HAS_THREADS)
+#include "ace/RW_Mutex.h"
+
+// Forward declaration.
+template <class PEER_HANDLER, class KEY>
+class Peer_Router;
+
+template <class PEER_HANDLER, class KEY>
+class Acceptor_Factory : public ACE_Acceptor<PEER_HANDLER, ACE_UPIPE_ACCEPTOR>
+{
+public:
+ Acceptor_Factory (Peer_Router<PEER_HANDLER, KEY> *pr);
+ Peer_Router<PEER_HANDLER, KEY> *router (void);
+
+ int init (int argc, ACE_TCHAR *argv[]);
+ // Initialize the acceptor when it's linked dynamically.
+
+private:
+ Peer_Router<PEER_HANDLER, KEY> *pr_;
+};
+
+// Receive input from a Peer..
+template <class ROUTER, class KEY>
+class Peer_Handler : public ACE_Svc_Handler<ACE_UPIPE_STREAM, ACE_MT_SYNCH>
+{
+public:
+ Peer_Handler (ACE_Thread_Manager * = 0);
+
+ virtual int open (void * = 0);
+ // Called by the ACE_Acceptor::handle_input() to activate this object.
+
+ virtual int handle_input (ACE_HANDLE);
+ // Receive input from the peer..
+
+ virtual int put (ACE_Message_Block *, ACE_Time_Value *tv = 0);
+ // Send output to a peer.
+
+protected:
+ ROUTER *router_task_;
+ // Pointer to write task..
+
+private:
+ // Don't need this method here...
+ virtual int svc (void);
+};
+
+// This abstract base class provides mechanisms for routing messages
+// to/from a ACE_Stream from/to one or more peers (which are typically
+// running on remote hosts). A subclass of Peer_Router overrides the
+// open(), close(), and put() methods in order to specialize the
+// behavior of the router to meet application-specific requirements.
+
+template <class PEER_HANDLER, class PEER_KEY>
+class Peer_Router : public ACE_Task<ACE_MT_SYNCH>
+{
+public:
+ Peer_Router (ACE_Thread_Manager * = 0);
+ ~Peer_Router (void);
+
+ typedef Peer_Handler<Peer_Router<PEER_HANDLER, PEER_KEY>, PEER_KEY> HANDLER;
+
+ // Remove a PEER_HANDLER from the PEER_MAP.
+ virtual int unbind_peer (PEER_KEY);
+
+ // Add a PEER_HANDLER to the PEER_MAP.
+ virtual int bind_peer (PEER_KEY, HANDLER *);
+
+ // Send the message block to the peer(s)..
+ int send_peers (ACE_Message_Block *mb);
+
+protected:
+// Handle control messages arriving from adjacent Modules.
+ virtual int control (ACE_Message_Block *);
+
+ // Map used to keep track of active peers.
+ ACE_Map_Manager <PEER_KEY, PEER_HANDLER *, ACE_RW_Mutex> peer_map_;
+
+ // Dynamic linking initialization hooks inherited from ACE_Task.
+ virtual int init (int argc, ACE_TCHAR *argv[]);
+ virtual int fini (void);
+
+ // Factory for accepting new PEER_HANDLERs.
+ Acceptor_Factory<PEER_HANDLER, PEER_KEY> *acceptor_;
+
+private:
+// Prevent copies and pass-by-value.
+ ACE_UNIMPLEMENTED_FUNC (Peer_Router (const Peer_Router<PEER_HANDLER, PEER_KEY> &))
+ ACE_UNIMPLEMENTED_FUNC (void operator= (const Peer_Router<PEER_HANDLER, PEER_KEY> &))
+};
+
+#if defined (__ACE_INLINE__)
+#define ACE_INLINE inline
+#else
+#define ACE_INLINE
+#endif /* __ACE_INLINE__ */
+
+#if defined (ACE_TEMPLATES_REQUIRE_SOURCE)
+#include "Peer_Router.cpp"
+#endif /* ACE_TEMPLATES_REQUIRE_SOURCE */
+
+#if defined (ACE_TEMPLATES_REQUIRE_PRAGMA)
+#pragma implementation ("Peer_Router.cpp")
+#endif /* ACE_TEMPLATES_REQUIRE_PRAGMA */
+
+#endif /* ACE_HAS_THREADS */
+#endif /* _PEER_ROUTER_H */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp b/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp
new file mode 100644
index 00000000000..8b4a53d6331
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp
@@ -0,0 +1,137 @@
+// $Id$
+
+#include "ace/OS_NS_stdio.h"
+#include "ace/OS_NS_string.h"
+#include "Options.h"
+#include "Supplier_Router.h"
+
+ACE_RCSID(UPIPE_Event_Server, Supplier_Router, "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+
+typedef Acceptor_Factory<Supplier_Handler, SUPPLIER_KEY> SUPPLIER_FACTORY;
+
+int
+Supplier_Handler::open (void *a)
+{
+ SUPPLIER_FACTORY *af = (SUPPLIER_FACTORY *) a;
+ this->router_task_ = af->router ();
+ return this->Peer_Handler<SUPPLIER_ROUTER, SUPPLIER_KEY>::open (a);
+}
+
+Supplier_Handler::Supplier_Handler (ACE_Thread_Manager *tm)
+ : Peer_Handler<SUPPLIER_ROUTER, SUPPLIER_KEY> (tm)
+{
+}
+
+// Create a new router and associate it with the REACTOR parameter..
+
+Supplier_Router::Supplier_Router (ACE_Thread_Manager *tm)
+ : SUPPLIER_ROUTER (tm)
+{
+}
+
+// Handle incoming messages in a separate thread..
+
+int
+Supplier_Router::svc (void)
+{
+ ACE_ASSERT (this->is_writer ());
+
+ ACE_Message_Block *message_block = 0;
+
+ if (options.debug ())
+ ACE_DEBUG ((LM_DEBUG, "(%t) starting svc in %s\n", this->name ()));
+
+ while (this->getq (message_block) > 0)
+ {
+ if (this->put_next (message_block) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, "(%t) put_next failed in %s\n", this->name ()), -1);
+ }
+
+ return 0;
+}
+
+// Initialize the Router..
+
+int
+Supplier_Router::open (void *)
+{
+ ACE_ASSERT (this->is_writer ());
+ ACE_TCHAR *argv[3];
+
+ argv[0] = (ACE_TCHAR *)this->name ();
+ argv[1] = (ACE_TCHAR *)options.supplier_file ();
+ argv[2] = 0;
+
+ if (this->init (1, &argv[1]) == -1)
+ return -1;
+
+ // Make this an active object.
+ // return this->activate (options.t_flags ());
+
+ // Until that's done, return 1 to indicate that the object wasn't activated.
+ return 1;
+}
+
+// Close down the router..
+
+int
+Supplier_Router::close (u_long)
+{
+ ACE_ASSERT (this->is_writer ());
+ this->peer_map_.close ();
+ this->msg_queue ()->deactivate();
+ return 0;
+}
+
+// Send a MESSAGE_BLOCK to the supplier(s)..
+
+int
+Supplier_Router::put (ACE_Message_Block *mb, ACE_Time_Value *)
+{
+ ACE_ASSERT (this->is_writer ());
+
+ if (mb->msg_type () == ACE_Message_Block::MB_IOCTL)
+ {
+ this->control (mb);
+ return this->put_next (mb);
+ }
+ else
+ {
+//printf("supplier-Router is routing: send_peers\n");
+ return this->send_peers (mb);
+ }
+}
+
+// Return information about the Supplier_Router ACE_Module..
+
+int
+Supplier_Router::info (ACE_TCHAR **strp, size_t length) const
+{
+ ACE_TCHAR buf[BUFSIZ];
+ ACE_UPIPE_Addr addr;
+ const ACE_TCHAR *mod_name = this->name ();
+ ACE_UPIPE_Acceptor &sa = (ACE_UPIPE_Acceptor &) *this->acceptor_;
+
+ if (sa.get_local_addr (addr) == -1)
+ return -1;
+
+#if !defined (ACE_WIN32) && defined (ACE_USES_WCHAR)
+# define FMTSTR ACE_TEXT ("%ls\t %ls/ %ls")
+#else
+# define FMTSTR ACE_TEXT ("%s\t %s/ %s")
+#endif
+
+ ACE_OS::sprintf (buf, FMTSTR,
+ mod_name, ACE_TEXT ("upipe"),
+ ACE_TEXT ("# supplier router\n"));
+
+ if (*strp == 0 && (*strp = ACE_OS::strdup (mod_name)) == 0)
+ return -1;
+ else
+ ACE_OS::strncpy (*strp, mod_name, length);
+ return ACE_OS::strlen (mod_name);
+}
+
+#endif /* ACE_HAS_THREADS */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.h b/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.h
new file mode 100644
index 00000000000..4d5d440e018
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.h
@@ -0,0 +1,57 @@
+/* -*- C++ -*- */
+// $Id$
+
+// The interface between a supplier and an Event Service ACE_Stream.
+
+#ifndef _SUPPLIER_ROUTER_H
+#define _SUPPLIER_ROUTER_H
+
+#include "ace/UPIPE_Addr.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "ace/UPIPE_Acceptor.h"
+#include "ace/Map_Manager.h"
+#include "ace/Svc_Handler.h"
+#include "Peer_Router.h"
+
+#if defined (ACE_HAS_THREADS)
+
+// Forward declaration.
+class Supplier_Handler;
+
+// Type of search key for SUPPLIER_MAP.
+typedef ACE_HANDLE SUPPLIER_KEY;
+
+// Instantiated type for routing messages to suppliers.
+
+typedef Peer_Router<Supplier_Handler, SUPPLIER_KEY> SUPPLIER_ROUTER;
+
+class Supplier_Handler
+ : public Peer_Handler<SUPPLIER_ROUTER, SUPPLIER_KEY>
+{
+public:
+ Supplier_Handler (ACE_Thread_Manager *tm = 0);
+ virtual int open (void *);
+};
+
+class Supplier_Router : public SUPPLIER_ROUTER
+{
+public:
+ Supplier_Router (ACE_Thread_Manager *);
+
+protected:
+ // ACE_Task hooks..
+ virtual int open (void *a = 0);
+ virtual int close (u_long flags = 0);
+ virtual int put (ACE_Message_Block *msg, ACE_Time_Value * = 0);
+ virtual int svc (void);
+
+ // Dynamic linking hooks inherited from Peer_Router.
+ virtual int info (ACE_TCHAR **info_string, size_t length) const;
+};
+
+#endif /* ACE_HAS_THREADS */
+#endif /* _SUPPLIER_ROUTER_H */
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/UPIPE_Event.mpc b/ACE/examples/ASX/UPIPE_Event_Server/UPIPE_Event.mpc
new file mode 100644
index 00000000000..a4c93dc1fdb
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/UPIPE_Event.mpc
@@ -0,0 +1,15 @@
+// -*- MPC -*-
+// $Id$
+
+project(*Server) : aceexe {
+ avoids += ace_for_tao
+ exename = UPIPE_Event_Server
+ Source_Files {
+ Consumer_Router.cpp
+ Event_Analyzer.cpp
+ Options.cpp
+ Peer_Router.cpp
+ Supplier_Router.cpp
+ event_server.cpp
+ }
+}
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/event_server.cpp b/ACE/examples/ASX/UPIPE_Event_Server/event_server.cpp
new file mode 100644
index 00000000000..91ae382fac1
--- /dev/null
+++ b/ACE/examples/ASX/UPIPE_Event_Server/event_server.cpp
@@ -0,0 +1,271 @@
+// $Id$
+
+// Test the event server.
+
+#include "ace/OS_main.h"
+#include "ace/Stream.h"
+#include "ace/Service_Config.h"
+#include "ace/UPIPE_Acceptor.h"
+#include "ace/UPIPE_Connector.h"
+
+// FUZZ: disable check_for_streams_include
+#include "ace/streams.h"
+
+#include "Options.h"
+#include "Consumer_Router.h"
+#include "Event_Analyzer.h"
+#include "Supplier_Router.h"
+#include "ace/Sig_Adapter.h"
+#include "ace/OS_NS_unistd.h"
+
+ACE_RCSID (UPIPE_Event_Server,
+ event_server,
+ "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+
+typedef ACE_Stream<ACE_MT_SYNCH> MT_Stream;
+typedef ACE_Module<ACE_MT_SYNCH> MT_Module;
+
+// Handle SIGINT and terminate the entire application.
+
+class Quit_Handler : public ACE_Sig_Adapter
+{
+public:
+ Quit_Handler (void);
+ virtual int handle_input (ACE_HANDLE fd);
+};
+
+Quit_Handler::Quit_Handler (void)
+ : ACE_Sig_Adapter (ACE_Sig_Handler_Ex (ACE_Reactor::end_event_loop))
+{
+ // Register to trap input from the user.
+ if (ACE_Event_Handler::register_stdin_handler (this,
+ ACE_Reactor::instance (),
+ ACE_Thread_Manager::instance ()) == -1)
+ ACE_ERROR ((LM_ERROR, "%p\n", "register_stdin_handler"));
+ // Register to trap the SIGINT signal.
+ else if (ACE_Reactor::instance ()->register_handler
+ (SIGINT, this) == -1)
+ ACE_ERROR ((LM_ERROR, "%p\n", "register_handler"));
+}
+
+int
+Quit_Handler::handle_input (ACE_HANDLE)
+{
+ options.stop_timer ();
+ ACE_DEBUG ((LM_INFO, " (%t) closing down the test\n"));
+ options.print_results ();
+
+ ACE_Reactor::end_event_loop();
+ return 0;
+}
+
+static void *
+consumer (void *)
+{
+ ACE_UPIPE_Stream c_stream;
+ ACE_UPIPE_Addr c_addr (ACE_TEXT ("/tmp/conupipe"));
+
+ int verb = options.verbose ();
+ int msiz = options.message_size ();
+ time_t secs, par1, par2;
+ time_t currsec;
+
+ if (verb)
+ cout << "consumer starting connect" << endl;
+
+ ACE_UPIPE_Connector con;
+
+ if (con.connect (c_stream, c_addr) == -1)
+ ACE_DEBUG ((LM_INFO, " (%t) connect failed\n"));
+ else
+ cout << "consumer :we're connected" << endl;
+
+ ACE_Message_Block *mb_p;
+
+ int done = 0;
+ int cnt = 0;
+ ACE_OS::time (&currsec);
+
+ par1 = currsec;
+
+ while (done == 0
+ && (c_stream.recv (mb_p) != -1))
+ if (mb_p->length () > 1)
+ {
+ cnt++;
+ if (verb)
+ cout << " consumer received message !!!!!! "
+ << mb_p->rd_ptr () << endl;
+ }
+ else
+ {
+ if (verb)
+ cout << "consumer got last mb"
+ << (char) * (mb_p->rd_ptr ()) << endl;
+ c_stream.close ();
+ done = 1;
+ }
+
+ ACE_OS::time (&currsec);
+ par2 = currsec;
+
+ secs = par2 - par1;
+
+ if (secs <= 0)
+ secs=1;
+
+ ACE_DEBUG ((LM_INFO,
+ ACE_TEXT ("consumer got %d messages of size %d ")
+ ACE_TEXT ("within %: seconds\n"),
+ cnt, msiz, secs));
+
+ ACE_OS::sleep (2);
+ cout << "consumer terminating " << endl;
+ return 0;
+}
+
+static void *
+supplier (void *dummy)
+{
+ ACE_UPIPE_Stream s_stream;
+ ACE_UPIPE_Addr serv_addr (ACE_TEXT ("/tmp/supupipe"));
+ ACE_UPIPE_Connector con;
+
+ int iter = options.iterations ();
+ int verb = options.verbose ();
+ int msiz = options.message_size ();
+ cout << "supplier starting connect" << endl;
+
+ if (con.connect (s_stream, serv_addr) == -1)
+ ACE_DEBUG ((LM_INFO, " (%t) connect failed\n"));
+
+ cout << "supplier : we're connected" << endl;
+ int n;
+ n = 0;
+ ACE_Message_Block * mb_p;
+
+ while (n < iter)
+ {
+ mb_p = new ACE_Message_Block (msiz);
+ ACE_OS::strcpy (mb_p->rd_ptr (), (char *) dummy);
+ mb_p->length (msiz);
+ if (verb)
+ cout << "supplier sending 1 message_block" << endl;
+ if (s_stream.send (mb_p) == -1)
+ {
+ cout << "supplier send failed" << endl;
+ return (void *) -1;
+ }
+ n++;
+ }
+
+ mb_p = new ACE_Message_Block (10);
+ mb_p->length (1);
+ *mb_p->rd_ptr () = 'g';
+
+ cout << "supplier sending last message_block" << endl;
+
+ if (s_stream.send (mb_p) == -1)
+ {
+ cout << "supplier send last mb failed" << endl;
+ return (void *) -1;
+ }
+ mb_p = new ACE_Message_Block (10);
+ mb_p->length (0);
+
+ if (verb)
+ cout << "supplier sending very last message_block" << endl;
+
+ if (s_stream.send (mb_p) == -1)
+ {
+ cout << "supplier send very last mb failed" << endl;
+ return (void *) -1;
+ }
+
+ ACE_OS::sleep (2);
+ cout << "supplier terminating" << endl;
+ return 0;
+}
+
+int
+ACE_TMAIN (int argc, ACE_TCHAR *argv[])
+{
+ ACE_Service_Config daemon;
+
+ options.parse_args (argc, argv);
+ options.start_timer ();
+
+ // Primary ACE_Stream for EVENT_SERVER application.
+ MT_Stream event_server;
+
+ // Enable graceful shutdowns....
+ Quit_Handler quit_handler;
+
+ // Create the modules..
+
+ MT_Module *sr = new MT_Module (ACE_TEXT ("Supplier_Router"),
+ new Supplier_Router (ACE_Thread_Manager::instance ()));
+ MT_Module *ea = new MT_Module (ACE_TEXT ("Event_Analyzer"),
+ new Event_Analyzer,
+ new Event_Analyzer);
+ MT_Module *cr = new MT_Module (ACE_TEXT ("Consumer_Router"),
+ 0, // 0 triggers the creation of a ACE_Thru_Task...
+ new Consumer_Router (ACE_Thread_Manager::instance ()));
+
+ // Push the modules onto the event_server stream.
+
+ if (event_server.push (sr) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("push (Supplier_Router)")), -1);
+
+ if (event_server.push (ea) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("push (Event_Analyzer)")), -1);
+
+ if (event_server.push (cr) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("push (Consumer_Router)")), -1);
+
+ // Set the high and low water marks appropriately.
+
+ int wm = options.low_water_mark ();
+
+ if (event_server.control (ACE_IO_Cntl_Msg::SET_LWM, &wm) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("push (setting low watermark)")), -1);
+
+ wm = options.high_water_mark ();
+ if (event_server.control (ACE_IO_Cntl_Msg::SET_HWM, &wm) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
+ ACE_TEXT ("push (setting high watermark)")), -1);
+
+ // spawn the two threads.
+
+ if (ACE_Thread_Manager::instance ()->spawn (ACE_THR_FUNC (consumer), (void *) 0,
+ THR_NEW_LWP | THR_DETACHED) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("spawn")), 1);
+
+ else if (ACE_Thread_Manager::instance ()->spawn (ACE_THR_FUNC (supplier), (void *) "hello",
+ THR_NEW_LWP | THR_DETACHED) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("spawn")), 1);
+
+ // Perform the main event loop waiting for the user to type ^C or to
+ // enter a line on the ACE_STDIN.
+
+ ACE_Reactor::instance ()->run_reactor_event_loop ();
+
+ ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("main exiting\n")));
+
+ return 0;
+}
+#else
+int
+ACE_TMAIN (int, ACE_TCHAR *[])
+{
+ ACE_ERROR_RETURN ((LM_ERROR,
+ ACE_TEXT ("test not defined for this platform\n")),
+ -1);
+}
+#endif /* ACE_HAS_THREADS */
View Lorry Controller Status