ACE Proactor Framework 5 – Echo 클라이언트 만들기

2013/06/28 by admin

EchoService.h
#pragma once

#include <ace/Asynch_IO.h>
#include <ace/Message_Block.h>
#include <ace/Message_Queue.h>
#include <ace/Synch.h>
#include <ace/Proactor.h>
#include <ace/SOCK_Stream.h>

class EchoService:public ACE_Handler
{
public:
    EchoService(void);
    ~EchoService(void);

   void open(ACE_Proactor* actor, const ACE_HANDLE& handle);
   int send( const char* data);

   virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result);
   virtual void handle_write_stream (const ACE_Asynch_Write_Stream::Result &result);

private:
    ACE_SOCK_Stream stream_;

    ACE_Asynch_Read_Stream read_require_;
    ACE_Message_Block read_msg_;

    ACE_Asynch_Write_Stream write_require_;
    ACE_Message_Queue<ACE_MT_SYNCH> write_msg_queue_;
    ACE_Atomic_Op<ACE_Thread_Mutex, bool> writable_;

private:
   void recv_();

   void close();
};

EchoService.cpp
#include "EchoService.h"
#include <iostream>

EchoService::EchoService(void)
    :read_msg_(1024),writable_(false),stream_(ACE_INVALID_HANDLE)
{
}

EchoService::~EchoService(void)
{
   this->close();
}

void
EchoService::open(ACE_Proactor* actor, const ACE_HANDLE& handle){
   this->proactor(actor);
   this->stream_.set_handle(handle);

   this->read_require_.open(*this, stream_.get_handle(), 0, this->proactor());
   this->write_require_.open(*this, stream_.get_handle(), 0, this->proactor());

   this->writable_ = true;
   this->recv_();

   if( this->write_msg_queue_.is_empty()) return;
   ACE_Message_Block* m;
   this->write_msg_queue_.dequeue_head(m);
   this->write_require_.write(*m, m->length());
   this->writable_ = false;
}

void
EchoService::close(){
   this->read_require_.cancel();
   this->write_require_.cancel();
   this->stream_.close();
   this->stream_.set_handle( ACE_INVALID_HANDLE);
   this->writable_ = false;
}

void
EchoService::recv_(){
    read_msg_.reset();
   this->read_require_.read( read_msg_, read_msg_.space() - 1);
}

void
EchoService::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result){
   if( !result.success()){
      this->close();
       std::cout<<std::endl<<"Error:"<<result.error();
      return;
   }

   if( 0==result.message_block().length()){
      this->close();
       std::cout<<std::endl<<"Disconnected!!";
      return;
   }

   result.message_block().copy("");
    std::cout<<std::endl<<"RECV:"<<(const char*) result.message_block().rd_ptr();
   this->recv_();

}

int
EchoService::send( const char* data){
   if( ACE_INVALID_HANDLE== stream_.get_handle()){
      return -1;
   }

   size_t len = ACE_OS::strlen( data )+1;
   ACE_Message_Block* m;
   ACE_NEW_RETURN(m, ACE_Message_Block(len), -1);
    m->copy( data);

   if( this->writable_.value() ){
      this->writable_ = false;
      return this->write_require_.write(*m, m->length());
   }

   return this->write_msg_queue_.enqueue_tail(m);
}

void
EchoService::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result){
   if( !result.success()){
      this->write_msg_queue_.enqueue_head(&result.message_block());
      this->close();
       std::cout<<std::endl<<"Error:"<<result.error();
      return;
   }

   if( 0==result.message_block().length()){
      result.message_block().release();

      if( this->write_msg_queue_.is_empty()){
         this->writable_ = true;
         return;
      }

      ACE_Message_Block* m;
      this->write_msg_queue_.dequeue_head(m);
      this->write_require_.write(*m, m->length());
      return;
   }

   this->write_require_.write(result.message_block(), result.message_block().length());
}

PConnector.h
#pragma once
#include <ace/Asynch_IO.h>
#include <ace/INET_Addr.h>
#include "EchoService.h"
#include <ace/Proactor.h>

class PConnector:public ACE_Handler
{
public:
    PConnector(ACE_Proactor* actor, const char* ipstr, EchoService& svc);
    ~PConnector(void);

   virtual void handle_connect (const ACE_Asynch_Connect::Result &result);
private:
   ACE_Asynch_Connect connect_require_;
   ACE_INET_Addr addr_;

   EchoService& service_;

private:
   void connect_();
};

PConnector.cpp
#include "PConnector.h"
#include <ace/OS.h>
#include <iostream>

PConnector::PConnector(ACE_Proactor* actor, const char* ipstr, EchoService& svc)
    :ACE_Handler(actor), addr_(ipstr), service_(svc)
{
   this->connect_require_.open(*this, ACE_INVALID_HANDLE,0, this->proactor());
   this->connect_();
}

PConnector::~PConnector(void)
{
   this->connect_require_.cancel();
}

void
PConnector::connect_(){
   this->connect_require_.connect(ACE_INVALID_HANDLE, addr_, ACE_Addr::sap_any, 1);
}

void
PConnector::handle_connect (const ACE_Asynch_Connect::Result &result){
   if(!result.success()){
       std::cout<<std::endl<<"Connect fail!!: "<<result.error();
      this->connect_();
      return;
   }

   if( ACE_INVALID_HANDLE==result.connect_handle()){
       std::cout<<std::endl<<"Connect fail!!:"<<result.error();;
      this->connect_();
      return;
   }

   if( 0!=result.error()){
       ACE_OS::closesocket(result.connect_handle());
       std::cout<<std::endl<<"Connect fail!!:"<<result.error();;
      this->connect_();
      return;
   }

    std::cout<<std::endl<<"Connect Success!!";
    ACE::clr_flags( result.connect_handle(), ACE_NONBLOCK);

    service_.open(this->proactor(), result.connect_handle());
}

PEchoClient.cpp
// PEchoClient.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"

#include <ace/ACE.h>
#include "PConnector.h"
#include <ace/Thread_Manager.h>
#include <string>
#include <iostream>

ACE_THR_FUNC_RETURN fn(void * data){
   EchoService* svc = (EchoService*)data;

    std::string cmd;
   while(true){
       std::cout<<std::endl<<"전송할 데이터를 입력하세요. 종료시 exit를 입력하세요"<<std::endl;
       std::getline( std::cin, cmd);
      if("exit"==cmd){
         ACE_Proactor::instance()->proactor_end_event_loop();
         break;
      }

      for( int i=0;i<1000;++i){
          svc->send( cmd.c_str());
      }

       svc->send("end");

       ACE_OS::sleep(4);

   }

   return 0;
}

int _tmain(int argc, _TCHAR* argv[])
{
    ACE::init();
   {
      EchoService svc;
      PConnector cn(ACE_Proactor::instance(), "192.168.0.2:1000",svc);

      ACE_Thread_Manager tm;
       tm.spawn(fn,&svc);

      ACE_Proactor::instance()->proactor_run_event_loop();

       tm.wait();
   }
    ACE::fini();

   return 0;
}

- 목록:

라이브러리 포팅하기
간단히 Echo 서버 만들어보기
간단히 Echo 클라이언트 만들기
ACE_Select_Reactor에 대한 이해
ACE_TP_Reactor 이해 – Leader/Followers 패턴
ACE_WMFO_Reactor 이해
Service Configurator Framework 1 – 개요
Service Configurator Framework 2 – ACE_Service_Config 클래스 이해
Service Configurator Framework 3 – 동적/정적 서비스
Service Configurator Framework 4 – 서비스를 등록하는 과정
Service Configurator Framework 5 – ACE_Service_Repository 클래스
ACE Task Framework 1 – 개요
ACE Task Framework 2 – Active Object 패턴 이해
ACE Task Framework 3 – Half-Sync/Half-Async 패턴 이해
ACE Task Framework 4 – ACE_Task_Base 클래스 이해
ACE Task Framework 5 – ACE_Task 클래스 이해
ACE Task Framework 6 – ACE_Message_Block 클래스 이해 1
ACE Task Framework 7 – ACE_Message_Block 클래스 이해 2
ACE Task Framework 8 – ACE_Message_Queue 이해 1
ACE Task Framework 9 – ACE_Message_Queue 이해 2
ACE Task Framework 10 – ACE_Message_Queue 이해 3
ACE Task Framework 11 – ACE_Thread_Manager 클래스 이해 1
ACE Task Framework 12 – ACE_Thread_Manager 클래스 이해 2
ACE Task Framework 13 – ACE_Thread_Manager 클래스 이해 3
ACE Task Framework 14 – 기본 샘플1
ACE Task Framework 15 – 기본 샘플2
ACE Task Framework 16 – 기본 샘플3
ACE Acceptor-Connector framework 1 – ACE_Svc_Handler 이해1
ACE Acceptor-Connector framework 2 – ACE_Svc_Handler 이해2
ACE Acceptor-Connector framework 3 – ACE_Acceptor 클래스 이해
ACE Acceptor-Connector framework 4 – ACE_Connector 클래스 이해
ACE Proactor Framework 1 – 비동기과 서버 모델
ACE Proactor Framework 2 – 비동기 IO 이해하기
ACE Proactor Framework 3 – 비동기 IO 기술 요소
ACE Proactor Framework 4 – Echo 서버 만들기
ACE Proactor Framework 5 – Echo 클라이언트 만들기
ACE Proactor Framework 6 – UDP 사용하기
ACE Proactor Framework 7 – serial port 사용하기
ACE Proactor Framework 8 – Review
ACE Proactor Framework 9 – Acceptor-Connector 예제 1
ACE Proactor Framework 10 – Acceptor-Connector 예제 2