diff options
Diffstat (limited to 'examples/ASX/Message_Queue')
-rw-r--r-- | examples/ASX/Message_Queue/Bounded_Buffer.dsp | 91 | ||||
-rw-r--r-- | examples/ASX/Message_Queue/Buffer_Stream.dsp | 91 | ||||
-rw-r--r-- | examples/ASX/Message_Queue/Makefile | 43 | ||||
-rw-r--r-- | examples/ASX/Message_Queue/Message_Queue_Example.dsw | 53 | ||||
-rw-r--r-- | examples/ASX/Message_Queue/Priority_Buffer.dsp | 91 | ||||
-rw-r--r-- | examples/ASX/Message_Queue/bounded_buffer.cpp | 123 | ||||
-rw-r--r-- | examples/ASX/Message_Queue/buffer_stream.cpp | 273 | ||||
-rw-r--r-- | examples/ASX/Message_Queue/priority_buffer.cpp | 141 |
8 files changed, 0 insertions, 906 deletions
diff --git a/examples/ASX/Message_Queue/Bounded_Buffer.dsp b/examples/ASX/Message_Queue/Bounded_Buffer.dsp deleted file mode 100644 index 7a6b1448b7b..00000000000 --- a/examples/ASX/Message_Queue/Bounded_Buffer.dsp +++ /dev/null @@ -1,91 +0,0 @@ -# Microsoft Developer Studio Project File - Name="Bounded_Buffer" - Package Owner=<4>
-# Microsoft Developer Studio Generated Build File, Format Version 5.00
-# ** DO NOT EDIT **
-
-# TARGTYPE "Win32 (x86) Console Application" 0x0103
-
-CFG=Bounded_Buffer - Win32 Debug
-!MESSAGE This is not a valid makefile. To build this project using NMAKE,
-!MESSAGE use the Export Makefile command and run
-!MESSAGE
-!MESSAGE NMAKE /f "Bounded_Buffer.mak".
-!MESSAGE
-!MESSAGE You can specify a configuration when running NMAKE
-!MESSAGE by defining the macro CFG on the command line. For example:
-!MESSAGE
-!MESSAGE NMAKE /f "Bounded_Buffer.mak" CFG="Bounded_Buffer - Win32 Debug"
-!MESSAGE
-!MESSAGE Possible choices for configuration are:
-!MESSAGE
-!MESSAGE "Bounded_Buffer - Win32 Release" (based on\
- "Win32 (x86) Console Application")
-!MESSAGE "Bounded_Buffer - Win32 Debug" (based on\
- "Win32 (x86) Console Application")
-!MESSAGE
-
-# Begin Project
-# PROP Scc_ProjName ""
-# PROP Scc_LocalPath ""
-CPP=cl.exe
-RSC=rc.exe
-
-!IF "$(CFG)" == "Bounded_Buffer - Win32 Release"
-
-# PROP BASE Use_MFC 0
-# PROP BASE Use_Debug_Libraries 0
-# PROP BASE Output_Dir "Release"
-# PROP BASE Intermediate_Dir "Release"
-# PROP BASE Target_Dir ""
-# PROP Use_MFC 0
-# PROP Use_Debug_Libraries 0
-# PROP Output_Dir ".\"
-# PROP Intermediate_Dir ".\Release"
-# PROP Ignore_Export_Lib 0
-# PROP Target_Dir ""
-# ADD BASE CPP /nologo /W3 /GX /O2 /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD CPP /nologo /MD /W3 /GX /O2 /I "..\..\..\\" /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD BASE RSC /l 0x409 /d "NDEBUG"
-# ADD RSC /l 0x409 /d "NDEBUG"
-BSC32=bscmake.exe
-# ADD BASE BSC32 /nologo
-# ADD BSC32 /nologo
-LINK32=link.exe
-# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /machine:I386
-# ADD LINK32 ace.lib /nologo /subsystem:console /machine:I386 /out:".\Bounded_Buffer-r.exe" /libpath:"..\..\..\ace"
-
-!ELSEIF "$(CFG)" == "Bounded_Buffer - Win32 Debug"
-
-# PROP BASE Use_MFC 0
-# PROP BASE Use_Debug_Libraries 1
-# PROP BASE Output_Dir "Bounded_"
-# PROP BASE Intermediate_Dir "Bounded_"
-# PROP BASE Target_Dir ""
-# PROP Use_MFC 0
-# PROP Use_Debug_Libraries 1
-# PROP Output_Dir ".\"
-# PROP Intermediate_Dir ".\Debug"
-# PROP Ignore_Export_Lib 0
-# PROP Target_Dir ""
-# ADD BASE CPP /nologo /W3 /Gm /GX /Zi /Od /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD CPP /nologo /MDd /W3 /Gm /GX /Zi /Od /I "..\..\..\\" /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD BASE RSC /l 0x409 /d "_DEBUG"
-# ADD RSC /l 0x409 /d "_DEBUG"
-BSC32=bscmake.exe
-# ADD BASE BSC32 /nologo
-# ADD BSC32 /nologo
-LINK32=link.exe
-# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept
-# ADD LINK32 aced.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept /libpath:"..\..\..\ace"
-
-!ENDIF
-
-# Begin Target
-
-# Name "Bounded_Buffer - Win32 Release"
-# Name "Bounded_Buffer - Win32 Debug"
-# Begin Source File
-
-SOURCE=.\bounded_buffer.cpp
-# End Source File
-# End Target
-# End Project
diff --git a/examples/ASX/Message_Queue/Buffer_Stream.dsp b/examples/ASX/Message_Queue/Buffer_Stream.dsp deleted file mode 100644 index ca5bfaadd09..00000000000 --- a/examples/ASX/Message_Queue/Buffer_Stream.dsp +++ /dev/null @@ -1,91 +0,0 @@ -# Microsoft Developer Studio Project File - Name="Buffer_Stream" - Package Owner=<4>
-# Microsoft Developer Studio Generated Build File, Format Version 5.00
-# ** DO NOT EDIT **
-
-# TARGTYPE "Win32 (x86) Console Application" 0x0103
-
-CFG=Buffer_Stream - Win32 Debug
-!MESSAGE This is not a valid makefile. To build this project using NMAKE,
-!MESSAGE use the Export Makefile command and run
-!MESSAGE
-!MESSAGE NMAKE /f "Buffer_Stream.mak".
-!MESSAGE
-!MESSAGE You can specify a configuration when running NMAKE
-!MESSAGE by defining the macro CFG on the command line. For example:
-!MESSAGE
-!MESSAGE NMAKE /f "Buffer_Stream.mak" CFG="Buffer_Stream - Win32 Debug"
-!MESSAGE
-!MESSAGE Possible choices for configuration are:
-!MESSAGE
-!MESSAGE "Buffer_Stream - Win32 Release" (based on\
- "Win32 (x86) Console Application")
-!MESSAGE "Buffer_Stream - Win32 Debug" (based on\
- "Win32 (x86) Console Application")
-!MESSAGE
-
-# Begin Project
-# PROP Scc_ProjName ""
-# PROP Scc_LocalPath ""
-CPP=cl.exe
-RSC=rc.exe
-
-!IF "$(CFG)" == "Buffer_Stream - Win32 Release"
-
-# PROP BASE Use_MFC 0
-# PROP BASE Use_Debug_Libraries 0
-# PROP BASE Output_Dir "Release"
-# PROP BASE Intermediate_Dir "Release"
-# PROP BASE Target_Dir ""
-# PROP Use_MFC 0
-# PROP Use_Debug_Libraries 0
-# PROP Output_Dir ".\"
-# PROP Intermediate_Dir ".\Release"
-# PROP Ignore_Export_Lib 0
-# PROP Target_Dir ""
-# ADD BASE CPP /nologo /W3 /GX /O2 /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD CPP /nologo /MD /W3 /GX /O2 /I "..\..\..\\" /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD BASE RSC /l 0x409 /d "NDEBUG"
-# ADD RSC /l 0x409 /d "NDEBUG"
-BSC32=bscmake.exe
-# ADD BASE BSC32 /nologo
-# ADD BSC32 /nologo
-LINK32=link.exe
-# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /machine:I386
-# ADD LINK32 ace.lib /nologo /subsystem:console /machine:I386 /out:".\Buffer_Stream-r.exe" /libpath:"..\..\..\ace"
-
-!ELSEIF "$(CFG)" == "Buffer_Stream - Win32 Debug"
-
-# PROP BASE Use_MFC 0
-# PROP BASE Use_Debug_Libraries 1
-# PROP BASE Output_Dir "Buffer_S"
-# PROP BASE Intermediate_Dir "Buffer_S"
-# PROP BASE Target_Dir ""
-# PROP Use_MFC 0
-# PROP Use_Debug_Libraries 1
-# PROP Output_Dir ".\"
-# PROP Intermediate_Dir ".\Debug"
-# PROP Ignore_Export_Lib 0
-# PROP Target_Dir ""
-# ADD BASE CPP /nologo /W3 /Gm /GX /Zi /Od /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD CPP /nologo /MDd /W3 /Gm /GX /Zi /Od /I "..\..\..\\" /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD BASE RSC /l 0x409 /d "_DEBUG"
-# ADD RSC /l 0x409 /d "_DEBUG"
-BSC32=bscmake.exe
-# ADD BASE BSC32 /nologo
-# ADD BSC32 /nologo
-LINK32=link.exe
-# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept
-# ADD LINK32 aced.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept /libpath:"..\..\..\ace"
-
-!ENDIF
-
-# Begin Target
-
-# Name "Buffer_Stream - Win32 Release"
-# Name "Buffer_Stream - Win32 Debug"
-# Begin Source File
-
-SOURCE=.\buffer_stream.cpp
-# End Source File
-# End Target
-# End Project
diff --git a/examples/ASX/Message_Queue/Makefile b/examples/ASX/Message_Queue/Makefile deleted file mode 100644 index 7961a617985..00000000000 --- a/examples/ASX/Message_Queue/Makefile +++ /dev/null @@ -1,43 +0,0 @@ -#---------------------------------------------------------------------------- -# $Id: Makefile 1.1 10/18/96 -# -# Makefile for Message_Queue tests -#---------------------------------------------------------------------------- - -#---------------------------------------------------------------------------- -# Local macros -#---------------------------------------------------------------------------- - -BIN = buffer_stream \ - bounded_buffer \ - priority_buffer - -VLDLIBS = $(LDLIBS:%=%$(VAR)) - -BUILD = $(VBIN) - -#---------------------------------------------------------------------------- -# Include macros and targets -#---------------------------------------------------------------------------- - -include $(ACE_ROOT)/include/makeinclude/wrapper_macros.GNU -include $(ACE_ROOT)/include/makeinclude/macros.GNU -include $(ACE_ROOT)/include/makeinclude/rules.common.GNU -include $(ACE_ROOT)/include/makeinclude/rules.nonested.GNU -include $(ACE_ROOT)/include/makeinclude/rules.bin.GNU -include $(ACE_ROOT)/include/makeinclude/rules.local.GNU - -#---------------------------------------------------------------------------- -# Local targets -#---------------------------------------------------------------------------- - -#---------------------------------------------------------------------------- -# Dependencies -#---------------------------------------------------------------------------- - -# DO NOT DELETE THIS LINE -- g++dep uses it. -# DO NOT PUT ANYTHING AFTER THIS LINE, IT WILL GO AWAY. - - - -# IF YOU PUT ANYTHING HERE IT WILL GO AWAY diff --git a/examples/ASX/Message_Queue/Message_Queue_Example.dsw b/examples/ASX/Message_Queue/Message_Queue_Example.dsw deleted file mode 100644 index 02c411a4b73..00000000000 --- a/examples/ASX/Message_Queue/Message_Queue_Example.dsw +++ /dev/null @@ -1,53 +0,0 @@ -Microsoft Developer Studio Workspace File, Format Version 5.00
-# WARNING: DO NOT EDIT OR DELETE THIS WORKSPACE FILE!
-
-###############################################################################
-
-Project: "Bounded_Buffer"=.\Bounded_Buffer.dsp - Package Owner=<4>
-
-Package=<5>
-{{{
-}}}
-
-Package=<4>
-{{{
-}}}
-
-###############################################################################
-
-Project: "Buffer_Stream"=.\Buffer_Stream.dsp - Package Owner=<4>
-
-Package=<5>
-{{{
-}}}
-
-Package=<4>
-{{{
-}}}
-
-###############################################################################
-
-Project: "Priority_Buffer"=.\Priority_Buffer.dsp - Package Owner=<4>
-
-Package=<5>
-{{{
-}}}
-
-Package=<4>
-{{{
-}}}
-
-###############################################################################
-
-Global:
-
-Package=<5>
-{{{
-}}}
-
-Package=<3>
-{{{
-}}}
-
-###############################################################################
-
diff --git a/examples/ASX/Message_Queue/Priority_Buffer.dsp b/examples/ASX/Message_Queue/Priority_Buffer.dsp deleted file mode 100644 index 069bcdddba2..00000000000 --- a/examples/ASX/Message_Queue/Priority_Buffer.dsp +++ /dev/null @@ -1,91 +0,0 @@ -# Microsoft Developer Studio Project File - Name="Priority_Buffer" - Package Owner=<4>
-# Microsoft Developer Studio Generated Build File, Format Version 5.00
-# ** DO NOT EDIT **
-
-# TARGTYPE "Win32 (x86) Console Application" 0x0103
-
-CFG=Priority_Buffer - Win32 Debug
-!MESSAGE This is not a valid makefile. To build this project using NMAKE,
-!MESSAGE use the Export Makefile command and run
-!MESSAGE
-!MESSAGE NMAKE /f "Priority_Buffer.mak".
-!MESSAGE
-!MESSAGE You can specify a configuration when running NMAKE
-!MESSAGE by defining the macro CFG on the command line. For example:
-!MESSAGE
-!MESSAGE NMAKE /f "Priority_Buffer.mak" CFG="Priority_Buffer - Win32 Debug"
-!MESSAGE
-!MESSAGE Possible choices for configuration are:
-!MESSAGE
-!MESSAGE "Priority_Buffer - Win32 Release" (based on\
- "Win32 (x86) Console Application")
-!MESSAGE "Priority_Buffer - Win32 Debug" (based on\
- "Win32 (x86) Console Application")
-!MESSAGE
-
-# Begin Project
-# PROP Scc_ProjName ""
-# PROP Scc_LocalPath ""
-CPP=cl.exe
-RSC=rc.exe
-
-!IF "$(CFG)" == "Priority_Buffer - Win32 Release"
-
-# PROP BASE Use_MFC 0
-# PROP BASE Use_Debug_Libraries 0
-# PROP BASE Output_Dir "Release"
-# PROP BASE Intermediate_Dir "Release"
-# PROP BASE Target_Dir ""
-# PROP Use_MFC 0
-# PROP Use_Debug_Libraries 0
-# PROP Output_Dir ".\"
-# PROP Intermediate_Dir ".\Release"
-# PROP Ignore_Export_Lib 0
-# PROP Target_Dir ""
-# ADD BASE CPP /nologo /W3 /GX /O2 /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD CPP /nologo /MD /W3 /GX /O2 /I "..\..\..\\" /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD BASE RSC /l 0x409 /d "NDEBUG"
-# ADD RSC /l 0x409 /d "NDEBUG"
-BSC32=bscmake.exe
-# ADD BASE BSC32 /nologo
-# ADD BSC32 /nologo
-LINK32=link.exe
-# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /machine:I386
-# ADD LINK32 ace.lib /nologo /subsystem:console /machine:I386 /out:".\Priority_Buffer-r.exe" /libpath:"..\..\..\ace"
-
-!ELSEIF "$(CFG)" == "Priority_Buffer - Win32 Debug"
-
-# PROP BASE Use_MFC 0
-# PROP BASE Use_Debug_Libraries 1
-# PROP BASE Output_Dir "Priority"
-# PROP BASE Intermediate_Dir "Priority"
-# PROP BASE Target_Dir ""
-# PROP Use_MFC 0
-# PROP Use_Debug_Libraries 1
-# PROP Output_Dir ".\"
-# PROP Intermediate_Dir ".\Debug"
-# PROP Ignore_Export_Lib 0
-# PROP Target_Dir ""
-# ADD BASE CPP /nologo /W3 /Gm /GX /Zi /Od /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD CPP /nologo /MDd /W3 /Gm /GX /Zi /Od /I "..\..\..\\" /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
-# ADD BASE RSC /l 0x409 /d "_DEBUG"
-# ADD RSC /l 0x409 /d "_DEBUG"
-BSC32=bscmake.exe
-# ADD BASE BSC32 /nologo
-# ADD BSC32 /nologo
-LINK32=link.exe
-# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept
-# ADD LINK32 aced.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept /libpath:"..\..\..\ace"
-
-!ENDIF
-
-# Begin Target
-
-# Name "Priority_Buffer - Win32 Release"
-# Name "Priority_Buffer - Win32 Debug"
-# Begin Source File
-
-SOURCE=.\priority_buffer.cpp
-# End Source File
-# End Target
-# End Project
diff --git a/examples/ASX/Message_Queue/bounded_buffer.cpp b/examples/ASX/Message_Queue/bounded_buffer.cpp deleted file mode 100644 index e0ea691d433..00000000000 --- a/examples/ASX/Message_Queue/bounded_buffer.cpp +++ /dev/null @@ -1,123 +0,0 @@ -// $Id$ - -// This short program copies stdin to stdout via the use of an ASX -// Message_Queue. It illustrates an implementation of the classic -// "bounded buffer" program. - -#include "ace/Message_Queue.h" -#include "ace/Thread_Manager.h" - -#if defined (ACE_HAS_THREADS) - -// Global thread manager. -static ACE_Thread_Manager thr_mgr; - -// The producer reads data from the stdin stream, creates a message, -// and then queues the message in the message list, where it is -// removed by the consumer thread. A 0-sized message is enqueued when -// there is no more data to read. The consumer uses this as a flag to -// know when to exit. - -static void * -producer (ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue) -{ - // Keep reading stdin, until we reach EOF. - - for (int n; ; ) - { - // Allocate a new message. - ACE_Message_Block *mb; - - ACE_NEW_RETURN (mb, ACE_Message_Block (BUFSIZ), 0); - - n = ACE_OS::read (ACE_STDIN, mb->rd_ptr (), mb->size ()); - - if (n <= 0) - { - // Send a shutdown message to the other thread and exit. - mb->length (0); - if (msg_queue->enqueue_tail (mb) == -1) - ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next")); - break; - } - - // Send the message to the other thread. - else - { - mb->msg_priority (n); - mb->wr_ptr (n); - if (msg_queue->enqueue_tail (mb) == -1) - ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next")); - } - } - - return 0; -} - -// The consumer dequeues a message from the ACE_Message_Queue, writes -// the message to the stderr stream, and deletes the message. The -// producer sends a 0-sized message to inform the consumer to stop -// reading and exit. - -static void *consumer (ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue) -{ - int result = 0; - - // Keep looping, reading a message out of the queue, until we timeout - // or get a message with a length == 0, which signals us to quit. - - for (;;) - { - ACE_Message_Block *mb; - - ACE_Time_Value timeout (ACE_OS::time (0) + 4, 0); // Wait for upto 4 seconds - - result = msg_queue->dequeue_head (mb, &timeout); - - if (result == -1) - break; - - int length = mb->length (); - - if (length > 0) - ACE_OS::write (ACE_STDOUT, mb->rd_ptr (), length); - - mb->release (); - - if (length == 0) - break; - } - - if (result == -1 && errno == EWOULDBLOCK) - ACE_ERROR ((LM_ERROR, "(%t) %p\n%a", "timed out waiting for message", 1)); - - return 0; -} - -// Spawn off two threads that copy stdin to stdout. - -int -main (int, char *[]) -{ - // Message list. - ACE_Message_Queue<ACE_MT_SYNCH> msg_queue; - - if (thr_mgr.spawn (ACE_THR_FUNC (producer), (void *) &msg_queue, - THR_NEW_LWP | THR_DETACHED) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 1); - else if (thr_mgr.spawn (ACE_THR_FUNC (consumer), (void *) &msg_queue, - THR_NEW_LWP | THR_DETACHED) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 1); - - // Wait for producer and consumer threads to exit. - thr_mgr.wait (); - return 0; -} -#else -int -main (int, char *[]) -{ - ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n")); - return 0; -} -#endif /* ACE_HAS_THREADS */ diff --git a/examples/ASX/Message_Queue/buffer_stream.cpp b/examples/ASX/Message_Queue/buffer_stream.cpp deleted file mode 100644 index 4d8c86b9703..00000000000 --- a/examples/ASX/Message_Queue/buffer_stream.cpp +++ /dev/null @@ -1,273 +0,0 @@ -// $Id$ - -// This short program copies stdin to stdout via the use of an ASX -// Stream. It illustrates an implementation of the classic "bounded -// buffer" program using an ASX Stream containing two Modules. Each -// ACE_Module contains two Tasks. Each ACE_Task contains a -// ACE_Message_Queue and a pointer to a ACE_Thread_Manager. Note how -// the use of these reusable components reduces the reliance on global -// variables, as compared with the bounded_buffer.C example. - -#include "ace/Synch.h" -#include "ace/Service_Config.h" -#include "ace/Stream.h" -#include "ace/Module.h" -#include "ace/Task.h" - -#if defined (ACE_HAS_THREADS) - -typedef ACE_Stream<ACE_MT_SYNCH> MT_Stream; -typedef ACE_Module<ACE_MT_SYNCH> MT_Module; -typedef ACE_Task<ACE_MT_SYNCH> MT_Task; - -class Common_Task : public MT_Task - // = TITLE - // Methods that are common to the producer and consumer. -{ -public: - Common_Task (void) {} - // ACE_Task hooks - virtual int open (void * = 0); - virtual int close (u_long = 0); -}; - -// Define the Producer interface. - -class Producer : public Common_Task -{ -public: - Producer (void) {} - - // Read data from stdin and pass to consumer. - virtual int svc (void); -}; - -class Consumer : public Common_Task - // = TITLE - // Define the Consumer interface. -{ -public: - Consumer (void) {} - - // Enqueue the message on the ACE_Message_Queue for subsequent - // handling in the svc() method. - virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0); - - // Receive message from producer and print to stdout. - virtual int svc (void); - -private: - - ACE_Time_Value timeout_; -}; - -class Filter : public MT_Task - // = TITLE - // Defines a Filter that prepends a line number in front of each - // line. -{ -public: - Filter (void): count_ (1) {} - - // Change the size of the message before passing it downstream. - virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0); - -private: - size_t count_; - // Count the number of lines passing through the filter. -}; - -// Spawn off a new thread. - -int -Common_Task::open (void *) -{ - if (this->activate (THR_NEW_LWP | THR_DETACHED) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), -1); - return 0; -} - -int -Common_Task::close (u_long exit_status) -{ - ACE_DEBUG ((LM_DEBUG, "(%t) thread is exiting with status %d in module %s\n", - exit_status, this->name ())); - - // Can do anything here that is required when a thread exits, e.g., - // storing thread-specific information in some other storage - // location, etc. - return 0; -} - -// The Consumer reads data from the stdin stream, creates a message, -// and then queues the message in the message list, where it is -// removed by the consumer thread. A 0-sized message is enqueued when -// there is no more data to read. The consumer uses this as a flag to -// know when to exit. - -int -Producer::svc (void) -{ - // Keep reading stdin, until we reach EOF. - - for (int n; ; ) - { - // Allocate a new message (add one to avoid nasty boundary - // conditions). - - ACE_Message_Block *mb; - - ACE_NEW_RETURN (mb, ACE_Message_Block (BUFSIZ + 1), -1); - - n = ACE_OS::read (ACE_STDIN, mb->rd_ptr (), mb->size ()); - - if (n <= 0) - { - // Send a shutdown message to the other thread and exit. - mb->length (0); - - if (this->put_next (mb) == -1) - ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next")); - break; - } - - // Send the message to the other thread. - else - { - mb->wr_ptr (n); - // NUL-terminate the string (since we use strlen() on it - // later). - mb->rd_ptr ()[n] = '\0'; - - if (this->put_next (mb) == -1) - ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next")); - } - } - - return 0; -} - -// Simply enqueue the Message_Block into the end of the queue. - -int -Consumer::put (ACE_Message_Block *mb, ACE_Time_Value *tv) -{ - return this->putq (mb, tv); -} - -// The consumer dequeues a message from the ACE_Message_Queue, writes -// the message to the stderr stream, and deletes the message. The -// Consumer sends a 0-sized message to inform the consumer to stop -// reading and exit. - -int -Consumer::svc (void) -{ - int result = 0; - - // Keep looping, reading a message out of the queue, until we - // timeout or get a message with a length == 0, which signals us to - // quit. - - for (;;) - { - ACE_Message_Block *mb; - - this->timeout_.sec (ACE_OS::time (0) + 4); // Wait for upto 4 seconds - - result = this->getq (mb, &this->timeout_); - - if (result == -1) - break; - - int length = mb->length (); - - if (length > 0) - ACE_OS::write (ACE_STDOUT, - mb->rd_ptr (), - ACE_OS::strlen (mb->rd_ptr ())); - - mb->release (); - - if (length == 0) - break; - } - - if (result == -1 && errno == EWOULDBLOCK) - ACE_ERROR ((LM_ERROR, - "(%t) %p\n%a", - "timed out waiting for message", - 1)); - return 0; -} - -// The filter prepends a line number in front of each line. - -int -Filter::put (ACE_Message_Block *mb, - ACE_Time_Value *tv) -{ - if (mb->length () == 0) - return this->put_next (mb, tv); - else - { - char buf[BUFSIZ]; - - // Stash a copy of the buffer away. - ACE_OS::strncpy (buf, mb->rd_ptr (), sizeof buf); - - // Increase the size of the buffer large enough that it will be - // reallocated (in order to test the reallocation mechanisms). - - mb->size (mb->length () + BUFSIZ); - mb->length (mb->size ()); - - // Prepend the line count in front of the buffer. - ACE_OS::sprintf (mb->rd_ptr (), "%d: %s", - this->count_++, buf); - - return this->put_next (mb, tv); - } -} - -// Main driver function. - -int -main (int, char *argv[]) -{ - ACE_Service_Config daemon (argv[0]); - - // This Stream controls hierachically-related active objects. - MT_Stream stream; - - MT_Module *pm; - MT_Module *fm; - MT_Module *cm; - - ACE_NEW_RETURN (pm, MT_Module ("Consumer", new Consumer), -1); - ACE_NEW_RETURN (fm, MT_Module ("Filter", new Filter), -1); - ACE_NEW_RETURN (cm, MT_Module ("Producer", new Producer), -1); - - // Create Producer and Consumer Modules and push them onto the - // Stream. All processing is performed in the Stream. - - if (stream.push (pm) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), 1); - else if (stream.push (fm) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), 1); - else if (stream.push (cm) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), 1); - - // Barrier synchronization: wait for the threads to exit, then exit - // ourselves. - ACE_Thread_Manager::instance ()->wait (); - return 0; -} -#else -int -main (int, char *[]) -{ - ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n")); - return 0; -} -#endif /* ACE_HAS_THREADS */ diff --git a/examples/ASX/Message_Queue/priority_buffer.cpp b/examples/ASX/Message_Queue/priority_buffer.cpp deleted file mode 100644 index 52976df02d5..00000000000 --- a/examples/ASX/Message_Queue/priority_buffer.cpp +++ /dev/null @@ -1,141 +0,0 @@ -// $Id$ - -// This short program prints the contents of stdin to stdout sorted by -// the length of each line via the use of an ASX Message_Queue. It -// illustrates how priorities can be used for ACE Message_Queues. - -#include "ace/Message_Queue.h" -#include "ace/Read_Buffer.h" -#include "ace/Thread_Manager.h" -#include "ace/Service_Config.h" - -#if defined (ACE_HAS_THREADS) - -// Global thread manager. -static ACE_Thread_Manager thr_mgr; - -// Make the queue be capable of being *very* large. -static const long max_queue = LONG_MAX; - -// The consumer dequeues a message from the ACE_Message_Queue, writes -// the message to the stderr stream, and deletes the message. The -// producer sends a 0-sized message to inform the consumer to stop -// reading and exit. - -static void * -consumer (ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue) -{ - // Keep looping, reading a message out of the queue, until we - // timeout or get a message with a length == 0, which signals us to - // quit. - - for (;;) - { - ACE_Message_Block *mb; - - if (msg_queue->dequeue_head (mb) == -1) - break; - - int length = mb->length (); - - if (length > 0) - ACE_OS::puts (mb->rd_ptr ()); - - // Free up the buffer memory and the Message_Block. - ACE_Allocator::instance ()->free (mb->rd_ptr ()); - mb->release (); - - if (length == 0) - break; - } - - return 0; -} - -// The producer reads data from the stdin stream, creates a message, -// and then queues the message in the message list, where it is -// removed by the consumer thread. A 0-sized message is enqueued when -// there is no more data to read. The consumer uses this as a flag to -// know when to exit. - -static void * -producer (ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue) -{ - ACE_Read_Buffer rb (ACE_STDIN); - - // Keep reading stdin, until we reach EOF. - - for (;;) - { - // Allocate a new buffer. - char *buffer = rb.read ('\n'); - - ACE_Message_Block *mb; - - if (buffer == 0) - { - // Send a 0-sized shutdown message to the other thread and - // exit. - - ACE_NEW_RETURN (mb, ACE_Message_Block ((size_t) 0), 0); - - if (msg_queue->enqueue_tail (mb) == -1) - ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next")); - break; - } - - // Enqueue the message in priority order. - else - { - // Allocate a new message, but have it "borrow" its memory - // from the buffer. - ACE_NEW_RETURN (mb, ACE_Message_Block (rb.size (), - ACE_Message_Block::MB_DATA, - 0, - buffer), - 0); - mb->msg_priority (rb.size ()); - mb->wr_ptr (rb.size ()); - - ACE_DEBUG ((LM_DEBUG, - "enqueueing message of size %d\n", - mb->msg_priority ())); - - // Enqueue in priority order. - if (msg_queue->enqueue_prio (mb) == -1) - ACE_ERROR ((LM_ERROR, "(%t) %p\n", "put_next")); - } - } - - // Now read all the items out in priority order (i.e., ordered by - // the size of the lines!). - consumer (msg_queue); - - return 0; -} - -// Spawn off one thread that copies stdin to stdout in order of the -// size of each line. - -int -main (int, char *[]) -{ - // Message queue. - ACE_Message_Queue<ACE_MT_SYNCH> msg_queue (max_queue); - - if (thr_mgr.spawn (ACE_THR_FUNC (producer), (void *) &msg_queue, - THR_NEW_LWP | THR_DETACHED) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 1); - - // Wait for producer and consumer threads to exit. - thr_mgr.wait (); - return 0; -} -#else -int -main (int, char *[]) -{ - ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n")); - return 0; -} -#endif /* ACE_HAS_THREADS */ |