src/transport/TcpTransport.h

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef __TCPTRANSPORT_H__
#define __TCPTRANSPORT_H__

#include <boost/atomic.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include "dataBlock.h"

extern "C" {
#include "event2/buffer.h"
#include "event2/bufferevent.h"
#include "event2/event.h"
#include "event2/thread.h"
}

namespace rocketmq {
//<!***************************************************************************
typedef enum {
  e_connectInit = 0,
  e_connectWaitResponse = 1,
  e_connectSuccess = 2,
  e_connectFail = 3
} tcpConnectStatus;

typedef void (*READ_CALLBACK)(void *context, const MemoryBlock &,
                              const std::string &);
class TcpRemotingClient;
class TcpTransport {
 public:
  TcpTransport(TcpRemotingClient *pTcpRemointClient,
               READ_CALLBACK handle = NULL);
  virtual ~TcpTransport();

  tcpConnectStatus connect(const std::string &strServerURL,
                           int timeOutMillisecs = 3000);
  void disconnect(const std::string &addr);
  tcpConnectStatus waitTcpConnectEvent(int timeoutMillisecs = 3000);
  void setTcpConnectStatus(tcpConnectStatus connectStatus);
  tcpConnectStatus getTcpConnectStatus();
  bool sendMessage(const char *pData, int len);
  const std::string getPeerAddrAndPort();
  const uint64_t getStartTime() const;

 private:
  void messageReceived(const MemoryBlock &mem);
  static void readNextMessageIntCallback(struct bufferevent *bev, void *ctx);
  static void eventcb(struct bufferevent *bev, short what, void *ctx);
  static void timeoutcb(evutil_socket_t fd, short what, void *arg);
  void runThread();
  void clearBufferEventCallback();
  void freeBufferEvent();
  void exitBaseDispatch();
  void setTcpConnectEvent(tcpConnectStatus connectStatus);
  u_long getInetAddr(std::string &hostname);

 private:
  uint64_t m_startTime;
  boost::mutex m_socketLock;
  struct event_base *m_eventBase;
  struct bufferevent *m_bufferEvent;
  boost::atomic<tcpConnectStatus> m_tcpConnectStatus;
  boost::mutex m_connectEventLock;
  boost::condition_variable_any m_connectEvent;

  boost::atomic<bool> m_event_base_status;
  boost::mutex        m_event_base_mtx;
  boost::condition_variable_any m_event_base_cv;

  //<!read data thread
  boost::thread *m_ReadDatathread;

  //<! read data callback
  READ_CALLBACK m_readcallback;
  TcpRemotingClient *m_tcpRemotingClient;
};

//<!************************************************************************
}  //<!end namespace;

#endif