You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2018/11/20 11:06:06 UTC
[GitHub] vongosling closed pull request #8: Add a key build script to build
static and share library
vongosling closed pull request #8: Add a key build script to build static and share library
URL: https://github.com/apache/rocketmq-client-cpp/pull/8
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5052367..161085a 100755
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -38,13 +38,15 @@ set(CMAKE_CONFIGURATION_TYPES "Release")
set(CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/cmake)
set(CMAKE_ALLOW_LOOSE_LOOP_CONSTRUCTS ON)
set(CMAKE_VERBOSE_MAKEFILE 1)
+option(BUILD_ROCKETMQ_STATIC "build rocketmq-client static library" ON)
+option(BUILD_ROCKETMQ_SHARED "build rocketmq-client shared library" ON)
#Find dependency
set(BOOST_INCLUDEDIR /usr/local/include/)
#set(BOOST_INCLUDEDIR C:/boost_1_56_0/)
-set(Boost_USE_STATIC_LIBS ON) # only find static libs
-set(Boost_USE_MULTITHREADED ON)
-set(Boost_USE_STATIC_RUNTIME ON)
+option(Boost_USE_STATIC_LIBS "only find boost static libs" ON) # only find static libs
+set(Boost_USE_MULTITHREADED ON)
+set(Boost_USE_STATIC_RUNTIME ON)
if(WIN32)
find_package(Boost 1.56 REQUIRED COMPONENTS atomic thread system chrono date_time
log log_setup regex serialization filesystem locale iostreams zlib)
@@ -60,7 +62,7 @@ if(Boost_FOUND)
endif()
#set(LIBEVENT_INCLUDE_DIR /usr/local/include/)
-set(Libevent_USE_STATIC_LIBS OFF) # only find static libs
+option(Libevent_USE_STATIC_LIBS "only find libevent static libs" ON) # only find static libs
#set(Libevent_DIR /usr/local/lib/) # only find static libs
find_package(Libevent 2.0.22 REQUIRED COMPONENTS)
@@ -71,7 +73,7 @@ if(LIBEVENT_FOUND)
endif()
#set(JSONCPP_INCLUDE_DIRS C:/jsoncpp-0.10.6/include)
-set(JSONCPP_USE_STATIC_LIBS OFF) # only find static libs
+option(JSONCPP_USE_STATIC_LIBS "only find jsoncpp static libs" ON) # only find static libs
find_package(Jsoncpp 0.10.6)
if(JSONCPP_FOUND)
include_directories(${JSONCPP_INCLUDE_DIRS})
@@ -97,14 +99,15 @@ IF (WIN32)
set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} /MTd")
ELSE()
set(C_FLAGS
- -g
+ #-g
-Wall
-Wno-deprecated
-fPIC
-fno-strict-aliasing
)
set(CXX_FLAGS
- -g
+ #-g
+ -O2
-Wall
-Wno-deprecated
-fPIC
diff --git a/README.md b/README.md
index 911c152..844192a 100644
--- a/README.md
+++ b/README.md
@@ -21,11 +21,12 @@ RocketMQ-Client-CPP is the C/C++ client of Apache RocketMQ which is a distribute
- reliability, based on nameServer snapshot and network disaster recovery strategy, no real-time impact on publish and subscribe when anyone of broker or nameSrv was broken.
## Dependency ##
-- libevent 2.0.22
-- jsoncpp 0.10.6
+- [libevent 2.0.22](https://github.com/libevent/libevent/archive/release-2.0.22-stable.zip "libevent 2.0.22")
-- boost 1.56.0
+- [jsoncpp 0.10.6](https://github.com/open-source-parsers/jsoncpp/archive/0.10.6.zip "jsoncpp 0.10.6")
+
+- [boost 1.56.0](http://sourceforge.net/projects/boost/files/boost/1.56.0/boost_1_56_0.tar.gz "boost 1.56.0")
## Documentation ##
doc/rocketmq-cpp_manaual_zh.docx
@@ -34,43 +35,29 @@ doc/rocketmq-cpp_manaual_zh.docx
### Linux platform ###
-**note**: *make sure the following compile tools or libraries has been installed before install dependency libraries*
+**note**: *make sure the following compile tools or libraries with them minimum version number have been installed before run the build script build.sh*
-- compile tools: **gcc-c++**、**cmake**、**automake**、**libtool**
+- compile tools:
+ - gcc-c++ 4.8.2: jsoncpp,boost rocket-client require it, need support C++11
+ - cmake 2.8.0: jsoncpp,rocketmq-client require it
+ - automake 1.11.1: libevent require it
+ - libtool 2.2.6: libevent require it
-- libraries: **openssl-devel**、**bzip2-devel**
+- libraries:
+ - bzip2-devel 1.0.6: boost dependcy it
-#### Dependency Installation ####
+one key build script will automatic build the dependency libraries include libevent json and boost, then it will build rocketmq-client static and shared library.
-1. install [libevent 2.0.22](https://github.com/libevent/libevent/archive/release-2.0.22-stable.zip "libevent 2.0.22")
-```shell
-./autogen.sh
-./configure
-make
-make install
-```
+if can't get internet to download three library source files by build script, you can copy three library source files (release-2.0.22-stable.zip 0.10.6.zip and boost_1_56_0.tar.gz) to rocketmq-client root dir, then build.sh will auto use these library files to build rocketmq-client.
-1. install [JsonCPP 0.10.6](https://github.com/open-source-parsers/jsoncpp/archive/0.10.6.zip "jsoncpp 0.10.6")
-```shell
-cmake .
-make
-make install
-```
+ sudo sh build.sh
-1. install [boost 1.56.0](http://sourceforge.net/projects/boost/files/boost/1.56.0/boost_1_56_0.tar.gz "boost 1.56.0")
-```shell
-./bootstrap.sh
-./b2 link=shared runtime-link=shared cxxflags=" -fPIC"
-./b2 install
-```
-#### 2. Make and Install ####
-```shell
-cmake .
-make
-make install
-```
+then there are librocketmq.a and librocketmq.so in /usr/local/lib. for use them to build application you should link with following libraries: -lrocketmq -lpthread -lz -ldl -lrt.
+
+ g++ -o consumer_example consumer_example.cpp -L. -lrocketmq -lpthread -lz -ldl -lrt
+
+### Windows platform: ###
-### Windows platform ###
#### Dependency Installation
1. install [libevent 2.0.22](https://github.com/libevent/libevent/archive/release-2.0.22-stable.zip "libevent 2.0.22")
extract libevent to C:/libevent
diff --git a/build.sh b/build.sh
new file mode 100644
index 0000000..6e2922e
--- /dev/null
+++ b/build.sh
@@ -0,0 +1,212 @@
+#!/usr/bin/bash
+basepath=$(cd `dirname $0`; pwd)
+down_dir="${basepath}/tmp_down_dir"
+build_dir="${basepath}/tmp_build_dir"
+packet_dir="${basepath}/tmp_packet_dir"
+sys_lib_dir="/usr/local/lib"
+bin_dir="${basepath}/bin"
+fname_libevent="release-2.0.22-stable.zip"
+fname_jsoncpp="0.10.6.zip"
+fname_boost="boost_1_58_0.tar.gz"
+
+function Help()
+{
+ echo "=========================================one key build help============================================"
+ echo "sh build.sh [build libevent:0/1 default:1] [build json:0/1 default:1] [build boost:0/1 default:1]"
+ echo "usage: sh build.sh 1 1 1"
+ echo "[[build libevent]: 1: need build libevent lib, 0: no need build libevent lib; default:1]"
+ echo "[[build json]: 1: need build json lib, 0: no need build json lib; default:1]"
+ echo "[[build boost]: 1: need build boost lib, 0: no need build boost lib; default:1]"
+ echo "=========================================one key build help============================================"
+ echo ""
+}
+
+if [ $# -eq 3 ];then
+ if [ "$1" != "0" -a "$1" != "1" ];then
+ echo "unsupport para value $1, please see the help"
+ Help
+ exit 1
+ fi
+ if [ "$2" != "0" -a "$2" != "1" ];then
+ echo "unsupport para value $2, please see the help"
+ Help
+ exit 1
+ fi
+ if [ "$3" != "0" -a "$3" != "1" ];then
+ echo "unsupport para value $3, please see the help"
+ Help
+ exit 1
+ fi
+elif [ $# -gt 0 ];then
+ echo "the number of parameter must 0 or 3, please see the help"
+ Help
+ exit 1
+fi
+
+if [ $# -ge 1 ];then
+ need_build_libevent=$1
+else
+ need_build_libevent=1
+fi
+
+if [ $# -ge 2 ];then
+ need_build_jsoncpp=$2
+else
+ need_build_jsoncpp=1
+fi
+
+if [ $# -ge 3 ];then
+ need_build_boost=$3
+else
+ need_build_boost=1
+fi
+
+function PrintParams()
+{
+ echo "###########################################################################"
+ echo "need_build_libevent: ${need_build_libevent}, need_build_jsoncpp:${need_build_jsoncpp}, need_build_boost:${need_build_boost}"
+ echo "###########################################################################"
+ echo ""
+}
+
+function Prepare()
+{
+ if [ -e ${down_dir} ]
+ then
+ echo "${down_dir} is exist"
+ cd ${down_dir}
+ ls |grep -v ${fname_libevent} |grep -v ${fname_jsoncpp} | grep -v ${fname_boost} |xargs sudo rm -rf
+ else
+ mkdir -p ${down_dir}
+ fi
+
+ cd ${basepath}
+ if [ -e ${fname_libevent} ]
+ then
+ mv -f ${basepath}/${fname_libevent} ${down_dir}
+ fi
+
+ if [ -e ${fname_jsoncpp} ]
+ then
+ mv -f ${basepath}/${fname_jsoncpp} ${down_dir}
+ fi
+
+ if [ -e ${fname_boost} ]
+ then
+ mv -f ${basepath}/${fname_boost} ${down_dir}
+ fi
+
+ if [ -e ${build_dir} ]
+ then
+ echo "${build_dir} is exist"
+ sudo rm -rf ${build_dir}/*
+ else
+ mkdir -p ${build_dir}
+ fi
+
+ if [ -e ${packet_dir} ]
+ then
+ echo "${packet_dir} is exist"
+ sudo rm -rf ${packet_dir}/*
+ else
+ mkdir -p ${packet_dir}
+ fi
+}
+
+function BuildLibevent()
+{
+ if [ "${need_build_libevent}" == "0" ];then
+ echo "no need build libevent lib"
+ return 0
+ fi
+
+ cd ${down_dir}
+ if [ -e ${fname_libevent} ]
+ then
+ echo "${fname_libevent} is exist"
+ else
+ wget https://github.com/libevent/libevent/archive/${fname_libevent}
+ fi
+ unzip ${fname_libevent}
+ cd libevent-release-2.0.22-stable
+ ./autogen.sh
+
+ echo "build libevent static #####################"
+ ./configure --disable-openssl --enable-static=yes --enable-shared=no CFLAGS=-fPIC CPPFLAGS=-fPIC
+ make
+ sudo make install
+}
+
+
+function BuildJsonCPP()
+{
+ if [ "${need_build_jsoncpp}" == "0" ];then
+ echo "no need build jsoncpp lib"
+ return 0
+ fi
+
+ cd ${down_dir}
+
+ if [ -e ${fname_jsoncpp} ]
+ then
+ echo "${fname_jsoncpp} is exist"
+ else
+ wget https://github.com/open-source-parsers/jsoncpp/archive/${fname_jsoncpp}
+ fi
+ unzip ${fname_jsoncpp}
+ cd jsoncpp-0.10.6
+
+ mkdir build; cd build
+ echo "build jsoncpp static #####################"
+ cmake .. -DCMAKE_CXX_FLAGS=-fPIC -DBUILD_STATIC_LIBS=ON -DBUILD_SHARED_LIBS=OFF
+ make
+ sudo make install
+}
+
+function BuildBoost()
+{
+ if [ "${need_build_boost}" == "0" ];then
+ echo "no need build boost lib"
+ return 0
+ fi
+
+ cd ${down_dir}
+ if [ -e ${fname_boost} ]
+ then
+ echo "${fname_boost} is exist"
+ else
+ wget http://sourceforge.net/projects/boost/files/boost/1.58.0/${fname_boost}
+ fi
+ tar -zxvf ${fname_boost}
+ cd boost_1_58_0
+ ./bootstrap.sh
+ echo "build boost static #####################"
+ sudo ./b2 cflags=-fPIC cxxflags=-fPIC --with-atomic --with-thread --with-system --with-chrono --with-date_time --with-log --with-regex --with-serialization --with-filesystem --with-locale --with-iostreams threading=multi link=static runtime-link=static release install
+}
+
+function BuildRocketMQClient()
+{
+ cd ${build_dir}
+ cmake ..
+ make
+ sudo make install
+
+ PackageRocketMQStatic
+}
+
+function PackageRocketMQStatic()
+{
+ #packet libevent,jsoncpp,boost,rocketmq,Signature to one librocketmq.a
+ ar -M < ${basepath}/package_rocketmq.mri
+ sudo rm -rf ${sys_lib_dir}/librocketmq.a
+ sudo rm -rf ${sys_lib_dir}/libSignature.a
+ cp -f librocketmq.a ${bin_dir}
+ sudo cp -f librocketmq.a ${sys_lib_dir}/
+}
+
+PrintParams
+Prepare
+BuildLibevent
+BuildJsonCPP
+BuildBoost
+BuildRocketMQClient
diff --git a/example/CMakeLists.txt b/example/CMakeLists.txt
index 04bf9b8..c6dc462 100755
--- a/example/CMakeLists.txt
+++ b/example/CMakeLists.txt
@@ -24,6 +24,9 @@ link_directories(${Boost_LIBRARY_DIRS})
link_directories(${LIBEVENT_LIBRARY})
link_directories(${JSONCPP_LIBRARY})
+if (BUILD_ROCKETMQ_SHARED)
+ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DBOOST_ALL_DYN_LINK -shared ")
+endif()
file(GLOB files "*.c*")
foreach(file ${files})
@@ -36,6 +39,21 @@ foreach(file ${files})
set_target_properties( ${basename} PROPERTIES LINK_FLAGS "/NODEFAULTLIB:LIBCMTD" )
endif()
endif()
- target_link_libraries (${basename} rocketmq_static ${deplibs}
- ${Boost_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES})
+
+ if (MSVC)
+ if (BUILD_ROCKETMQ_SHARED)
+ target_link_libraries (${basename} rocketmq_shared ${deplibs}
+ ${Boost_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES})
+ else()
+ target_link_libraries (${basename} rocketmq_static ${deplibs}
+ ${Boost_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES})
+ endif()
+ else()
+ #if (BUILD_ROCKETMQ_SHARED)
+ target_link_libraries (${basename} rocketmq_shared ${deplibs})
+ #else()
+ # target_link_libraries (${basename} rocketmq_static ${deplibs})
+ #endif()
+ endif()
+
endforeach()
diff --git a/include/MQMessage.h b/include/MQMessage.h
index fc18ac9..ec637ca 100755
--- a/include/MQMessage.h
+++ b/include/MQMessage.h
@@ -92,8 +92,23 @@ class ROCKETMQCLIENT_API MQMessage {
static const std::string PROPERTY_PRODUCER_GROUP;
static const std::string PROPERTY_MIN_OFFSET;
static const std::string PROPERTY_MAX_OFFSET;
- static const std::string KEY_SEPARATOR;
+
+ static const std::string PROPERTY_BUYER_ID;
+ static const std::string PROPERTY_ORIGIN_MESSAGE_ID;
+ static const std::string PROPERTY_TRANSFER_FLAG;
+ static const std::string PROPERTY_CORRECTION_FLAG;
+ static const std::string PROPERTY_MQ2_FLAG;
+ static const std::string PROPERTY_RECONSUME_TIME;
+ static const std::string PROPERTY_MSG_REGION;
+ static const std::string PROPERTY_TRACE_SWITCH;
+ static const std::string PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX;
+ static const std::string PROPERTY_MAX_RECONSUME_TIMES;
+ static const std::string PROPERTY_CONSUME_START_TIMESTAMP;
+ static const std::string PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET;
+ static const std::string PROPERTY_TRANSACTION_CHECK_TIMES;
+ static const std::string PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS;
+ static const std::string KEY_SEPARATOR;
private:
std::string m_topic;
int m_flag;
diff --git a/include/SendResult.h b/include/SendResult.h
index 0f51854..23a2dfc 100755
--- a/include/SendResult.h
+++ b/include/SendResult.h
@@ -34,7 +34,7 @@ enum SendStatus {
class ROCKETMQCLIENT_API SendResult {
public:
SendResult();
- SendResult(const SendStatus& sendStatus, const std::string& msgId,
+ SendResult(const SendStatus& sendStatus, const std::string& msgId, const std::string& offsetMsgId,
const MQMessageQueue& messageQueue, int64 queueOffset);
virtual ~SendResult();
@@ -42,6 +42,7 @@ class ROCKETMQCLIENT_API SendResult {
SendResult& operator=(const SendResult& other);
const std::string& getMsgId() const;
+ const std::string& getOffsetMsgId() const;
SendStatus getSendStatus() const;
MQMessageQueue getMessageQueue() const;
int64 getQueueOffset() const;
@@ -49,6 +50,7 @@ class ROCKETMQCLIENT_API SendResult {
private:
SendStatus m_sendStatus;
std::string m_msgId;
+ std::string m_offsetMsgId;
MQMessageQueue m_messageQueue;
int64 m_queueOffset;
};
diff --git a/libs/signature/CMakeLists.txt b/libs/signature/CMakeLists.txt
index 84ca0c5..6a2aa1f 100755
--- a/libs/signature/CMakeLists.txt
+++ b/libs/signature/CMakeLists.txt
@@ -23,3 +23,7 @@ aux_source_directory(src/ DIR_LIB_SRCS)
add_library(Signature STATIC ${DIR_LIB_SRCS})
target_link_libraries(Signature ${deplibs})
set_target_properties(Signature PROPERTIES OUTPUT_NAME "Signature")
+
+# install
+install (TARGETS Signature DESTINATION lib)
+#install (DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/include/ DESTINATION include/rocketmq)
diff --git a/package_rocketmq.mri b/package_rocketmq.mri
new file mode 100644
index 0000000..a233b5a
--- /dev/null
+++ b/package_rocketmq.mri
@@ -0,0 +1,22 @@
+create librocketmq.a
+addlib /usr/local/lib/libboost_chrono.a
+addlib /usr/local/lib/libboost_date_time.a
+addlib /usr/local/lib/libboost_filesystem.a
+addlib /usr/local/lib/libboost_iostreams.a
+addlib /usr/local/lib/libboost_locale.a
+addlib /usr/local/lib/libboost_log.a
+addlib /usr/local/lib/libboost_log_setup.a
+addlib /usr/local/lib/libboost_regex.a
+addlib /usr/local/lib/libboost_serialization.a
+addlib /usr/local/lib/libboost_system.a
+addlib /usr/local/lib/libboost_thread.a
+addlib /usr/local/lib/libboost_wserialization.a
+addlib /usr/local/lib/libevent.a
+addlib /usr/local/lib/libevent_core.a
+addlib /usr/local/lib/libevent_extra.a
+addlib /usr/local/lib/libevent_pthreads.a
+addlib /usr/local/lib/libjsoncpp.a
+addlib /usr/local/lib/librocketmq.a
+addlib /usr/local/lib/libSignature.a
+save
+end
diff --git a/project/CMakeLists.txt b/project/CMakeLists.txt
index 30b8fba..7921b9c 100755
--- a/project/CMakeLists.txt
+++ b/project/CMakeLists.txt
@@ -42,26 +42,33 @@ foreach(dir ${LIB_DIRS})
endforeach()
# static
-add_library(rocketmq_static STATIC ${SRC_FILES})
-set_target_properties(rocketmq_static PROPERTIES OUTPUT_NAME "rocketmq")
-add_dependencies(rocketmq_static Signature)
-target_link_libraries(rocketmq_static ${deplibs})
-target_link_libraries(rocketmq_static Signature)
+if(BUILD_ROCKETMQ_STATIC)
+ add_library(rocketmq_static STATIC ${SRC_FILES})
+ set_target_properties(rocketmq_static PROPERTIES OUTPUT_NAME "rocketmq")
+ add_dependencies(rocketmq_static Signature)
+ target_link_libraries(rocketmq_static ${deplibs})
+ target_link_libraries(rocketmq_static Signature)
+endif()
# shared
-set(CMAKE_SHARED_LINKER_FLAGS "-DBOOST_ALL_DYN_LINK -shared ")
-add_library(rocketmq_shared SHARED ${SRC_FILES})
-set_target_properties(rocketmq_shared PROPERTIES OUTPUT_NAME "rocketmq")
-add_dependencies(rocketmq_shared Signature)
-target_link_libraries(rocketmq_shared ${deplibs})
-target_link_libraries(rocketmq_shared Signature)
-target_link_libraries(rocketmq_shared jsoncpp)
-target_link_libraries(rocketmq_shared ${LIBEVENT_LIBRARIES})
-target_link_libraries(rocketmq_shared ${Boost_LIBRARIES})
-
+if(BUILD_ROCKETMQ_SHARED)
+ set(CMAKE_SHARED_LINKER_FLAGS "-DBOOST_ALL_DYN_LINK -shared ")
+ add_library(rocketmq_shared SHARED ${SRC_FILES})
+ set_target_properties(rocketmq_shared PROPERTIES OUTPUT_NAME "rocketmq")
+ add_dependencies(rocketmq_shared Signature)
+ target_link_libraries(rocketmq_shared ${deplibs})
+ target_link_libraries(rocketmq_shared Signature)
+ target_link_libraries(rocketmq_shared jsoncpp)
+ target_link_libraries(rocketmq_shared ${LIBEVENT_LIBRARIES})
+ target_link_libraries(rocketmq_shared ${Boost_LIBRARIES})
+endif()
# install
-install (TARGETS rocketmq_static DESTINATION lib)
-install (TARGETS rocketmq_shared DESTINATION lib)
+if(BUILD_ROCKETMQ_STATIC)
+ install (TARGETS rocketmq_static DESTINATION lib)
+endif()
+if(BUILD_ROCKETMQ_SHARED)
+ install (TARGETS rocketmq_shared DESTINATION lib)
+endif()
install (DIRECTORY ${CMAKE_SOURCE_DIR}/include/ DESTINATION include/rocketmq)
install (DIRECTORY ${CMAKE_SOURCE_DIR}/doc/ DESTINATION doc)
diff --git a/src/MQClientAPIImpl.cpp b/src/MQClientAPIImpl.cpp
index 287230c..6217086 100644
--- a/src/MQClientAPIImpl.cpp
+++ b/src/MQClientAPIImpl.cpp
@@ -514,7 +514,8 @@ SendResult MQClientAPIImpl::processSendResponse(const string& brokerName,
(SendMessageResponseHeader*)pResponse->getCommandHeader();
MQMessageQueue messageQueue(msg.getTopic(), brokerName,
responseHeader->queueId);
- return SendResult(sendStatus, responseHeader->msgId, messageQueue,
+ string unique_msgId = msg.getProperty(MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
+ return SendResult(sendStatus, unique_msgId, responseHeader->msgId, messageQueue,
responseHeader->queueOffset);
}
LOG_ERROR("processSendResponse error remark:%s, error code:%d",
diff --git a/src/message/MQMessage.cpp b/src/message/MQMessage.cpp
index db5487f..5f3812e 100755
--- a/src/message/MQMessage.cpp
+++ b/src/message/MQMessage.cpp
@@ -31,6 +31,21 @@ const string MQMessage::PROPERTY_TRANSACTION_PREPARED = "TRAN_MSG";
const string MQMessage::PROPERTY_PRODUCER_GROUP = "PGROUP";
const string MQMessage::PROPERTY_MIN_OFFSET = "MIN_OFFSET";
const string MQMessage::PROPERTY_MAX_OFFSET = "MAX_OFFSET";
+const string MQMessage::PROPERTY_BUYER_ID = "BUYER_ID";
+const string MQMessage::PROPERTY_ORIGIN_MESSAGE_ID = "ORIGIN_MESSAGE_ID";
+const string MQMessage::PROPERTY_TRANSFER_FLAG = "TRANSFER_FLAG";
+const string MQMessage::PROPERTY_CORRECTION_FLAG = "CORRECTION_FLAG";
+const string MQMessage::PROPERTY_MQ2_FLAG = "MQ2_FLAG";
+const string MQMessage::PROPERTY_RECONSUME_TIME = "RECONSUME_TIME";
+const string MQMessage::PROPERTY_MSG_REGION = "MSG_REGION";
+const string MQMessage::PROPERTY_TRACE_SWITCH = "TRACE_ON";
+const string MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX = "UNIQ_KEY";
+const string MQMessage::PROPERTY_MAX_RECONSUME_TIMES = "MAX_RECONSUME_TIMES";
+const string MQMessage::PROPERTY_CONSUME_START_TIMESTAMP = "CONSUME_START_TIME";
+const string MQMessage::PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET = "TRAN_PREPARED_QUEUE_OFFSET";
+const string MQMessage::PROPERTY_TRANSACTION_CHECK_TIMES = "TRANSACTION_CHECK_TIMES";
+const string MQMessage::PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS = "CHECK_IMMUNITY_TIME_IN_SECONDS";
+
const string MQMessage::KEY_SEPARATOR = " ";
//<!************************************************************************
MQMessage::MQMessage() { Init("", "", "", 0, "", true); }
diff --git a/src/producer/DefaultMQProducer.cpp b/src/producer/DefaultMQProducer.cpp
index 9e32f3c..6811590 100755
--- a/src/producer/DefaultMQProducer.cpp
+++ b/src/producer/DefaultMQProducer.cpp
@@ -29,6 +29,7 @@
#include "MessageSysFlag.h"
#include "TopicPublishInfo.h"
#include "Validators.h"
+#include "StringIdMaker.h"
namespace rocketmq {
@@ -333,8 +334,12 @@ SendResult DefaultMQProducer::sendKernelImpl(MQMessage& msg,
if (!brokerAddr.empty()) {
try {
- LOG_DEBUG("produce before:%s to %s", msg.toString().c_str(),
- mq.toString().c_str());
+ //msgId is produced by client, offsetMsgId produced by broker. (same with java sdk)
+ string unique_id = StringIdMaker::get_mutable_instance().get_unique_id();
+ msg.setProperty(MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, unique_id);
+
+ LOG_DEBUG("produce before:%s to %s", msg.toString().c_str(), mq.toString().c_str());
+
int sysFlag = 0;
if (tryToCompressMessage(msg)) {
sysFlag |= MessageSysFlag::CompressedFlag;
diff --git a/src/producer/SendResult.cpp b/src/producer/SendResult.cpp
index 7fd844e..7bc31ca 100755
--- a/src/producer/SendResult.cpp
+++ b/src/producer/SendResult.cpp
@@ -22,16 +22,18 @@ namespace rocketmq {
//<!***************************************************************************
SendResult::SendResult() : m_sendStatus(SEND_OK), m_queueOffset(0) {}
-SendResult::SendResult(const SendStatus& sendStatus, const string& msgId,
+SendResult::SendResult(const SendStatus& sendStatus, const std::string& msgId, const std::string& offsetMsgId,
const MQMessageQueue& messageQueue, int64 queueOffset)
: m_sendStatus(sendStatus),
m_msgId(msgId),
+ m_offsetMsgId(offsetMsgId),
m_messageQueue(messageQueue),
m_queueOffset(queueOffset) {}
SendResult::SendResult(const SendResult& other) {
m_sendStatus = other.m_sendStatus;
m_msgId = other.m_msgId;
+ m_offsetMsgId = other.m_offsetMsgId;
m_messageQueue = other.m_messageQueue;
m_queueOffset = other.m_queueOffset;
}
@@ -40,6 +42,7 @@ SendResult& SendResult::operator=(const SendResult& other) {
if (this != &other) {
m_sendStatus = other.m_sendStatus;
m_msgId = other.m_msgId;
+ m_offsetMsgId = other.m_offsetMsgId;
m_messageQueue = other.m_messageQueue;
m_queueOffset = other.m_queueOffset;
}
@@ -50,6 +53,8 @@ SendResult::~SendResult() {}
const string& SendResult::getMsgId() const { return m_msgId; }
+const string& SendResult::getOffsetMsgId() const { return m_offsetMsgId; }
+
SendStatus SendResult::getSendStatus() const { return m_sendStatus; }
MQMessageQueue SendResult::getMessageQueue() const { return m_messageQueue; }
diff --git a/src/producer/StringIdMaker.cpp b/src/producer/StringIdMaker.cpp
new file mode 100644
index 0000000..c556cb6
--- /dev/null
+++ b/src/producer/StringIdMaker.cpp
@@ -0,0 +1,145 @@
+#include "StringIdMaker.h"
+#include <unistd.h>
+#include <boost/asio.hpp>
+
+namespace rocketmq {
+StringIdMaker::StringIdMaker() {
+ memset(_buff, 0, sizeof(_buff));
+ memset(_0x_buff, 0, sizeof(_0x_buff));
+ srand((uint32_t)time(NULL));
+ init_prefix();
+}
+StringIdMaker::~StringIdMaker() {}
+
+void StringIdMaker::init_prefix() {
+ uint32_t pid = getpid();
+ uint32_t ip = get_ip();
+ uint32_t random_num = (rand() % 0xFFFF);
+
+ memcpy(_buff + 2, &pid, 4);
+ memcpy(_buff, &ip, 4);
+ memcpy(_buff + 6, &random_num, 4);
+
+ hexdump(_buff, _0x_buff, 10);
+
+ set_start_and_next_tm();
+}
+
+uint32_t StringIdMaker::get_ip() {
+ char name[1024];
+ boost::system::error_code ec;
+ if (boost::asio::detail::socket_ops::gethostname(name, sizeof(name), ec) != 0) {
+ return 0;
+ }
+
+ boost::asio::io_service io_service;
+ boost::asio::ip::tcp::resolver resolver(io_service);
+ boost::asio::ip::tcp::resolver::query query(name, "");
+ boost::system::error_code error;
+ boost::asio::ip::tcp::resolver::iterator iter = resolver.resolve(query, error);
+ if (error) {
+ return 0;
+ }
+ boost::asio::ip::tcp::resolver::iterator end; // End marker.
+ boost::asio::ip::tcp::endpoint ep;
+ while (iter != end) {
+ ep = *iter++;
+ }
+ std::string s_localIpAddress = ep.address().to_string();
+
+ int a[4];
+ std::string IP = s_localIpAddress;
+ std::string strTemp;
+ size_t pos;
+ size_t i = 3;
+
+ do {
+ pos = IP.find(".");
+
+ if (pos != std::string::npos) {
+ strTemp = IP.substr(0, pos);
+ a[i] = atoi(strTemp.c_str());
+ i--;
+ IP.erase(0, pos + 1);
+ } else {
+ strTemp = IP;
+ a[i] = atoi(strTemp.c_str());
+ break;
+ }
+
+ } while (1);
+
+ uint32_t nResult = (a[3] << 24) + (a[2] << 16) + (a[1] << 8) + a[0];
+ return nResult;
+}
+
+uint64_t StringIdMaker::get_curr_ms() {
+ struct timeval time_now;
+ gettimeofday(&time_now, NULL);
+ uint64_t ms_time = time_now.tv_sec * 1000 + time_now.tv_usec / 1000;
+ return ms_time;
+}
+
+void StringIdMaker::set_start_and_next_tm() {
+ time_t tmNow = time(NULL);
+ tm *ptmNow = localtime(&tmNow);
+ tm mon_begin;
+ mon_begin.tm_year = ptmNow->tm_year;
+ mon_begin.tm_mon = ptmNow->tm_mon;
+ mon_begin.tm_mday = 0;
+ mon_begin.tm_hour = 0;
+ mon_begin.tm_min = 0;
+ mon_begin.tm_sec = 0;
+
+ tm mon_next_begin;
+ if (ptmNow->tm_mon == 12) {
+ mon_next_begin.tm_year = ptmNow->tm_year + 1;
+ mon_next_begin.tm_mon = 1;
+ } else {
+ mon_next_begin.tm_year = ptmNow->tm_year;
+ mon_next_begin.tm_mon = ptmNow->tm_mon + 1;
+ }
+ mon_next_begin.tm_mday = 0;
+ mon_next_begin.tm_hour = 0;
+ mon_next_begin.tm_min = 0;
+ mon_next_begin.tm_sec = 0;
+
+ time_t mon_begin_tm = mktime(&mon_begin);
+ time_t mon_end_tm = mktime(&mon_next_begin);
+
+ _start_tm = mon_begin_tm * 1000;
+ _next_start_tm = mon_end_tm * 1000;
+}
+
+int StringIdMaker::atomic_incr(int id) {
+ __sync_add_and_fetch(&id, 1);
+ return id;
+}
+std::string StringIdMaker::get_unique_id() {
+ uint64_t now_time = get_curr_ms();
+
+ if (now_time > _next_start_tm) {
+ set_start_and_next_tm();
+ }
+ uint32_t tm_period = now_time - _start_tm;
+ seqid = atomic_incr(seqid) & 0xFF;
+
+ std::size_t prifix_len = 10; // 10 = prefix len
+ unsigned char *write_index = _buff + prifix_len;
+
+ memcpy(write_index, &tm_period, 4);
+ write_index += 4;
+
+ memcpy(write_index, &seqid, 2);
+
+ hexdump(_buff + prifix_len, (_0x_buff + (2 * prifix_len)), 6);
+ _0x_buff[32] = '\0';
+ return std::string(_0x_buff);
+}
+
+void StringIdMaker::hexdump(unsigned char *buffer, char *out_buff, unsigned long index) {
+ for (unsigned long i = 0; i < index; i++) {
+ sprintf(out_buff + 2 * i, "%02X ", buffer[i]);
+ }
+}
+}
diff --git a/src/producer/StringIdMaker.h b/src/producer/StringIdMaker.h
new file mode 100644
index 0000000..fc52f5a
--- /dev/null
+++ b/src/producer/StringIdMaker.h
@@ -0,0 +1,44 @@
+/*
+ ip: 4
+ pid: 4
+ 随机数 :2
+ 时间：4
+ 自增数:2
+*/
+#ifndef __STRINGID_MAKER_H__
+#define __STRINGID_MAKER_H__
+
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/time.h>
+#include <time.h>
+#include <unistd.h>
+#include <boost/serialization/singleton.hpp>
+#include <string>
+
+namespace rocketmq {
+class StringIdMaker : public boost::serialization::singleton<StringIdMaker> {
+ public:
+ StringIdMaker();
+ ~StringIdMaker();
+ std::string get_unique_id();
+
+ private:
+ uint32_t get_ip();
+ void init_prefix();
+ uint64_t get_curr_ms();
+ int atomic_incr(int id);
+ void set_start_and_next_tm();
+
+ void hexdump(unsigned char *buffer, char *out_buff, unsigned long index);
+
+ private:
+ uint64_t _start_tm;
+ uint64_t _next_start_tm;
+ unsigned char _buff[16];
+ char _0x_buff[33];
+ int16_t seqid;
+};
+}
+#endif
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services