ACE_Proactor UDP V2.0

2023-12-06 20:18
文章标签 udp v2.0 ace proactor

本文主要是介绍ACE_Proactor UDP V2.0，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

单次发送单次接收

下面的程序使用Proactor模式用UDP通信：

（1）发送端发送一个复合消息，并打印发送的内容

（2）接收端接收一个复合消息并打印接收到的内容

 由于UDP是无连接的，所以这里没有Connector和Acceptor

本例是对ACE自带的example的稍微修改了一下（打印发送和接收的内容，这样更加直观）

发送端：client_main.cpp

[cpp] view plain copy
  1. #include <vector>  
  2. #include <fstream>  
  3. #include <iterator>  
  4. #include <iostream>  
  5. #include <string>  
  6. using namespace std;  
  7. #include "ace/Reactor.h"  
  8. #include "ace/Message_Queue.h"  
  9. #include "ace/Asynch_IO.h"  
  10. #include "ace/OS.h"  
  11. #include "ace/Proactor.h"  
  12. #include "ace/Asynch_Connector.h"  
  13. #include <ace/SOCK_Dgram.h>   
  14.   
  15.   
  16.   
  17.   
  18. //=============================================================================  
  19. /** 
  20.  *  @file    test_udp_proactor.cpp 
  21.  * 
  22.  *  $Id: test_udp_proactor.cpp 93639 2011-03-24 13:32:13Z johnnyw $ 
  23.  * 
  24.  *  This program illustrates how the <ACE_Proactor> can be used to 
  25.  *  implement an application that does asynchronous operations using 
  26.  *  datagrams. 
  27.  * 
  28.  * 
  29.  *  @author Irfan Pyarali <irfan@cs.wustl.edu> and Roger Tragin <r.tragin@computer.org> 
  30.  */  
  31. //=============================================================================  
  32.   
  33.   
  34. #include "ace/OS_NS_string.h"  
  35. #include "ace/OS_main.h"  
  36. #include "ace/Proactor.h"  
  37. #include "ace/Asynch_IO.h"  
  38. #include "ace/INET_Addr.h"  
  39. #include "ace/SOCK_Dgram.h"  
  40. #include "ace/Message_Block.h"  
  41. #include "ace/Get_Opt.h"  
  42. #include "ace/Log_Msg.h"  
  43.   
  44.   
  45.   
  46.   
  47.   
  48.   
  49.   
  50. // Keep track of when we're done.  
  51. static int done = 0;  
  52.   
  53. /** 
  54.  * @class Sender 
  55.  * 
  56.  * @brief The class will be created by <main>. 
  57.  */  
  58. class Sender : public ACE_Handler  
  59. {  
  60. public:  
  61.   Sender (void);  
  62.   ~Sender (void);  
  63.   
  64.   //FUZZ: disable check_for_lack_ACE_OS  
  65.   ///FUZZ: enable check_for_lack_ACE_OS  
  66.   int open (const ACE_TCHAR *host, u_short port);  
  67.   
  68. protected:  
  69.   // These methods are called by the freamwork  
  70.   
  71.   /// This is called when asynchronous writes from the dgram socket  
  72.   /// complete  
  73.   virtual void handle_write_dgram (const ACE_Asynch_Write_Dgram::Result &result);  
  74.   
  75. private:  
  76.   
  77.   /// Network I/O handle  
  78.   ACE_SOCK_Dgram sock_dgram_;  
  79.   
  80.   /// wd (write dgram): for writing to the socket  
  81.   ACE_Asynch_Write_Dgram wd_;  
  82.   
  83.   const char* completion_key_;  
  84.   const char* act_;  
  85. };  
  86.   
  87. Sender::Sender (void)  
  88.   : completion_key_ ("Sender completion key"),  
  89.     act_ ("Sender ACT")  
  90. {  
  91. }  
  92.   
  93. Sender::~Sender (void)  
  94. {  
  95.   this->sock_dgram_.close ();  
  96. }  
  97.   
  98. int  
  99. Sender::open (const ACE_TCHAR *host,  
  100.               u_short port)  
  101. {  
  102.   // Initialize stuff  
  103.   
  104.   if (this->sock_dgram_.open (ACE_INET_Addr::sap_any) == -1)  
  105.     ACE_ERROR_RETURN ((LM_ERROR,  
  106.                        "[%D][line:%l]%p\n",  
  107.                        "ACE_SOCK_Dgram::open"), -1);  
  108.   
  109.   // Initialize the asynchronous read.  
  110.   if (this->wd_.open (*this,  
  111.                       this->sock_dgram_.get_handle (),  
  112.                       this->completion_key_,  
  113.                       ACE_Proactor::instance ()) == -1)  
  114.     ACE_ERROR_RETURN ((LM_ERROR,  
  115.                        "[%D][line:%l]%p\n",  
  116.                        "ACE_Asynch_Write_Dgram::open"), -1);  
  117.   
  118.   // We are using scatter/gather to send the message header and  
  119.   // message body using 2 buffers  
  120.   
  121.   // create a message block for the message header  
  122.   ACE_Message_Block* msg = 0;  
  123.   ACE_NEW_RETURN (msg, ACE_Message_Block (100), -1);  
  124.   const char raw_msg [] = "To be or not to be.";  
  125.   // Copy buf into the Message_Block and update the wr_ptr ().  
  126.   msg->copy (raw_msg, ACE_OS::strlen (raw_msg) + 1);  
  127.   
  128.   // create a message block for the message body  
  129.   ACE_Message_Block* body = 0;  
  130.   ACE_NEW_RETURN (body, ACE_Message_Block (100), -1);  
  131.   ACE_OS::memset (body->wr_ptr (), 'X', 100);  
  132.   body->wr_ptr (100); // always remember to update the wr_ptr ()  
  133.   
  134.   // set body as the cont of msg.  This associates the 2 message blocks so  
  135.   // that a send will send the first block (which is the header) up to  
  136.   // length (), and use the cont () to get the next block to send.  You can  
  137.   // chain up to IOV_MAX message block using this method.  
  138.   msg->cont (body);  
  139.   
  140.   // do the asynch send  
  141.   size_t number_of_bytes_sent = 0;  
  142.   ACE_INET_Addr serverAddr (port, host);  
  143.   int res = this->wd_.send (msg, number_of_bytes_sent, 0, serverAddr, this->act_);  
  144.     
  145.   
  146.   ACE_Message_Block* p = 0;  
  147.   p= msg;  
  148.   
  149.   switch (res)  
  150.     {  
  151.     case 0:  
  152.       // this is a good error.  The proactor will call our handler when the  
  153.       // send has completed.  
  154.       break;  
  155.     case 1:  
  156.       // actually sent something, we will handle it in the handler callback  
  157.       ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  158.       ACE_DEBUG ((LM_DEBUG,  
  159.                   "%s = %d\n",  
  160.                   "bytes sent immediately",  
  161.                   number_of_bytes_sent));  
  162.   
  163.       while (p != NULL)  
  164.       {  
  165.           ACE_DEBUG ((LM_DEBUG,"YOU SEND[%s]\n",p->rd_ptr()));  
  166.           p = p->cont();  
  167.       }  
  168.         
  169.       ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  170.       res = 0;  
  171.       break;  
  172.     case -1:  
  173.       // Something else went wrong.  
  174.       ACE_ERROR ((LM_ERROR,  
  175.                   "[%D][line:%l]%p\n",  
  176.                   "ACE_Asynch_Write_Dgram::recv"));  
  177.       // the handler will not get called in this case so lets clean up our msg  
  178.       msg->release ();  
  179.       break;  
  180.     default:  
  181.       // Something undocumented really went wrong.  
  182.       ACE_ERROR ((LM_ERROR,  
  183.                   "[%D][line:%l]%p\n",  
  184.                   "ACE_Asynch_Write_Dgram::recv"));  
  185.       msg->release ();  
  186.       break;  
  187.     }  
  188.   return res;  
  189. }  
  190.   
  191. void  
  192. Sender::handle_write_dgram (const ACE_Asynch_Write_Dgram::Result &result)  
  193. {  
  194.   ACE_DEBUG ((LM_DEBUG,  
  195.               "handle_write_dgram called\n"));  
  196.   
  197.   ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  198.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_write", result.bytes_to_write ()));  
  199.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "handle", result.handle ()));  
  200.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_transfered", result.bytes_transferred ()));  
  201.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "flags", result.flags ()));  
  202.   ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "act", result.act ()));  
  203.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "success", result.success ()));  
  204.   ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "completion_key", result.completion_key ()));  
  205.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "error", result.error ()));  
  206.   ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  207.   
  208.   ACE_DEBUG ((LM_DEBUG,  
  209.               "Sender completed\n"));  
  210.   
  211.   // No need for this message block anymore.  
  212.   result.message_block ()->release ();  
  213.   
  214.   // Note that we are done with the test.  
  215.   done++;  
  216. }  
  217.   
  218.   
  219.   
  220. int  
  221. ACE_TMAIN (int argc, ACE_TCHAR *argv[])  
  222. {  
  223.   
  224.     //ACE_LOG_MSG->clr_flags(0);  
  225.     //ACE_LOG_MSG->set_flags(ACE_Log_Msg::STDERR | ACE_Log_Msg::VERBOSE);  
  226.   
  227.     Sender sender;  
  228.     // Port that we're receiving connections on.  
  229.     u_short port = ACE_DEFAULT_SERVER_PORT;  
  230.     // Host that we're connecting to.  
  231.     string host("localhost");  
  232.     if (sender.open (host.c_str(), port) == -1)  
  233.     return -1;  
  234.   
  235.     while (true)  
  236.     {  
  237.         ACE_Proactor::instance ()->handle_events ();  
  238.     }  
  239.       
  240.     return 0;  
  241. }  

接收端server_main.cpp

[cpp] view plain copy
  1. #include "ace/OS_NS_string.h"  
  2. #include "ace/OS_main.h"  
  3. #include "ace/Proactor.h"  
  4. #include "ace/Asynch_IO.h"  
  5. #include "ace/INET_Addr.h"  
  6. #include "ace/SOCK_Dgram.h"  
  7. #include "ace/Message_Block.h"  
  8. #include "ace/Get_Opt.h"  
  9. #include "ace/Log_Msg.h"  
  10.   
  11.   
  12.   
  13. // Host that we're connecting to.  
  14. static ACE_TCHAR *host = 0;  
  15.   
  16. // Port that we're receiving connections on.  
  17. static u_short port = ACE_DEFAULT_SERVER_PORT;  
  18.   
  19. // Keep track of when we're done.  
  20. static int done = 0;  
  21.   
  22. /** 
  23.  * @class Receiver 
  24.  * 
  25.  * @brief This class will receive data from 
  26.  * the network connection and dump it to a file. 
  27.  */  
  28. class Receiver : public ACE_Service_Handler  
  29. {  
  30. public:  
  31.   // = Initialization and termination.  
  32.   Receiver (void);  
  33.   ~Receiver (void);  
  34.   
  35.   int open_addr (const ACE_INET_Addr &localAddr);  
  36.   
  37. protected:  
  38.   // These methods are called by the framework  
  39.   
  40.   /// This method will be called when an asynchronous read completes on  
  41.   /// a UDP socket.  
  42.   virtual void handle_read_dgram (const ACE_Asynch_Read_Dgram::Result &result);  
  43.   
  44. private:  
  45.   ACE_SOCK_Dgram sock_dgram_;  
  46.   
  47.   /// rd (read dgram): for reading from a UDP socket.  
  48.   ACE_Asynch_Read_Dgram rd_;  
  49.   const char* completion_key_;  
  50.   const char* act_;  
  51. };  
  52.   
  53. Receiver::Receiver (void)  
  54.   : completion_key_ ("Receiver Completion Key"),  
  55.     act_ ("Receiver ACT")  
  56. {  
  57. }  
  58.   
  59. Receiver::~Receiver (void)  
  60. {  
  61.   sock_dgram_.close ();  
  62. }  
  63.   
  64. int  
  65. Receiver::open_addr (const ACE_INET_Addr &localAddr)  
  66. {  
  67.   ACE_DEBUG ((LM_DEBUG,  
  68.               "[%D][line:%l]Receiver::open_addr called\n"));  
  69.   
  70.   // Create a local UDP socket to receive datagrams.  
  71.   if (this->sock_dgram_.open (localAddr) == -1)  
  72.     ACE_ERROR_RETURN ((LM_ERROR,  
  73.                        "[%D][line:%l]%p\n",  
  74.                        "ACE_SOCK_Dgram::open"), -1);  
  75.   
  76.   // Initialize the asynchronous read.  
  77.   if (this->rd_.open (*this,  
  78.                       this->sock_dgram_.get_handle (),  
  79.                       this->completion_key_,  
  80.                       ACE_Proactor::instance ()) == -1)  
  81.     ACE_ERROR_RETURN ((LM_ERROR,  
  82.                        "[%D][line:%l]%p\n",  
  83.                        "ACE_Asynch_Read_Dgram::open"), -1);  
  84.   
  85.   // Create a buffer to read into.  We are using scatter/gather to  
  86.   // read the message header and message body into 2 buffers  
  87.   
  88.   // create a message block to read the message header  
  89.   ACE_Message_Block* msg = 0;  
  90.   ACE_NEW_RETURN (msg, ACE_Message_Block (1024), -1);  
  91.   
  92.   // the next line sets the size of the header, even though we  
  93.   // allocated a the message block of 1k, by setting the size to 20  
  94.   // bytes then the first 20 bytes of the reveived datagram will be  
  95.   // put into this message block.  
  96.   msg->size (20); // size of header to read is 20 bytes  
  97.   
  98.   // create a message block to read the message body  
  99.   ACE_Message_Block* body = 0;  
  100.   ACE_NEW_RETURN (body, ACE_Message_Block (1024), -1);  
  101.   // The message body will not exceed 1024 bytes, at least not in this test.  
  102.   
  103.   // set body as the cont of msg.  This associates the 2 message  
  104.   // blocks so that a read will fill the first block (which is the  
  105.   // header) up to size (), and use the cont () block for the rest of  
  106.   // the data.  You can chain up to IOV_MAX message block using this  
  107.   // method.  
  108.   msg->cont (body);  
  109.   
  110.   // ok lets do the asynch read  
  111.   size_t number_of_bytes_recvd = 0;  
  112.   
  113.   int res = rd_.recv (msg,  
  114.                       number_of_bytes_recvd,  
  115.                       0,  
  116.                       PF_INET,  
  117.                       this->act_);  
  118.   switch (res)  
  119.     {  
  120.     case 0:  
  121.       // this is a good error.  The proactor will call our handler when the  
  122.       // read has completed.  
  123.       break;  
  124.     case 1:  
  125.       // actually read something, we will handle it in the handler callback  
  126.       ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  127.       ACE_DEBUG ((LM_DEBUG,  
  128.                   "%s = %d\n",  
  129.                   "bytes recieved immediately",  
  130.                   number_of_bytes_recvd));  
  131.   
  132.       ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  133.       res = 0;  
  134.       break;  
  135.     case -1:  
  136.       // Something else went wrong.  
  137.       ACE_ERROR ((LM_ERROR,  
  138.                   "[%D][line:%l]%p\n",  
  139.                   "ACE_Asynch_Read_Dgram::recv"));  
  140.       // the handler will not get called in this case so lets clean up our msg  
  141.       msg->release ();  
  142.       break;  
  143.     default:  
  144.       // Something undocumented really went wrong.  
  145.       ACE_ERROR ((LM_ERROR,  
  146.                   "[%D][line:%l]%p\n",  
  147.                   "ACE_Asynch_Read_Dgram::recv"));  
  148.       msg->release ();  
  149.       break;  
  150.     }  
  151.   
  152.   return res;  
  153. }  
  154.   
  155. void  
  156. Receiver::handle_read_dgram (const ACE_Asynch_Read_Dgram::Result &result)  
  157. {  
  158.   ACE_DEBUG ((LM_DEBUG,  
  159.               "handle_read_dgram called\n"));  
  160.   
  161.   ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  162.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_read", result.bytes_to_read ()));  
  163.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "handle", result.handle ()));  
  164.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_transfered", result.bytes_transferred ()));  
  165.   ACE_INET_Addr peerAddr;  
  166.   result.remote_address (peerAddr);  
  167.   ACE_DEBUG ((LM_DEBUG, "%s = %s:%d\n", "peer_address", peerAddr.get_host_addr (), peerAddr.get_port_number ()));  
  168.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "flags", result.flags ()));  
  169.   ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "act", result.act ()));  
  170.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "success", result.success ()));  
  171.   ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "completion_key", result.completion_key ()));  
  172.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "error", result.error ()));  
  173.   ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  174.   
  175.   if (result.success () && result.bytes_transferred () != 0)  
  176.     {  
  177.       // loop through our message block and print out the contents  
  178.       for (const ACE_Message_Block* msg = result.message_block (); msg != 0; msg = msg->cont ())  
  179.         { // use msg->length () to get the number of bytes written to the message  
  180.           // block.  
  181.           ACE_DEBUG ((LM_DEBUG, "Buf=[size=<%d>", msg->length ()));  
  182.           for (u_long i = 0; i < msg->length (); ++i)  
  183.             ACE_DEBUG ((LM_DEBUG,  
  184.                         "%c", (msg->rd_ptr ())[i]));  
  185.           ACE_DEBUG ((LM_DEBUG, "]\n"));  
  186.         }  
  187.     }  
  188.   
  189.   ACE_DEBUG ((LM_DEBUG,  
  190.               "Receiver completed\n"));  
  191.   
  192.   // No need for this message block anymore.  
  193.   result.message_block ()->release ();  
  194.   
  195.   // Note that we are done with the test.  
  196.   done++;  
  197. }  
  198.   
  199.   
  200. int  
  201.     ACE_TMAIN (int argc, ACE_TCHAR *argv[])  
  202. {  
  203.   
  204.     //ACE_LOG_MSG->set_flags(ACE_Log_Msg::STDERR | ACE_Log_Msg::VERBOSE);  
  205.   
  206.     Receiver receiver;  
  207.   
  208.     if (receiver.open_addr (ACE_INET_Addr (port)) == -1)  
  209.         return -1;  
  210.   
  211.     while (true)  
  212.     {  
  213.         ACE_Proactor::instance ()->handle_events ();  
  214.     }  
  215.   
  216.     return 0;  
  217. }  

先运行接收端，再运行发送端，你懂的。

发送端程序运行结果：

接收端运行结果：


定时多目标发送

程序的功能：

（1）UDP发送内容到P1，IP2，...,IPn（地址列表从文件读取）
（1）发送内容从文件中读取；
（1）发送时间间隔从文件中读取；

[cpp] view plain copy
  1. //=============================================================================  
  2. /** 
  3.  *  @file    test_udp_proactor.cpp 
  4.  * 
  5.  *  $Id: test_udp_proactor.cpp 93639 2011-03-24 13:32:13Z johnnyw $ 
  6.  * 
  7.  *  This program illustrates how the <ACE_Proactor> can be used to 
  8.  *  implement an application that does asynchronous operations using 
  9.  *  datagrams. 
  10.  * 
  11.  * 
  12.  *  @author Irfan Pyarali <irfan@cs.wustl.edu> and Roger Tragin <r.tragin@computer.org> 
  13.  */  
  14. //=============================================================================  
  15.   
  16. #include <vector>  
  17. #include <fstream>  
  18. #include <iterator>  
  19. #include <iostream>  
  20. #include <string>  
  21. using namespace std;  
  22. //#include "ace/Reactor.h"  
  23. #include "ace/Message_Queue.h"  
  24. #include "ace/Asynch_IO.h"  
  25. #include "ace/OS.h"  
  26. #include "ace/Proactor.h"  
  27. #include "ace/Asynch_Connector.h"  
  28. #include <ace/SOCK_Dgram.h>   
  29.   
  30. #include "ace/OS_NS_string.h"  
  31. #include "ace/OS_main.h"  
  32. #include "ace/INET_Addr.h"  
  33. #include "ace/SOCK_Dgram.h"  
  34. #include "ace/Message_Block.h"  
  35. #include "ace/Get_Opt.h"  
  36. #include "ace/Log_Msg.h"  
  37. #include "ace/Event_Handler.h"  
  38. #include "ace/Date_Time.h"  
  39. #include "ace/WIN32_Proactor.h"  
  40.   
  41. namespace global  
  42. {  
  43.     int delay = 2;  
  44.     //int interval = 60*10;//每interval 秒计时一次  
  45.     int interval = 2;//每interval 秒计时一次  
  46.     void print_current_time(void)  
  47.     {  
  48.         ACE_Date_Time date(ACE_OS::gettimeofday());  
  49.         cout<<"当前时间："  
  50.             <<date.year()<<"年"  
  51.             <<date.month()<<"月"  
  52.             <<date.day()<<"日"  
  53.             <<date.hour()<<"时"  
  54.             <<date.minute()<<"分"  
  55.             <<date.second()<<"秒"<<endl;  
  56.     }  
  57.     template<typename T>  
  58.     bool read_server_addr(vector<T>& addrs)  
  59.     {  
  60.         ifstream fin("server_addr.ini");  
  61.         if (!fin)  
  62.         {  
  63.             cout<<"找不到配置文件：local_port.ini"<<endl;  
  64.             return false;  
  65.         }  
  66.         istream_iterator<T> first(fin),last;  
  67.         vector<string> temp_addrs(first,last);  
  68.   
  69.         if (temp_addrs.size()==0)  
  70.         {  
  71.             cout<<"配置文件中找不到服务器地址！"<<endl;  
  72.             return false;  
  73.         }  
  74.         else  
  75.         {  
  76.             addrs.swap(temp_addrs);  
  77.             return true;  
  78.         }  
  79.     }  
  80.     template<typename T>  
  81.     bool read_interval(T& interval)  
  82.     {  
  83.   
  84.         ifstream fin("interval_second.ini");  
  85.         if (!fin)  
  86.         {  
  87.             cout<<"找不到配置文件：interval_second.ini"<<endl;  
  88.             return false;  
  89.         }  
  90.         //istream_iterator<T> first(fin),last;  
  91.         //vector<string> temp_addrs(first,last);  
  92.         fin>>interval;  
  93.         if (!fin)  
  94.         {  
  95.             cout<<"配置文件中找不到发送时间间隔数据！"<<endl;  
  96.             return false;  
  97.         }  
  98.         else  
  99.         {  
  100.             global::interval = interval;  
  101.             return true;  
  102.         }  
  103.     }  
  104. }  
  105.   
  106. /** 
  107.  * @class Sender 
  108.  * 
  109.  * @brief The class will be created by <main>. 
  110.  */  
  111. class Sender : public ACE_Handler, public ACE_Event_Handler  
  112.   
  113. {  
  114. public:  
  115.   Sender (const int delay,const int interval);  
  116.   ~Sender (void);  
  117.   
  118.   //FUZZ: disable check_for_lack_ACE_OS  
  119.   ///FUZZ: enable check_for_lack_ACE_OS  
  120.   int open (const ACE_TCHAR *host, u_short port);  
  121.   int handle_timeout(const ACE_Time_Value& , const void *act /* = 0 */);//计时器到期后执行的回调函数  
  122. protected:  
  123.   /// This is called when asynchronous writes from the dgram socket  
  124.   /// complete  
  125.   virtual void handle_write_dgram (const ACE_Asynch_Write_Dgram::Result &result);  
  126.   
  127. private:  
  128.   void start_timing(void);  
  129.   int send_to_one_server(const string&,const string&);  
  130.   int send_to_multi_server(void);  
  131.   void read_content(string&);  
  132.   /// Network I/O handle  
  133.   ACE_SOCK_Dgram sock_dgram_;  
  134.   
  135.   /// wd (write dgram): for writing to the socket  
  136.   ACE_Asynch_Write_Dgram wd_;  
  137.   
  138.   const char* completion_key_;  
  139.   const char* act_;  
  140.   long time_handle_;//在计时器队列中的ID  
  141.   int delay_;//启动多久开始第一次触发超时  
  142.   int interval_;//循环计时的间隔  
  143. };  
  144.   
  145. Sender::Sender (const int delay,const int interval)  
  146.   : completion_key_ ("Sender completion key"),  
  147.     act_ ("Sender ACT"),  
  148.     delay_(delay),  
  149.     interval_(interval)  
  150. {  
  151.     ACE_DEBUG ((LM_DEBUG, "Sender::Sender (const int delay,const int interval)\n"));  
  152. }  
  153.   
  154. Sender::~Sender (void)  
  155. {  
  156.   this->sock_dgram_.close ();  
  157. }  
  158.   
  159. int  
  160. Sender::open (const ACE_TCHAR *host,  
  161.               u_short port)  
  162. {  
  163.     ACE_DEBUG ((LM_DEBUG, "Sender::open(%s,%d)\n",host,port));  
  164.   
  165.   // Initialize stuff  
  166.   //初始化和socket有关的成员  
  167.   if (this->sock_dgram_.open (ACE_INET_Addr::sap_any) == -1)  
  168.     ACE_ERROR_RETURN ((LM_ERROR,  
  169.                        "[%D][line:%l]%p\n",  
  170.                        "ACE_SOCK_Dgram::open"), -1);  
  171.   
  172.   // Initialize the asynchronous read.  
  173.   if (this->wd_.open (*this,  
  174.                       this->sock_dgram_.get_handle (),  
  175.                       this->completion_key_,  
  176.                       ACE_Proactor::instance ()) == -1)  
  177.     ACE_ERROR_RETURN ((LM_ERROR,  
  178.                        "[%D][line:%l]%p\n",  
  179.                        "ACE_Asynch_Write_Dgram::open"), -1);  
  180.   
  181.   //init time clock  
  182.   //启动计时  
  183.   start_timing();  
  184.   return 0;  
  185.     
  186. }  
  187.   
  188. void Sender::start_timing(void)  
  189. {  
  190.     ACE_DEBUG ((LM_DEBUG, "Sender::start_timing:delay[%d]interval[%d]\n",  
  191.         this->delay_,this->interval_));  
  192.     this->reactor(ACE_Reactor::instance());  
  193.     this->time_handle_ = this->reactor()->schedule_timer(this,//在这里注册定时器  
  194.         0,  
  195.         ACE_Time_Value(this->delay_),//程序一开始延迟delay秒开始首次执行到期函数  
  196.         ACE_Time_Value(this->interval_));//循环计时，每隔interval秒重复执行  
  197. }  
  198.   
  199. int Sender::handle_timeout(const ACE_Time_Value& , const void *act /* = 0 */)  
  200. {  
  201.     cout<<"\n\n\n计时器"<<this->interval_/60<<"分钟到期"<<endl;  
  202.     global::print_current_time();  
  203.       
  204.     //if (date.minute() /10 == 3 )//在30分-39分之间这个时间段的时候开始任务  
  205.     if (true )//在30分-39分之间这个时间段的时候开始任务  
  206.     {  
  207.         cout<<"当前时间在31分-39分之间， ，开始查找文件并尝试发送到对端";  
  208.         cout<<"开始发送文件到对端"<<endl;  
  209.         int num = send_to_multi_server();  
  210.         ACE_DEBUG ((LM_DEBUG, "UDP成功发送文件内容到%d个目标地址\n",num));  
  211.     }  
  212.     else  
  213.     {  
  214.         global::print_current_time();  
  215.         cout<<"不需要上传"<<endl;  
  216.     }  
  217.       
  218.     return 0;  
  219. }  
  220.   
  221. void Sender::read_content(string& content)  
  222. {  
  223.     string file_name("temp_file.xml");  
  224.     ifstream fin(file_name);  
  225.     istream_iterator<char> iter_begin(fin),iter_end;  
  226.     string send_str(iter_begin,iter_end);  
  227.     content.swap(send_str);  
  228. }  
  229.   
  230. int Sender::send_to_one_server(const string& addr,const string& sent_content)  
  231. {  
  232.     // create a message block for the message header  
  233.     ACE_Message_Block* msg = 0;  
  234.     ACE_NEW_RETURN (msg, ACE_Message_Block (100), -1);  
  235.     // Copy buf into the Message_Block and update the wr_ptr ().  
  236.     msg->copy (sent_content.c_str(), sent_content.size());  
  237.     // do the asynch send  
  238.   
  239.     size_t number_of_bytes_sent = 0;  
  240.     ACE_INET_Addr serverAddr (addr.c_str());  
  241.     int res = this->wd_.send (msg, number_of_bytes_sent, 0, serverAddr, this->act_);  
  242.   
  243.     ACE_Message_Block* p = 0;  
  244.     p= msg;  
  245.   
  246.     switch (res)  
  247.     {  
  248.     case 0:  
  249.         // this is a good error.  The proactor will call our handler when the  
  250.         // send has completed.  
  251.         break;  
  252.     case 1:  
  253.         // actually sent something, we will handle it in the handler callback  
  254.         ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  255.         ACE_DEBUG ((LM_DEBUG,  
  256.             "%s = %d\n",  
  257.             "bytes sent immediately",  
  258.             number_of_bytes_sent));  
  259.   
  260.         while (p != NULL)  
  261.         {  
  262.             string temp;  
  263.             for (int i=0;i<p->length();++i)  
  264.             {  
  265.                 temp.push_back(*(p->rd_ptr()+i));  
  266.             }  
  267.             ACE_DEBUG ((LM_DEBUG,"YOU SEND[%s]\n",temp.c_str()));  
  268.               
  269.             p = p->cont();  
  270.         }  
  271.   
  272.         ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  273.         res = 0;  
  274.         break;  
  275.     case -1:  
  276.         // Something else went wrong.  
  277.         ACE_ERROR ((LM_ERROR,  
  278.             "[%D][line:%l]%p\n",  
  279.             "ACE_Asynch_Write_Dgram::recv"));  
  280.         // the handler will not get called in this case so lets clean up our msg  
  281.         msg->release ();  
  282.         break;  
  283.     default:  
  284.         // Something undocumented really went wrong.  
  285.         ACE_ERROR ((LM_ERROR,  
  286.             "[%D][line:%l]%p\n",  
  287.             "ACE_Asynch_Write_Dgram::recv"));  
  288.         msg->release ();  
  289.         break;  
  290.     }  
  291.     return res;  
  292. }  
  293. int Sender::send_to_multi_server(void)  
  294. {  
  295.   
  296.     string send_content;  
  297.     this->read_content(send_content);  
  298.   
  299.     vector<string> server_addrs;  
  300.     global::read_server_addr(server_addrs);  
  301.     int send_success_number = 0;  
  302.     for (vector<string>::const_iterator iter = server_addrs.cbegin();  
  303.         iter != server_addrs.cend();  
  304.         ++iter)  
  305.     {  
  306.         if (send_to_one_server(*iter,send_content))  
  307.         {  
  308.             ++send_success_number;  
  309.         }  
  310.     }  
  311.     return send_success_number;  
  312. }  
  313.   
  314. void  
  315. Sender::handle_write_dgram (const ACE_Asynch_Write_Dgram::Result &result)  
  316. {  
  317.   ACE_DEBUG ((LM_DEBUG,  
  318.               "handle_write_dgram called\n"));  
  319.   
  320.   ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  321.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_write", result.bytes_to_write ()));  
  322.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "handle", result.handle ()));  
  323.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_transfered", result.bytes_transferred ()));  
  324.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "flags", result.flags ()));  
  325.   ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "act", result.act ()));  
  326.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "success", result.success ()));  
  327.   ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "completion_key", result.completion_key ()));  
  328.   ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "error", result.error ()));  
  329.   ACE_DEBUG ((LM_DEBUG, "********************\n"));  
  330.   
  331.   ACE_DEBUG ((LM_DEBUG,  
  332.               "Sender completed\n"));  
  333.   
  334.   // No need for this message block anymore.  
  335.   result.message_block ()->release ();  
  336.   
  337. }  
  338.   
  339.   
  340.   
  341. int  
  342. ACE_TMAIN (int argc, ACE_TCHAR *argv[])  
  343. {  
  344.   
  345.     ACE_DEBUG ((LM_DEBUG, "(%t|%P) work starup/n"));  
  346.     ACE_Proactor::close_singleton ();   
  347.   
  348.     ACE_WIN32_Proactor *impl = new ACE_WIN32_Proactor (0, 1);   
  349.     ACE_Proactor::instance (new ACE_Proactor (impl, 1), 1);  
  350.   
  351.     ACE_Reactor::instance ()->register_handler(impl, impl->get_handle ());  
  352.   
  353.     global::read_interval(global::interval);  
  354.   
  355.     //ACE_LOG_MSG->clr_flags(0);  
  356.     //ACE_LOG_MSG->set_flags(ACE_Log_Msg::STDERR | ACE_Log_Msg::VERBOSE);  
  357.   
  358.     Sender sender(global::delay,global::interval);  
  359.     // Port that we're receiving connections on.  
  360.     u_short port = ACE_DEFAULT_SERVER_PORT;  
  361.     // Host that we're connecting to.  
  362.     string host("localhost");  
  363.     if (sender.open (host.c_str(), port) == -1)  
  364.     return -1;  
  365.   
  366.   
  367.   
  368.     ACE_Reactor::instance()->run_event_loop();  
  369.     ACE_Reactor::instance ()->remove_handler (impl,  
  370.         ACE_Event_Handler::DONT_CALL);  
  371.     ACE_DEBUG ((LM_DEBUG, "(%t|%P) work complete/n"));  
  372.   
  373.   
  374.     return 0;  
  375. }  


所需配置文件：

这篇关于ACE_Proactor UDP V2.0的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/463172。

相关文章

VC环境下window网络程序：UDP Socket程序

VC环境下window网络程序：UDP Socket程序

最近在学Windows网络编程，正好在做UDPsocket的程序，贴上来： 服务器框架函数：              socket();    bind();    recfrom();  sendto();  closesocket(); 客户机框架函数：            socket();      recfrom();  sendto();  closesocket();
阅读更多...
9.7（UDP局域网多客户端聊天室）

9.7（UDP局域网多客户端聊天室）

服务器端 #include<myhead.h>#define SERIP "192.168.0.132"#define SERPORT 8888#define MAX 50//定义用户结构体typedef struct{struct sockaddr_in addr;int flag;}User;User users[MAX];//用户列表void add_user(struct s
阅读更多...
应用层简单实现udp / tcp网络通信

应用层简单实现udp / tcp网络通信

一、常见网络接口总结 1、创建 socket 文件描述符 (TCP/UDP, 客户端 + 服务器) int socket(int domain, int type, int protocol); domain：AF_INET：网络通信，AF_LOCAL：本地通信 type：UDP：SOCK_DGRAM，TCP：SOCK_STREAM protocol：协议编号一开始设0 返回值：文件描
阅读更多...
C#通过ACE OLEDB驱动程序访问 Access和 Excel

C#通过ACE OLEDB驱动程序访问 Access和 Excel

ACE 代表 Access Connectivity Engine。它是 Microsoft 提供的一组组件，用于访问和操作 Microsoft Access 数据库以及其他类似的文件格式，如 Excel 工作簿。ACE 主要包括以下几部分： ACE OLEDB 驱动程序：用于通过 OLE DB 提供程序访问 Access 数据库和 Excel 文件。例如，Microsoft.ACE.OLED
阅读更多...
[网络编程]TCP和UDP的比较 及 通过java用UDP实现网络编程

[网络编程]TCP和UDP的比较 及 通过java用UDP实现网络编程

文章目录 一. 网络编程套接字TCP和UDP的特点有连接 vs 无连接可靠传输 vs 不可靠传输面向字节流 vs 面向数据报全双工 vs 半双工 二. java用UDP实现网络编程代码实现:运行代码:启动多个客户端别人能否使用?实现翻译功能 一. 网络编程套接字 网络编程套接字, 就是指操作系统提供的网络编程的api, 称为"socket api" 操作系统, 提供的sock
阅读更多...
Java高级Day37-UDP网络编程

Java高级Day37-UDP网络编程

109.netstat指令 netstat -an 可以查看当前主机网络情况，包括端口监听情况和网络连接情况 netstat -an|more 可以分页显示 要求在dos控制台下执行 说明： LISTENING表示某个端口在监听 如果有一个外部程序（客户端）连接到该端口，就会显示一条连接信息 ctrl + c 退出指令 110.TCP连接秘密 1.当客户端连接到服务器后，
阅读更多...
传输层协议UDP

传输层协议UDP

本篇将主要介绍 UDP 协议，介绍了有关 UDP 协议的报头、协议特点、UDP 协议在操作系统中的缓冲区、UDP 协议使用的注意事项，以及有关 UDP 的 Socket 编程程序，同时重点介绍了操作系统对于 UDP 协议报文的管理。         接着介绍了有关端口号的映射。 目录 UDP协议 1. UDP协议报头 2. UDP协议特点 3. UDP的缓冲区 4.
阅读更多...
若依库存管理 ruoyi-wms V2.0发布：升级到jdk17和vue3，支持一物一码

若依库存管理 ruoyi-wms V2.0发布：升级到jdk17和vue3，支持一物一码

开源地址 https://gitee.com/zccbbg/wms-ruoyi 项目代码、文档 均开源免费可商用 遵循开源协议在项目中保留开源协议文件即可 活到老写到老 为兴趣而开源 为学习而开源 为让大家真正可以学到技术而开源 若依wms是一套基于若依的wms仓库管理系统，支持lodop和网页打印入库单、出库单。毫无保留给个人及企业免费使用。 前端采用Vue、Element UI。
阅读更多...
UDP通信实现

UDP通信实现

目录 前言 一、基础知识 1、跨主机传输         1、字节序          2、主机字节序和网络字节序          3、IP转换 2、套接字 3、什么是UDP通信  二、如何实现UDP通信          1、socket():创建套接字         2、bind():绑定套接字           3、sendto():发送指定套接字文件数据
阅读更多...
MarkTool之网络篇UDP

MarkTool之网络篇UDP

UDP客户端         1、连接服务端         2、校验方式         3、接收数据类型         4、发送数据类型         5、暂停显示         6、将数据区域保存到文件         7、定时发送         8、快捷指令         9、批量发送         10、历史发送
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2