sysnc from master repo

src/common/NameSpaceUtil.cpp

@@ -28,7 +28,7 @@ bool NameSpaceUtil::isEndPointURL(string nameServerAddr) {
 }
 
 string NameSpaceUtil::formatNameServerURL(string nameServerAddr) {
-  unsigned int index = nameServerAddr.find(ENDPOINT_PREFIX);
+  auto index = nameServerAddr.find(ENDPOINT_PREFIX);
   if (index != string::npos) {
     LOG_DEBUG("Get Name Server from endpoint [%s]",
               nameServerAddr.substr(ENDPOINT_PREFIX_LENGTH, nameServerAddr.length() - ENDPOINT_PREFIX_LENGTH).c_str());

src/common/noncopyable.h

@@ -0,0 +1,33 @@
+/*
+ * 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 __NONCOPYABLE_H__
+#define __NONCOPYABLE_H__
+
+namespace rocketmq {
+
+class noncopyable {
+ protected:
+  noncopyable() = default;
+  ~noncopyable() = default;
+
+  noncopyable(const noncopyable&) = delete;
+  noncopyable& operator=(const noncopyable&) = delete;
+};
+
+}  // namespace rocketmq
+
+#endif  //__NONCOPYABLE_H__

src/protocol/CommandHeader.cpp

@@ -73,7 +73,7 @@ void SendMessageRequestHeader::Encode(Json::Value& outData) {
   outData["properties"] = properties;
   outData["reconsumeTimes"] = UtilAll::to_string(reconsumeTimes);
   outData["unitMode"] = UtilAll::to_string(unitMode);
-  // outData["batch"] = batch;
+  outData["batch"] = UtilAll::to_string(batch);
 }
 
 int SendMessageRequestHeader::getReconsumeTimes() {
@@ -106,7 +106,7 @@ void SendMessageRequestHeader::SetDeclaredFieldOfCommandHeader(map<string, strin
   requestMap.insert(pair<string, string>("properties", properties));
   requestMap.insert(pair<string, string>("reconsumeTimes", UtilAll::to_string(reconsumeTimes)));
   requestMap.insert(pair<string, string>("unitMode", UtilAll::to_string(unitMode)));
-  // requestMap.insert(pair<string, string>("batch", UtilAll::to_string(batch)));
+  requestMap.insert(pair<string, string>("batch", UtilAll::to_string(batch)));
 }
 
 //<!************************************************************************

src/transport/EventLoop.cpp

@@ -0,0 +1,241 @@
+/*
+ * 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.
+ */
+#include "EventLoop.h"
+
+#if !defined(WIN32) && !defined(__APPLE__)
+#include <sys/prctl.h>
+#endif
+
+#include <event2/thread.h>
+
+#include "Logging.h"
+#include "UtilAll.h"
+
+namespace rocketmq {
+
+EventLoop* EventLoop::GetDefaultEventLoop() {
+  static EventLoop defaultEventLoop;
+  return &defaultEventLoop;
+}
+
+EventLoop::EventLoop(const struct event_config* config, bool run_immediately)
+    : m_eventBase(nullptr), m_loopThread(nullptr), _is_running(false) {
+  // tell libevent support multi-threads
+#ifdef WIN32
+  evthread_use_windows_threads();
+#else
+  evthread_use_pthreads();
+#endif
+
+  if (config == nullptr) {
+    m_eventBase = event_base_new();
+  } else {
+    m_eventBase = event_base_new_with_config(config);
+  }
+
+  if (m_eventBase == nullptr) {
+    // failure...
+    LOG_ERROR("Failed to create event base!");
+    return;
+  }
+
+  evthread_make_base_notifiable(m_eventBase);
+
+  if (run_immediately) {
+    start();
+  }
+}
+
+EventLoop::~EventLoop() {
+  stop();
+
+  if (m_eventBase != nullptr) {
+    event_base_free(m_eventBase);
+    m_eventBase = nullptr;
+  }
+}
+
+void EventLoop::start() {
+  if (m_loopThread == nullptr) {
+    // start event loop
+#if !defined(WIN32) && !defined(__APPLE__)
+    string taskName = UtilAll::getProcessName();
+    prctl(PR_SET_NAME, "EventLoop", 0, 0, 0);
+#endif
+    m_loopThread = new std::thread(&EventLoop::runLoop, this);
+#if !defined(WIN32) && !defined(__APPLE__)
+    prctl(PR_SET_NAME, taskName.c_str(), 0, 0, 0);
+#endif
+  }
+}
+
+void EventLoop::stop() {
+  if (m_loopThread != nullptr /*&& m_loopThread.joinable()*/) {
+    _is_running = false;
+    m_loopThread->join();
+
+    delete m_loopThread;
+    m_loopThread = nullptr;
+  }
+}
+
+void EventLoop::runLoop() {
+  _is_running = true;
+
+  while (_is_running) {
+    int ret;
+
+    ret = event_base_dispatch(m_eventBase);
+    //    ret = event_base_loop(m_eventBase, EVLOOP_NONBLOCK);
+
+    if (ret == 1) {
+      // no event
+      std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+  }
+}
+
+#define OPT_UNLOCK_CALLBACKS (BEV_OPT_DEFER_CALLBACKS | BEV_OPT_UNLOCK_CALLBACKS)
+
+BufferEvent* EventLoop::createBufferEvent(socket_t fd, int options) {
+  struct bufferevent* event = bufferevent_socket_new(m_eventBase, fd, options);
+  if (event == nullptr) {
+    return nullptr;
+  }
+
+  bool unlock = (options & OPT_UNLOCK_CALLBACKS) == OPT_UNLOCK_CALLBACKS;
+
+  return new BufferEvent(event, unlock);
+}
+
+BufferEvent::BufferEvent(struct bufferevent* event, bool unlockCallbacks)
+    : m_bufferEvent(event),
+      m_unlockCallbacks(unlockCallbacks),
+      m_readCallback(nullptr),
+      m_writeCallback(nullptr),
+      m_eventCallback(nullptr),
+      m_callbackTransport() {
+#ifdef ROCKETMQ_BUFFEREVENT_PROXY_ALL_CALLBACK
+  if (m_bufferEvent != nullptr) {
+    bufferevent_setcb(m_bufferEvent, read_callback, write_callback, event_callback, this);
+  }
+#endif  // ROCKETMQ_BUFFEREVENT_PROXY_ALL_CALLBACK
+}
+
+BufferEvent::~BufferEvent() {
+  if (m_bufferEvent != nullptr) {
+    // free function will set all callbacks to NULL first.
+    bufferevent_free(m_bufferEvent);
+    m_bufferEvent = nullptr;
+  }
+}
+
+void BufferEvent::setCallback(BufferEventDataCallback readCallback,
+                              BufferEventDataCallback writeCallback,
+                              BufferEventEventCallback eventCallback,
+                              std::shared_ptr<TcpTransport> transport) {
+  // use lock in bufferevent
+  bufferevent_lock(m_bufferEvent);
+
+  // wrap callback
+  m_readCallback = readCallback;
+  m_writeCallback = writeCallback;
+  m_eventCallback = eventCallback;
+  m_callbackTransport = transport;
+
+#ifndef ROCKETMQ_BUFFEREVENT_PROXY_ALL_CALLBACK
+  bufferevent_data_cb readcb = readCallback != nullptr ? read_callback : nullptr;
+  bufferevent_data_cb writecb = writeCallback != nullptr ? write_callback : nullptr;
+  bufferevent_event_cb eventcb = eventCallback != nullptr ? event_callback : nullptr;
+
+  bufferevent_setcb(m_bufferEvent, readcb, writecb, eventcb, this);
+#endif  // ROCKETMQ_BUFFEREVENT_PROXY_ALL_CALLBACK
+
+  bufferevent_unlock(m_bufferEvent);
+}
+
+void BufferEvent::read_callback(struct bufferevent* bev, void* ctx) {
+  auto event = static_cast<BufferEvent*>(ctx);
+
+  if (event->m_unlockCallbacks)
+    bufferevent_lock(event->m_bufferEvent);
+
+  BufferEventDataCallback callback = event->m_readCallback;
+  std::shared_ptr<TcpTransport> transport = event->m_callbackTransport.lock();
+
+  if (event->m_unlockCallbacks)
+    bufferevent_unlock(event->m_bufferEvent);
+
+  if (callback) {
+    callback(event, transport.get());
+  }
+}
+
+void BufferEvent::write_callback(struct bufferevent* bev, void* ctx) {
+  auto event = static_cast<BufferEvent*>(ctx);
+
+  if (event->m_unlockCallbacks)
+    bufferevent_lock(event->m_bufferEvent);
+
+  BufferEventDataCallback callback = event->m_writeCallback;
+  std::shared_ptr<TcpTransport> transport = event->m_callbackTransport.lock();
+
+  if (event->m_unlockCallbacks)
+    bufferevent_unlock(event->m_bufferEvent);
+
+  if (callback) {
+    callback(event, transport.get());
+  }
+}
+
+static std::string buildPeerAddrPort(socket_t fd) {
+  sockaddr_in addr;
+  socklen_t len = sizeof(addr);
+
+  getpeername(fd, (struct sockaddr*)&addr, &len);
+
+  LOG_DEBUG("socket: %d, addr: %s, port: %d", fd, inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
+  std::string addrPort(inet_ntoa(addr.sin_addr));
+  addrPort.append(":");
+  addrPort.append(UtilAll::to_string(ntohs(addr.sin_port)));
+
+  return addrPort;
+}
+
+void BufferEvent::event_callback(struct bufferevent* bev, short what, void* ctx) {
+  auto event = static_cast<BufferEvent*>(ctx);
+
+  if (what & BEV_EVENT_CONNECTED) {
+    socket_t fd = event->getfd();
+    event->m_peerAddrPort = buildPeerAddrPort(fd);
+  }
+
+  if (event->m_unlockCallbacks)
+    bufferevent_lock(event->m_bufferEvent);
+
+  BufferEventEventCallback callback = event->m_eventCallback;
+  std::shared_ptr<TcpTransport> transport = event->m_callbackTransport.lock();
+
+  if (event->m_unlockCallbacks)
+    bufferevent_unlock(event->m_bufferEvent);
+
+  if (callback) {
+    callback(event, what, transport.get());
+  }
+}
+
+}  // namespace rocketmq