summaryrefslogtreecommitdiff
path: root/examples/Reactor/Proactor
diff options
context:
space:
mode:
Diffstat (limited to 'examples/Reactor/Proactor')
-rw-r--r--examples/Reactor/Proactor/test_multiple_loops.cpp5
-rw-r--r--examples/Reactor/Proactor/test_proactor.cpp100
-rw-r--r--examples/Reactor/Proactor/test_proactor.mak97
-rw-r--r--examples/Reactor/Proactor/test_proactor.mdpbin52736 -> 53248 bytes
4 files changed, 133 insertions, 69 deletions
diff --git a/examples/Reactor/Proactor/test_multiple_loops.cpp b/examples/Reactor/Proactor/test_multiple_loops.cpp
index 9ffbf72f596..c13ebfce3eb 100644
--- a/examples/Reactor/Proactor/test_multiple_loops.cpp
+++ b/examples/Reactor/Proactor/test_multiple_loops.cpp
@@ -110,10 +110,5 @@ main (int, char *[])
ACE_ERROR_RETURN ((LM_ERROR, "%p.\n", "main"), -1);
ACE_Thread_Manager::instance ()->wait ();
-
- // Remove from reactor
- ACE_Reactor::instance ()->remove_handler (&proactor,
- ACE_Event_Handler::DONT_CALL);
-
return 0;
}
diff --git a/examples/Reactor/Proactor/test_proactor.cpp b/examples/Reactor/Proactor/test_proactor.cpp
index f6ebed26c53..4582fb82888 100644
--- a/examples/Reactor/Proactor/test_proactor.cpp
+++ b/examples/Reactor/Proactor/test_proactor.cpp
@@ -34,7 +34,6 @@ static u_short port = ACE_DEFAULT_SERVER_PORT;
static char *file = "test_proactor.cpp";
static char *dump_file = "output";
static int done = 0;
-static int initial_read_size = BUFSIZ;
class Receiver : public ACE_Service_Handler
//
@@ -130,18 +129,13 @@ Receiver::open (ACE_HANDLE handle,
return;
}
- // Duplicate the message block so that we can keep it around
- ACE_Message_Block &duplicate = *message_block.duplicate ();
-
- // Initial data (data which came with the AcceptEx call)
- ACE_Asynch_Read_Stream::Result fake_result (*this,
- this->handle_,
- duplicate,
- initial_read_size,
- 0,
- ACE_INVALID_HANDLE);
- // This will call the callback
- fake_result.complete (message_block.length (), 1, 0);
+ // Print any initial data which came with the AcceptEx call
+ message_block.rd_ptr ()[message_block.length ()] = '\0';
+ ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "Initial data", message_block.rd_ptr ()));
+
+ // Initiate new read from the stream
+ if (this->initiate_read_stream () == -1)
+ return;
}
int
@@ -165,7 +159,7 @@ Receiver::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
ACE_DEBUG ((LM_DEBUG, "handle_read_stream called\n"));
// Reset pointers
- result.message_block ().rd_ptr ()[result.bytes_transferred ()] = '\0';
+ result.message_block ().rd_ptr ()[result.message_block ().length ()] = '\0';
ACE_DEBUG ((LM_DEBUG, "********************\n"));
ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_read", result.bytes_to_read ()));
@@ -182,7 +176,7 @@ Receiver::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
{
// Successful read: Write the data to the file.
if (this->wf_.write (result.message_block (),
- result.bytes_transferred (),
+ result.message_block ().length (),
this->file_offset_) == -1)
{
ACE_ERROR ((LM_ERROR, "%p\n", "ACE_Asynch_Write_File::write"));
@@ -217,10 +211,6 @@ Receiver::handle_write_file (const ACE_Asynch_Write_File::Result &result)
if (result.success ())
// Write successful: Increment file offset
this->file_offset_ += result.bytes_transferred ();
-
- // This code is not robust enough to deal with short file writes
- // (which hardly ever happen) ;-)
- ACE_ASSERT (result.bytes_to_write () == result.bytes_transferred ());
}
class Sender : public ACE_Handler
@@ -351,6 +341,22 @@ Sender::open (const char *host,
return 0;
}
+int
+Sender::initiate_read_file (void)
+{
+ // Create Message_Block
+ ACE_Message_Block *mb = 0;
+ ACE_NEW_RETURN (mb, ACE_Message_Block (BUFSIZ + 1), -1);
+
+ // Inititiate an asynchronous read from the file
+ if (this->rf_.read (*mb,
+ mb->size () - 1,
+ this->file_offset_) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "ACE_Asynch_Read_File::read"), -1);
+
+ return 0;
+}
+
int
Sender::transmit_file (void)
{
@@ -410,28 +416,12 @@ Sender::handle_transmit_file (const ACE_Asynch_Transmit_File::Result &result)
done = 1;
}
-int
-Sender::initiate_read_file (void)
-{
- // Create Message_Block
- ACE_Message_Block *mb = 0;
- ACE_NEW_RETURN (mb, ACE_Message_Block (BUFSIZ + 1), -1);
-
- // Inititiate an asynchronous read from the file
- if (this->rf_.read (*mb,
- mb->size () - 1,
- this->file_offset_) == -1)
- ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "ACE_Asynch_Read_File::read"), -1);
-
- return 0;
-}
-
void
Sender::handle_read_file (const ACE_Asynch_Read_File::Result &result)
{
ACE_DEBUG ((LM_DEBUG, "handle_read_file called\n"));
- result.message_block ().rd_ptr ()[result.bytes_transferred ()] = '\0';
+ result.message_block ().rd_ptr ()[result.message_block ().length ()] = '\0';
ACE_DEBUG ((LM_DEBUG, "********************\n"));
ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_read", result.bytes_to_read ()));
@@ -449,7 +439,7 @@ Sender::handle_read_file (const ACE_Asynch_Read_File::Result &result)
// Read successful: increment offset and write data to network
this->file_offset_ += result.bytes_transferred ();
if (this->ws_.write (result.message_block (),
- result.bytes_transferred ()) == -1)
+ result.message_block ().length ()) == -1)
{
ACE_ERROR ((LM_ERROR, "%p\n", "ACE_Asynch_Write_Stream::write"));
return;
@@ -483,33 +473,15 @@ Sender::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
ACE_DEBUG ((LM_DEBUG, "********************\n"));
ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "message_block", result.message_block ().rd_ptr ()));
- if (result.success ())
- {
- // Partial write to socket
- int unsent_data = result.bytes_to_write () - result.bytes_transferred ();
- if (unsent_data != 0)
- {
- // Reset pointers
- result.message_block ().rd_ptr (result.bytes_transferred ());
-
- // Duplicate the message block and retry remaining data
- if (this->ws_.write (*result.message_block ().duplicate (),
- unsent_data) == -1)
- {
- ACE_ERROR ((LM_ERROR, "%p\n", "ACE_Asynch_Write_Stream::write"));
- return;
- }
- }
- else if (!(this->file_size_ > this->file_offset_))
- {
- this->stream_write_done_ = 1;
- if (this->transmit_file_done_)
- done = 1;
- }
- }
-
- // Release message block
result.message_block ().release ();
+
+ if (result.success ())
+ if (!(this->file_size_ > this->file_offset_))
+ {
+ this->stream_write_done_ = 1;
+ if (this->transmit_file_done_)
+ done = 1;
+ }
}
static int
@@ -560,7 +532,7 @@ main (int argc, char *argv[])
if (host == 0)
{
if (acceptor.open (ACE_INET_Addr (port),
- initial_read_size,
+ BUFSIZ,
1) == -1)
return -1;
}
diff --git a/examples/Reactor/Proactor/test_proactor.mak b/examples/Reactor/Proactor/test_proactor.mak
index 6a77e8c8547..5c011346451 100644
--- a/examples/Reactor/Proactor/test_proactor.mak
+++ b/examples/Reactor/Proactor/test_proactor.mak
@@ -245,22 +245,119 @@ CPP_PROJ=/nologo /MDd /W3 /Gm /GX /Zi /Od /D "WIN32" /D "_DEBUG" /D "_CONSOLE"\
SOURCE=.\test_proactor.cpp
DEP_CPP_TEST_=\
+ "..\..\..\ace\config-win32.h"\
{$(INCLUDE)}"\ace\ACE.h"\
+ {$(INCLUDE)}"\ace\ACE.i"\
+ {$(INCLUDE)}"\ace\Addr.h"\
+ {$(INCLUDE)}"\ace\Addr.i"\
+ {$(INCLUDE)}"\ace\Asynch_Acceptor.cpp"\
{$(INCLUDE)}"\ace\Asynch_Acceptor.h"\
{$(INCLUDE)}"\ace\Asynch_Acceptor.i"\
{$(INCLUDE)}"\ace\Asynch_IO.h"\
+ {$(INCLUDE)}"\ace\Asynch_IO.i"\
+ {$(INCLUDE)}"\ace\Atomic_Op.i"\
+ {$(INCLUDE)}"\ace\Auto_Ptr.cpp"\
+ {$(INCLUDE)}"\ace\Auto_Ptr.h"\
+ {$(INCLUDE)}"\ace\Auto_Ptr.i"\
+ {$(INCLUDE)}"\ace\config-win32-common.h"\
+ {$(INCLUDE)}"\ace\config.h"\
+ {$(INCLUDE)}"\ace\Containers.cpp"\
+ {$(INCLUDE)}"\ace\Containers.h"\
+ {$(INCLUDE)}"\ace\Containers.i"\
+ {$(INCLUDE)}"\ace\Event_Handler.h"\
+ {$(INCLUDE)}"\ace\Event_Handler.i"\
+ {$(INCLUDE)}"\ace\Free_List.cpp"\
+ {$(INCLUDE)}"\ace\Free_List.h"\
+ {$(INCLUDE)}"\ace\Free_List.i"\
{$(INCLUDE)}"\ace\Get_Opt.h"\
{$(INCLUDE)}"\ace\Get_Opt.i"\
+ {$(INCLUDE)}"\ace\Handle_Set.h"\
+ {$(INCLUDE)}"\ace\Handle_Set.i"\
+ {$(INCLUDE)}"\ace\High_Res_Timer.h"\
+ {$(INCLUDE)}"\ace\High_Res_Timer.i"\
{$(INCLUDE)}"\ace\INET_Addr.h"\
+ {$(INCLUDE)}"\ace\INET_Addr.i"\
+ {$(INCLUDE)}"\ace\IPC_SAP.h"\
+ {$(INCLUDE)}"\ace\IPC_SAP.i"\
+ {$(INCLUDE)}"\ace\Log_Msg.h"\
+ {$(INCLUDE)}"\ace\Log_Priority.h"\
+ {$(INCLUDE)}"\ace\Log_Record.h"\
+ {$(INCLUDE)}"\ace\Log_Record.i"\
+ {$(INCLUDE)}"\ace\Malloc.h"\
+ {$(INCLUDE)}"\ace\Malloc.i"\
+ {$(INCLUDE)}"\ace\Malloc_T.cpp"\
+ {$(INCLUDE)}"\ace\Malloc_T.h"\
+ {$(INCLUDE)}"\ace\Malloc_T.i"\
+ {$(INCLUDE)}"\ace\Managed_Object.cpp"\
+ {$(INCLUDE)}"\ace\Managed_Object.h"\
+ {$(INCLUDE)}"\ace\Managed_Object.i"\
+ {$(INCLUDE)}"\ace\Mem_Map.h"\
+ {$(INCLUDE)}"\ace\Mem_Map.i"\
+ {$(INCLUDE)}"\ace\Memory_Pool.h"\
+ {$(INCLUDE)}"\ace\Memory_Pool.i"\
{$(INCLUDE)}"\ace\Message_Block.h"\
+ {$(INCLUDE)}"\ace\Message_Block.i"\
+ {$(INCLUDE)}"\ace\Object_Manager.h"\
+ {$(INCLUDE)}"\ace\Object_Manager.i"\
{$(INCLUDE)}"\ace\OS.h"\
+ {$(INCLUDE)}"\ace\OS.i"\
{$(INCLUDE)}"\ace\Proactor.h"\
+ {$(INCLUDE)}"\ace\Proactor.i"\
+ {$(INCLUDE)}"\ace\Reactor.h"\
+ {$(INCLUDE)}"\ace\Reactor.i"\
+ {$(INCLUDE)}"\ace\Reactor_Impl.h"\
{$(INCLUDE)}"\ace\Service_Config.h"\
+ {$(INCLUDE)}"\ace\Service_Config.i"\
+ {$(INCLUDE)}"\ace\Service_Object.h"\
+ {$(INCLUDE)}"\ace\Service_Object.i"\
+ {$(INCLUDE)}"\ace\Shared_Object.h"\
+ {$(INCLUDE)}"\ace\Shared_Object.i"\
+ {$(INCLUDE)}"\ace\Signal.h"\
+ {$(INCLUDE)}"\ace\Signal.i"\
+ {$(INCLUDE)}"\ace\SOCK.h"\
+ {$(INCLUDE)}"\ace\SOCK.i"\
{$(INCLUDE)}"\ace\SOCK_Acceptor.h"\
{$(INCLUDE)}"\ace\SOCK_Acceptor.i"\
{$(INCLUDE)}"\ace\SOCK_Connector.h"\
+ {$(INCLUDE)}"\ace\SOCK_Connector.i"\
+ {$(INCLUDE)}"\ace\SOCK_IO.h"\
+ {$(INCLUDE)}"\ace\SOCK_IO.i"\
{$(INCLUDE)}"\ace\SOCK_Stream.h"\
+ {$(INCLUDE)}"\ace\SOCK_Stream.i"\
+ {$(INCLUDE)}"\ace\SString.h"\
+ {$(INCLUDE)}"\ace\SString.i"\
+ {$(INCLUDE)}"\ace\stdcpp.h"\
+ {$(INCLUDE)}"\ace\SV_Semaphore_Complex.h"\
+ {$(INCLUDE)}"\ace\SV_Semaphore_Complex.i"\
+ {$(INCLUDE)}"\ace\SV_Semaphore_Simple.h"\
+ {$(INCLUDE)}"\ace\SV_Semaphore_Simple.i"\
+ {$(INCLUDE)}"\ace\Svc_Conf_Tokens.h"\
+ {$(INCLUDE)}"\ace\Synch.h"\
+ {$(INCLUDE)}"\ace\Synch.i"\
+ {$(INCLUDE)}"\ace\Synch_T.cpp"\
+ {$(INCLUDE)}"\ace\Synch_T.h"\
+ {$(INCLUDE)}"\ace\Synch_T.i"\
+ {$(INCLUDE)}"\ace\Thread.h"\
+ {$(INCLUDE)}"\ace\Thread.i"\
+ {$(INCLUDE)}"\ace\Thread_Manager.h"\
+ {$(INCLUDE)}"\ace\Thread_Manager.i"\
{$(INCLUDE)}"\ace\Time_Value.h"\
+ {$(INCLUDE)}"\ace\Timer_Heap.h"\
+ {$(INCLUDE)}"\ace\Timer_Heap_T.cpp"\
+ {$(INCLUDE)}"\ace\Timer_Heap_T.h"\
+ {$(INCLUDE)}"\ace\Timer_List.h"\
+ {$(INCLUDE)}"\ace\Timer_List_T.cpp"\
+ {$(INCLUDE)}"\ace\Timer_List_T.h"\
+ {$(INCLUDE)}"\ace\Timer_Queue.h"\
+ {$(INCLUDE)}"\ace\Timer_Queue_T.cpp"\
+ {$(INCLUDE)}"\ace\Timer_Queue_T.h"\
+ {$(INCLUDE)}"\ace\Timer_Queue_T.i"\
+ {$(INCLUDE)}"\ace\Timer_Wheel.h"\
+ {$(INCLUDE)}"\ace\Timer_Wheel_T.cpp"\
+ {$(INCLUDE)}"\ace\Timer_Wheel_T.h"\
+ {$(INCLUDE)}"\ace\Trace.h"\
+ {$(INCLUDE)}"\ace\Version.h"\
+ {$(INCLUDE)}"\ace\ws2tcpip.h"\
"$(INTDIR)\test_proactor.obj" : $(SOURCE) $(DEP_CPP_TEST_) "$(INTDIR)"
diff --git a/examples/Reactor/Proactor/test_proactor.mdp b/examples/Reactor/Proactor/test_proactor.mdp
index 2518187c3b9..7714f2926ef 100644
--- a/examples/Reactor/Proactor/test_proactor.mdp
+++ b/examples/Reactor/Proactor/test_proactor.mdp
Binary files differ
View Lorry Controller Status