首页 > 代码库 > ACE_Proactor UDP（接收什么就打印什么）V1.0
ACE_Proactor UDP（接收什么就打印什么）V1.0
下面的程序使用Proactor模式用UDP通信，由于UDP是无连接的，所以这里没有Connector和Acceptor
发送端发送一个复合消息，并打印发送的内容
接收端接收一个复合消息并打印接收到的内容
本例是对ACE自带的example的稍微修改了一下（打印发送和接收的内容，这样更加直观）
发送端：client_main.cpp
#include <vector>#include <fstream>#include <iterator>#include <iostream>#include <string>using namespace std;#include "ace/Reactor.h"#include "ace/Message_Queue.h"#include "ace/Asynch_IO.h"#include "ace/OS.h"#include "ace/Proactor.h"#include "ace/Asynch_Connector.h"#include <ace/SOCK_Dgram.h> //=============================================================================/** * @file test_udp_proactor.cpp * * $Id: test_udp_proactor.cpp 93639 2011-03-24 13:32:13Z johnnyw $ * * This program illustrates how the <ACE_Proactor> can be used to * implement an application that does asynchronous operations using * datagrams. * * * @author Irfan Pyarali <irfan@cs.wustl.edu> and Roger Tragin <r.tragin@computer.org> *///=============================================================================#include "ace/OS_NS_string.h"#include "ace/OS_main.h"#include "ace/Proactor.h"#include "ace/Asynch_IO.h"#include "ace/INET_Addr.h"#include "ace/SOCK_Dgram.h"#include "ace/Message_Block.h"#include "ace/Get_Opt.h"#include "ace/Log_Msg.h"// Keep track of when we're done.static int done = 0;/** * @class Sender * * @brief The class will be created by <main>. */class Sender : public ACE_Handler{public: Sender (void); ~Sender (void); //FUZZ: disable check_for_lack_ACE_OS ///FUZZ: enable check_for_lack_ACE_OS int open (const ACE_TCHAR *host, u_short port);protected: // These methods are called by the freamwork /// This is called when asynchronous writes from the dgram socket /// complete virtual void handle_write_dgram (const ACE_Asynch_Write_Dgram::Result &result);private: /// Network I/O handle ACE_SOCK_Dgram sock_dgram_; /// wd (write dgram): for writing to the socket ACE_Asynch_Write_Dgram wd_; const char* completion_key_; const char* act_;};Sender::Sender (void) : completion_key_ ("Sender completion key"), act_ ("Sender ACT"){}Sender::~Sender (void){ this->sock_dgram_.close ();}intSender::open (const ACE_TCHAR *host, u_short port){ // Initialize stuff if (this->sock_dgram_.open (ACE_INET_Addr::sap_any) == -1) ACE_ERROR_RETURN ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_SOCK_Dgram::open"), -1); // Initialize the asynchronous read. if (this->wd_.open (*this, this->sock_dgram_.get_handle (), this->completion_key_, ACE_Proactor::instance ()) == -1) ACE_ERROR_RETURN ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_Asynch_Write_Dgram::open"), -1); // We are using scatter/gather to send the message header and // message body using 2 buffers // create a message block for the message header ACE_Message_Block* msg = 0; ACE_NEW_RETURN (msg, ACE_Message_Block (100), -1); const char raw_msg [] = "To be or not to be."; // Copy buf into the Message_Block and update the wr_ptr (). msg->copy (raw_msg, ACE_OS::strlen (raw_msg) + 1); // create a message block for the message body ACE_Message_Block* body = 0; ACE_NEW_RETURN (body, ACE_Message_Block (100), -1); ACE_OS::memset (body->wr_ptr (), 'X', 100); body->wr_ptr (100); // always remember to update the wr_ptr () // set body as the cont of msg. This associates the 2 message blocks so // that a send will send the first block (which is the header) up to // length (), and use the cont () to get the next block to send. You can // chain up to IOV_MAX message block using this method. msg->cont (body); // do the asynch send size_t number_of_bytes_sent = 0; ACE_INET_Addr serverAddr (port, host); int res = this->wd_.send (msg, number_of_bytes_sent, 0, serverAddr, this->act_); ACE_Message_Block* p = 0; p= msg; switch (res) { case 0: // this is a good error. The proactor will call our handler when the // send has completed. break; case 1: // actually sent something, we will handle it in the handler callback ACE_DEBUG ((LM_DEBUG, "********************\n")); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes sent immediately", number_of_bytes_sent)); while (p != NULL) { ACE_DEBUG ((LM_DEBUG,"YOU SEND[%s]\n",p->rd_ptr())); p = p->cont(); } ACE_DEBUG ((LM_DEBUG, "********************\n")); res = 0; break; case -1: // Something else went wrong. ACE_ERROR ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_Asynch_Write_Dgram::recv")); // the handler will not get called in this case so lets clean up our msg msg->release (); break; default: // Something undocumented really went wrong. ACE_ERROR ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_Asynch_Write_Dgram::recv")); msg->release (); break; } return res;}voidSender::handle_write_dgram (const ACE_Asynch_Write_Dgram::Result &result){ ACE_DEBUG ((LM_DEBUG, "handle_write_dgram called\n")); ACE_DEBUG ((LM_DEBUG, "********************\n")); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_write", result.bytes_to_write ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "handle", result.handle ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_transfered", result.bytes_transferred ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "flags", result.flags ())); ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "act", result.act ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "success", result.success ())); ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "completion_key", result.completion_key ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "error", result.error ())); ACE_DEBUG ((LM_DEBUG, "********************\n")); ACE_DEBUG ((LM_DEBUG, "Sender completed\n")); // No need for this message block anymore. result.message_block ()->release (); // Note that we are done with the test. done++;}intACE_TMAIN (int argc, ACE_TCHAR *argv[]){ //ACE_LOG_MSG->clr_flags(0); //ACE_LOG_MSG->set_flags(ACE_Log_Msg::STDERR | ACE_Log_Msg::VERBOSE); Sender sender; // Port that we're receiving connections on. u_short port = ACE_DEFAULT_SERVER_PORT; // Host that we're connecting to. string host("localhost"); if (sender.open (host.c_str(), port) == -1) return -1; while (true) { ACE_Proactor::instance ()->handle_events (); } return 0;}接收端server_main.cpp
#include "ace/OS_NS_string.h"#include "ace/OS_main.h"#include "ace/Proactor.h"#include "ace/Asynch_IO.h"#include "ace/INET_Addr.h"#include "ace/SOCK_Dgram.h"#include "ace/Message_Block.h"#include "ace/Get_Opt.h"#include "ace/Log_Msg.h"// Host that we're connecting to.static ACE_TCHAR *host = 0;// Port that we're receiving connections on.static u_short port = ACE_DEFAULT_SERVER_PORT;// Keep track of when we're done.static int done = 0;/** * @class Receiver * * @brief This class will receive data from * the network connection and dump it to a file. */class Receiver : public ACE_Service_Handler{public: // = Initialization and termination. Receiver (void); ~Receiver (void); int open_addr (const ACE_INET_Addr &localAddr);protected: // These methods are called by the framework /// This method will be called when an asynchronous read completes on /// a UDP socket. virtual void handle_read_dgram (const ACE_Asynch_Read_Dgram::Result &result);private: ACE_SOCK_Dgram sock_dgram_; /// rd (read dgram): for reading from a UDP socket. ACE_Asynch_Read_Dgram rd_; const char* completion_key_; const char* act_;};Receiver::Receiver (void) : completion_key_ ("Receiver Completion Key"), act_ ("Receiver ACT"){}Receiver::~Receiver (void){ sock_dgram_.close ();}intReceiver::open_addr (const ACE_INET_Addr &localAddr){ ACE_DEBUG ((LM_DEBUG, "[%D][line:%l]Receiver::open_addr called\n")); // Create a local UDP socket to receive datagrams. if (this->sock_dgram_.open (localAddr) == -1) ACE_ERROR_RETURN ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_SOCK_Dgram::open"), -1); // Initialize the asynchronous read. if (this->rd_.open (*this, this->sock_dgram_.get_handle (), this->completion_key_, ACE_Proactor::instance ()) == -1) ACE_ERROR_RETURN ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_Asynch_Read_Dgram::open"), -1); // Create a buffer to read into. We are using scatter/gather to // read the message header and message body into 2 buffers // create a message block to read the message header ACE_Message_Block* msg = 0; ACE_NEW_RETURN (msg, ACE_Message_Block (1024), -1); // the next line sets the size of the header, even though we // allocated a the message block of 1k, by setting the size to 20 // bytes then the first 20 bytes of the reveived datagram will be // put into this message block. msg->size (20); // size of header to read is 20 bytes // create a message block to read the message body ACE_Message_Block* body = 0; ACE_NEW_RETURN (body, ACE_Message_Block (1024), -1); // The message body will not exceed 1024 bytes, at least not in this test. // set body as the cont of msg. This associates the 2 message // blocks so that a read will fill the first block (which is the // header) up to size (), and use the cont () block for the rest of // the data. You can chain up to IOV_MAX message block using this // method. msg->cont (body); // ok lets do the asynch read size_t number_of_bytes_recvd = 0; int res = rd_.recv (msg, number_of_bytes_recvd, 0, PF_INET, this->act_); switch (res) { case 0: // this is a good error. The proactor will call our handler when the // read has completed. break; case 1: // actually read something, we will handle it in the handler callback ACE_DEBUG ((LM_DEBUG, "********************\n")); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes recieved immediately", number_of_bytes_recvd)); ACE_DEBUG ((LM_DEBUG, "********************\n")); res = 0; break; case -1: // Something else went wrong. ACE_ERROR ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_Asynch_Read_Dgram::recv")); // the handler will not get called in this case so lets clean up our msg msg->release (); break; default: // Something undocumented really went wrong. ACE_ERROR ((LM_ERROR, "[%D][line:%l]%p\n", "ACE_Asynch_Read_Dgram::recv")); msg->release (); break; } return res;}voidReceiver::handle_read_dgram (const ACE_Asynch_Read_Dgram::Result &result){ ACE_DEBUG ((LM_DEBUG, "handle_read_dgram called\n")); ACE_DEBUG ((LM_DEBUG, "********************\n")); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_to_read", result.bytes_to_read ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "handle", result.handle ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "bytes_transfered", result.bytes_transferred ())); ACE_INET_Addr peerAddr; result.remote_address (peerAddr); ACE_DEBUG ((LM_DEBUG, "%s = %s:%d\n", "peer_address", peerAddr.get_host_addr (), peerAddr.get_port_number ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "flags", result.flags ())); ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "act", result.act ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "success", result.success ())); ACE_DEBUG ((LM_DEBUG, "%s = %s\n", "completion_key", result.completion_key ())); ACE_DEBUG ((LM_DEBUG, "%s = %d\n", "error", result.error ())); ACE_DEBUG ((LM_DEBUG, "********************\n")); if (result.success () && result.bytes_transferred () != 0) { // loop through our message block and print out the contents for (const ACE_Message_Block* msg = result.message_block (); msg != 0; msg = msg->cont ()) { // use msg->length () to get the number of bytes written to the message // block. ACE_DEBUG ((LM_DEBUG, "Buf=[size=<%d>", msg->length ())); for (u_long i = 0; i < msg->length (); ++i) ACE_DEBUG ((LM_DEBUG, "%c", (msg->rd_ptr ())[i])); ACE_DEBUG ((LM_DEBUG, "]\n")); } } ACE_DEBUG ((LM_DEBUG, "Receiver completed\n")); // No need for this message block anymore. result.message_block ()->release (); // Note that we are done with the test. done++;}int ACE_TMAIN (int argc, ACE_TCHAR *argv[]){ //ACE_LOG_MSG->set_flags(ACE_Log_Msg::STDERR | ACE_Log_Msg::VERBOSE); Receiver receiver; if (receiver.open_addr (ACE_INET_Addr (port)) == -1) return -1; while (true) { ACE_Proactor::instance ()->handle_events (); } return 0;}先运行接收端，再运行发送端，你懂的。
发送端程序运行结果：
接收端运行结果：
ACE_Proactor UDP（接收什么就打印什么）V1.0
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。