bool CoAPRDService::initialize()
{
    try
    {
        std::string conf("service.conf");
        std::string reactor("service.reactor");

        _confPtr = ServiceGetter::findByName<ConfService>(context(), conf);
        _reactorPtr = ServiceGetter::findByName<ReactorService>(context(), reactor);

    }
    catch(toolkit::NullPointerException &e)
    {
        ACE_DEBUG((LM_DEBUG, "get service failed at SerialPortService\n"));

        return false;
    }


    _coapWrapperPtr = new CoAPWrapper();

    if (_coapWrapperPtr->Create(_confPtr->_rdAddr,
                                _confPtr->_rdAddrPort,
                                _confPtr->_coapDebugLevel
                               ) < 0)
    {

        ACE_DEBUG((LM_DEBUG,"Failed to create coap\n"));
        return false;
    }

    _rdRes = new CoAP_RD_Resource(_coapWrapperPtr.get());

    if ((_rdRes->Create()) == 0)
    {
        ACE_DEBUG((LM_DEBUG,"Failed to craete rd resource\n"));
        return false;
    }

    _rdLookup = new CoAPRDLookUpResource(_coapWrapperPtr.get());

    if ((_rdLookup->Create()) == 0)
    {
        ACE_DEBUG((LM_DEBUG,"Failed to craete lookup resource\n"));
        return false;
    }

    ACE_Time_Value timeout;
    timeout.sec(5);

    _reactorPtr->register_handler(this, ACE_Event_Handler::READ_MASK);
    _reactorPtr->schedule_timer(this, 0, timeout);


    return true;
}

 Quit_Handler (ACE_Reactor *r) : ACE_Event_Handler (r) {
   SetConsoleMode (ACE_STDIN, ENABLE_LINE_INPUT
                              | ENABLE_ECHO_INPUT
                              | ENABLE_PROCESSED_INPUT);
   if (reactor ()->register_handler
         (this, quit_seen_.handle ()) == -1
       || ACE_Event_Handler::register_stdin_handler
            (this, r, ACE_Thread_Manager::instance ()) == -1)
     r->end_reactor_event_loop ();
 }

int TCPConnectionHandler::handle_output (ACE_HANDLE handle)
{
  ACE_UNUSED_ARG (handle);
  if (!buffers_)
    ACE_ERROR ((LM_ERROR,
                ACE_TEXT ("TCPConnectionHandler::handle_output call for empty buffers (%d)\n"), handle));
  if (0 > sendBuffers ()) // socket broken, kill yourself
    return -1;

  if (!buffers_) // everything already send, unregister
    {
      reactor ()->cancel_wakeup (this,
                                 ACE_Event_Handler::WRITE_MASK | ACE_Event_Handler::DONT_CALL);
      reactor ()->remove_handler (this,
                                  ACE_Event_Handler::WRITE_MASK  | ACE_Event_Handler::DONT_CALL);
    }

  return 0;
}

bool
OpenDDS::DCPS::ReliableMulticast::detail::ReactivePacketSender::open(
  )
{
  if (socket_.ACE_SOCK_Dgram::open(
    local_address_
    ) == -1)
  {
    logError("ReactivePacketSender: failure to open\n");
    return false;
  }


  ACE_INET_Addr address;
  if (this->socket_.get_local_addr (address) != 0)
  {
    ACE_ERROR_RETURN ((LM_ERROR,
      ACE_TEXT ("(%P|%t) ERROR: ReliableMulticast::open_socket ")
      ACE_TEXT ("- %p"),
      ACE_TEXT ("cannot get local addr\n")),
      false);
  }
  this->local_address_.set_port_number (address.get_port_number ());

  if (reactor()->register_handler(
    this,
    ACE_Event_Handler::READ_MASK
    ) == -1)
  {
    logError("ReactivePacketSender: failure to register_handler\n");
    return false;
  }
  if (reactor()->schedule_timer(
    this,
    0,
    ACE_Time_Value(1),
    ACE_Time_Value(1)
    ) == -1)
  {
    logError("ReactivePacketSender: failure to schedule_timer\n");
  }
  return true;
}

int AC_Input_Handler::close (u_long) {
  ACE_Message_Block *shutdown_message = 0;
  ACE_NEW_RETURN
    (shutdown_message,
     ACE_Message_Block (0, ACE_Message_Block::MB_STOP), -1);
  output_handler_->put (shutdown_message);

  reactor ()->remove_handler
    (connected_clients_, ACE_Event_Handler::READ_MASK);
  return output_handler_->wait ();
}

int KSG_Service_Handler::release()
{
	if(curr_blk_)
	{
		// 当接收数据一半的时候关闭连接，则需要释放资源
		curr_blk_->release();
		curr_blk_ = NULL;
	}
	reactor()->cancel_timer(this);
	//reactor()->remove_handler(this,ACE_Event_Handler::ALL_EVENTS_MASK);
	return 0;
}

int
Stop_Handler::handle_input (ACE_HANDLE handle)
{
  ACE_UNUSED_ARG (handle);

  ACE_DEBUG ((LM_INFO,
              ACE_TEXT ("(%P|%t) Stop_Handler::handle_input - entered\n")));

  for (size_t i = 0; i < HANDLERS_TO_STOP_TABLE_SIZE; ++i)
    {
      // remove from the reactor's tables all non-null entries
      if (this->handlers_to_stop_[i])
        {
#if defined ACE_HAS_EXCEPTIONS

          // protect from deleted pointer
          try
            {
#endif // ACE_HAS_EXCEPTIONS

              this->reactor ()->cancel_timer (this->handlers_to_stop_[i]);
              this->reactor ()->remove_handler
                (this->handlers_to_stop_[i],
                 ACE_Event_Handler::ALL_EVENTS_MASK
                 | ACE_Event_Handler::DONT_CALL);
#if defined ACE_HAS_EXCEPTIONS
            }
          catch (...)
            {
              ACE_ERROR ((LM_ERROR,
                          ACE_TEXT ("(%P|%t) Stop_Handler::handle_input - ")
                          ACE_TEXT ("EXCEPTION CATCHED. Most probably ")
                          ACE_TEXT ("handler's pointer has been deleted.\n")));
            }
#endif // ACE_HAS_EXCEPTIONS
          this->handlers_to_stop_[i] = 0;
        }
    }

  this->reactor ()->remove_handler (this->registered_signals_);

  if (reactor ()->end_reactor_event_loop () == -1)
    {
      ACE_ERROR_RETURN ((LM_DEBUG,
                         ACE_TEXT ("(%P|%t) Stop_Handler::handle_signal:%p\n"),
                         ACE_TEXT ("end_reactor_event_loop")),
                        -1);
    }

  ACE_DEBUG ((LM_INFO,
              ACE_TEXT ("(%P|%t) Stop_Handler::handle_input - completed.\n")));
  return 0;
}

void P2pEndpoint::close()
{
    const ACE_Reactor_Mask masks = ACE_Event_Handler::ALL_EVENTS_MASK |
        ACE_Event_Handler::DONT_CALL;
    reactor()->remove_handler(this, masks);

    clearAddresses();

    if (isOpened()) {
        udp_->close();
        NSRPC_LOG_DEBUG(ACE_TEXT("UDP socket closed"));
    }
}

int TCPAcceptorHandler::make_svc_handler (TCPConnectionHandler *& sh)
{
  sh = new TCPConnectionHandler (true);
  sh->reactor (reactor ());
  if (handlersRegister_->TCPServersNo () >= HandlersNo)
    ACE_ERROR ((LM_ERROR,
                ACE_TEXT ("TCPAcceptorHandler::make_svc_handler called to many times!\n")));
  else
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("TCPAcceptorHandler::make_svc_handler new TCP server created\n")));

  return 0;
}

int Logging_Event_Handler_Ex::open () {
  int result = PARENT::open ();
  if (result != -1) {
    ACE_Time_Value reschedule (max_client_timeout_.sec () / 4);
    result =
      reactor ()->schedule_timer
        (this,
         0,
         max_client_timeout_,  // Initial timeout.
         reschedule);          // Subsequent timeouts.
  }
  return result;
}

  ReactorTask::ReactorTask(ACE_Sched_Params *pschedp,
                           bool shutdownOnException,
                           int size) :
      schedp_(ACE_SCHED_OTHER, 0),
      reactor_(),
      shutdownOnException_(shutdownOnException)
  {
    MIRO_LOG_CTOR("Miro::ReactorTask");

    reactor(&reactor_);
    reactor_.open(size);
    if (pschedp)
      schedp_ = (*pschedp);
  }

ACE_INT32 Handle_data::open( )
{	
	ACE_INT32 ret = 0;
	ACE_INET_Addr remote_addr;	
	get_peer().get_remote_addr( remote_addr );		
	ACE_DEBUG( (LM_DEBUG, "the remote addr is %s\n", remote_addr.get_host_addr())  );
	
	ret = reactor()->register_handler( this, ACE_Event_Handler::READ_MASK );
	if (ret != -1)	
	{		
		ACE_DEBUG( (LM_DEBUG, "handle data register ok!\n")  );		
	}	
	return ret;
}

int Logging_Acceptor::handle_input(ACE_HANDLE)
{
    Logging_Event_Handler *peer_handler = 0;
    ACE_NEW_RETURN(peer_handler, Logging_Event_Handler(reactor()), -1);
    if (acceptor_.accept(peer_handler->peer()) == -1) {
        delete peer_handler;
        return -1;
    } else if (peer_handler->open() == -1) {
        peer_handler->handle_close();
        return -1;
    }

    return 0;
}

int KSG_Service_Handler::handle_timeout(const ACE_Time_Value &current_time, const void *act )
{
	if(current_time - this->request_time_ >= max_client_timeout_
		|| current_time - this->connect_time_ >= max_client_timeout_)
	{
		ACE_DEBUG((LM_INFO,"连接超时，关闭连接..."));
		
		reactor()->remove_handler(this,ACE_Event_Handler::ALL_EVENTS_MASK|ACE_Event_Handler::DONT_CALL);
		//reactor()->notify(this);
		closed_ = 1;
		return -1;
	}
	return 0;
}

int Ogre_TCPAccept_Hdl::handle_close ()
{
    //
    if (peer_acceptor_.get_handle () != ZCE_INVALID_SOCKET)
    {
        reactor()->remove_handler (this, true);
        peer_acceptor_.close ();
    }

    //删除自己
    delete this;

    return 0;
}

int WorldSocket::open (void *a)
{
    ACE_UNUSED_ARG (a);

    // Prevent double call to this func.
    if (m_OutBuffer)
        return -1;

    // This will also prevent the socket from being Updated
    // while we are initializing it.
    m_OutActive = true;

    // Hook for the manager.
    if (sWorldSocketMgr->OnSocketOpen(this) == -1)
        return -1;

    // Allocate the buffer.
    ACE_NEW_RETURN (m_OutBuffer, ACE_Message_Block (m_OutBufferSize), -1);

    // Store peer address.
    ACE_INET_Addr remote_addr;

    if (peer().get_remote_addr(remote_addr) == -1)
    {
        sLog->outError(LOG_FILTER_NETWORKIO, "WorldSocket::open: peer().get_remote_addr errno = %s", ACE_OS::strerror (errno));
        return -1;
    }

    m_Address = remote_addr.get_host_addr();

    // not an opcode. this packet sends raw string WORLD OF WARCRAFT CONNECTION - SERVER TO CLIENT"
    // because of our implementation, bytes "WO" become the opcode
    WorldPacket packet(MSG_VERIFY_CONNECTIVITY);
    packet << "RLD OF WARCRAFT CONNECTION - SERVER TO CLIENT";

    if (SendPacket(packet) == -1)
        return -1;

    // Register with ACE Reactor
    if (reactor()->register_handler(this, ACE_Event_Handler::READ_MASK | ACE_Event_Handler::WRITE_MASK) == -1)
    {
        sLog->outError(LOG_FILTER_NETWORKIO, "WorldSocket::open: unable to register client handler errno = %s", ACE_OS::strerror (errno));
        return -1;
    }

    // reactor takes care of the socket from now on
    remove_reference();

    return 0;
}

int
ACE_Process_Manager::handle_signal (int,
                                    siginfo_t *si,
                                    ucontext_t *)
{
#if defined (ACE_WIN32)
  ACE_HANDLE proc = si->si_handle_;
  ACE_exitcode status = 0;
  BOOL result = ::GetExitCodeProcess (proc,
                                      &status);
  if (result)
    {
      if (status != STILL_ACTIVE)
        {
          {
            ACE_MT (ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, ace_mon, lock_, -1));

            ssize_t i = this->find_proc (proc);
            if (i == -1)
              return -1;
#if 0
            pid_t pid = i != -1
              ? process_table_[i].process_->getpid ()
              : ACE_INVALID_PID;
#endif
            this->notify_proc_handler (i, status);
            this->remove_proc (i);
          }
          return -1; // remove this HANDLE/Event_Handler combination
        }
      else
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_LIB_TEXT ("Process still active")
                           ACE_LIB_TEXT (" -- shouldn't have been called yet!\n")),
                          0); // return 0 : stay registered
    }
  else
    {
      // <GetExitCodeProcess> failed.
      ACE_ERROR_RETURN ((LM_ERROR,
                         ACE_LIB_TEXT ("GetExitCodeProcess failed")),
                        -1); // return -1: unregister
    }
#else /* !ACE_WIN32 */
  ACE_UNUSED_ARG (si);
  return reactor ()->notify
    (this,
     ACE_Event_Handler::READ_MASK);
#endif /* !ACE_WIN32 */
}

int RealmSocket::handle_close(ACE_HANDLE h, ACE_Reactor_Mask)
{
    // As opposed to WorldSocket::handle_close, we don't need locks here.
    closing_ = true;

    if (h == ACE_INVALID_HANDLE)
        peer().close_writer();

    if (session_)
        session_->OnClose();

    reactor()->remove_handler(this, ACE_Event_Handler::DONT_CALL | ACE_Event_Handler::ALL_EVENTS_MASK);
    return 0;
}

bool RealmSocket::send(const char *buf, size_t len)
{
    if (buf == NULL || len == 0)
        return true;

    ACE_Data_Block db(
            len,
            ACE_Message_Block::MB_DATA,
            (const char*)buf,
            0,
            0,
            ACE_Message_Block::DONT_DELETE,
            0);

    ACE_Message_Block message_block(
            &db,
            ACE_Message_Block::DONT_DELETE,
            0);

    message_block.wr_ptr(len);

    if (msg_queue()->is_empty())
    {
        // Try to send it directly.
        ssize_t n = noblk_send(message_block);

        if (n < 0)
            return false;
        size_t un = size_t(n);
        if (un == len)
            return true;

        // fall down
        message_block.rd_ptr(un);
    }

    ACE_Message_Block *mb = message_block.clone();

    if (msg_queue()->enqueue_tail(mb, (ACE_Time_Value *) &ACE_Time_Value::zero) == -1)
    {
        mb->release();
        return false;
    }

    if (reactor()->schedule_wakeup(this, ACE_Event_Handler::WRITE_MASK) == -1)
        return false;

    return true;
}

int WorldSocket::open(void* a)
{
    ACE_UNUSED_ARG(a);

    // Prevent double call to this func.
    if (m_OutBuffer)
        { return -1; }

    // This will also prevent the socket from being Updated
    // while we are initializing it.
    m_OutActive = true;

    // Hook for the manager.
    if (sWorldSocketMgr->OnSocketOpen(this) == -1)
        { return -1; }

    // Allocate the buffer.
    ACE_NEW_RETURN(m_OutBuffer, ACE_Message_Block(m_OutBufferSize), -1);

    // Store peer address.
    ACE_INET_Addr remote_addr;

    if (peer().get_remote_addr(remote_addr) == -1)
    {
        sLog.outError("WorldSocket::open: peer ().get_remote_addr errno = %s", ACE_OS::strerror(errno));
        return -1;
    }

    m_Address = remote_addr.get_host_addr();

    // Send startup packet.
    WorldPacket packet(SMSG_AUTH_CHALLENGE, 4);
    packet << m_Seed;

    if (SendPacket(packet) == -1)
        { return -1; }

    // Register with ACE Reactor
    if (reactor()->register_handler(this, ACE_Event_Handler::READ_MASK | ACE_Event_Handler::WRITE_MASK) == -1)
    {
        sLog.outError("WorldSocket::open: unable to register client handler errno = %s", ACE_OS::strerror(errno));
        return -1;
    }

    // reactor takes care of the socket from now on
    remove_reference();

    return 0;
}