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