Use OpenSSL to add support for SSL communication with the server. Script add debug mode support. (#273)
* Modification item:
（1）Add OpenSSL and debug options to the ‘build.sh’ script.
（2）Add the ‘ENABLE_OPENSSL’ macro to cmakelist.txt.
（3）Link the ‘libevent_openssl.a’ static library to support OpenSSL.
* Modification item:
（1）Link static libraries on demand.
（2）Add OpenSSL support and use the macro ‘ENABLE_OPENSSL’ to control OpenSSL compilation blocks.
（3）Inline member defaults. The default value of the inline member is intuitive and error free, and the redundant structure list can be removed, which is also recommended.
（4）Add ignore for vscode configuration in .gitignore.
* Add debugging information in debug mode.
* Modification item:
（1）Add C compiler debugging options.
（2）Remove the compiler debugging option from the child test cmake, which will result in inconsistent behavior between the upper and lower cmakes.
（3）Move the SSL initialization context into the constructor so that it is constructed only once and the memory leak in the last commit code is fixed.
（4）SSL handle is managed to bufferevent without memory management. Forced release will result in double-free.
* Modification item:
（1）Extract SSL context initialization into a function.
（2）Customize std::unique_ptr to manage SSL content and strengthen memory leak management.
（3）Add the default certificate configuration path.
（4）Add environment variables to dynamically modify the certificate configuration path.
（5）Enhance friendly output prompt.
（6）The communication function with the server is verified by self signed certificate.
（7）Asan and lsan tests have been passed.
* Modification item:
（1）Add OpenSSL library and compile script.
（2）Remove the system dynamic library connection, and use the project OpenSSL static library link instead.
（3）Fix some errors in CMakeList.txt.
* Modification item:
（1）Add OpenSSL library and compile script.
（2）Remove the system dynamic library connection, and use the project OpenSSL static library link instead.
（3）Fix some errors in CMakeList.txt.
* Modification item:
（1）Remove the OpenSSL option that controls whether or not it is required.
（2）Remove the ENABLE_OPENSSL macro. OpenSSL is supported by default.
（3）Add API to control whether OpenSSL support is enabled.
（4）Add the ReadProperties utility function.
（5）Add certificate related configuration in the properties configuration file.
（6）Add friendly output prompts.
* By mistake, I wrote a letter ‘t’ short.
* Modification item:
（1）Format the code with the 'format.sh' script.
（2）SSL is turned off by default.
* change openssl static library install path.
* using built-in openssl to compile libevent.
* Modification item:
（1）Moving/Returning temporary object prevents copy elision, which may cause RVO and NRVO are invalid technologies.
（2）Libevent only verifies whether the OpenSSL dynamic library has 'SSL_new' function as the basis for the existence of OpenSSL.
（3）Add a dependent static library to MRI.
* remove comments for code format.
* Fix the missing packing problem of static library packing script under Mac OS. For example, there is buffer. o in libcrypto. a, and buffer. o in libevent_core. a, but the symbols of the two are complementary. The method of using 'ar x' and 'ar cru' will result in the loss of symbol in one of the above two buffer. o.
* chors(style): format code style for eventloop.diff --git a/.gitignore b/.gitignore
index 22c7350..0640d20 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,4 @@
libs/signature/lib
tmp_*
Testing
+.vscode
diff --git a/CMakeLists.txt b/CMakeLists.txt
index a2b178c..baa227d 100755
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -44,6 +44,24 @@
option(BUILD_ROCKETMQ_STATIC "build rocketmq-client static library" ON)
option(BUILD_ROCKETMQ_SHARED "build rocketmq-client shared library" ON)
+option(OPENSSL_USE_STATIC_LIBS "only find openssl static libs" ON) # only find static libs
+if (WIN32)
+ find_package(OpenSSL 1.1.1 REQUIRED COMPONENTS)
+ if (OPENSSL_FOUND)
+ include_directories(${OPENSSL_INCLUDE_DIR})
+ message(STATUS "** OpenSSL Include dir: ${OPENSSL_INCLUDE_DIR}")
+ message(STATUS "** OpenSSL Libraries: ${OPENSSL_LIBRARIES}")
+ endif ()
+else ()
+ #find_package(OpenSSL 1.1.1 REQUIRED COMPONENTS)
+ set(OPENSSL_INCLUDE_DIR ${PROJECT_SOURCE_DIR}/bin/include)
+ set(OPENSSL_LIBRARIES_DIR ${PROJECT_SOURCE_DIR}/bin/lib)
+ set(OPENSSL_LIBRARIES ${OPENSSL_LIBRARIES_DIR}/libssl.a;${OPENSSL_LIBRARIES_DIR}/libcrypto.a)
+ include_directories(${OPENSSL_INCLUDE_DIR})
+endif ()
+message(STATUS "** OpenSSL_INCLUDE_DIR: ${OPENSSL_INCLUDE_DIR}")
+message(STATUS "** OpenSSL_LIBRARIES: ${OPENSSL_LIBRARIES}")
+
#Find dependency
option(Boost_USE_STATIC_LIBS "only find boost static libs" ON) # only find static libs
set(Boost_USE_MULTITHREADED ON)
@@ -61,11 +79,11 @@
#find_package(Boost 1.56 REQUIRED COMPONENTS atomic thread system chrono date_time log log_setup regex serialization filesystem locale iostreams)
set(Boost_INCLUDE_DIR ${PROJECT_SOURCE_DIR}/bin/include)
set(Boost_LIBRARY_DIRS ${PROJECT_SOURCE_DIR}/bin/lib)
- set(Boost_LIBRARIES ${Boost_LIBRARY_DIRS}/libboost_atomic.a;${Boost_LIBRARY_DIRS}/libboost_thread.a;${Boost_LIBRARY_DIRS}/libboost_system.a;${Boost_LIBRARY_DIRS}/libboost_chrono.a;
- ${Boost_LIBRARY_DIRS}/libboost_date_time.a;${Boost_LIBRARY_DIRS}/libboost_log.a;${Boost_LIBRARY_DIRS}/libboost_log_setup.a;
- ${Boost_LIBRARY_DIRS}/libboost_regex.a;${Boost_LIBRARY_DIRS}/libboost_serialization.a;${Boost_LIBRARY_DIRS}/libboost_filesystem.a;
- ${Boost_LIBRARY_DIRS}/libboost_locale.a;${Boost_LIBRARY_DIRS}/libboost_iostreams.a)
- include_directories(${Boost_INCLUDE_DIRS})
+ set(Boost_LIBRARIES ${Boost_LIBRARY_DIRS}/libboost_atomic.a;${Boost_LIBRARY_DIRS}/libboost_thread.a;${Boost_LIBRARY_DIRS}/libboost_system.a;
+ ${Boost_LIBRARY_DIRS}/libboost_chrono.a;${Boost_LIBRARY_DIRS}/libboost_date_time.a;${Boost_LIBRARY_DIRS}/libboost_log.a;
+ ${Boost_LIBRARY_DIRS}/libboost_log_setup.a;${Boost_LIBRARY_DIRS}/libboost_regex.a;${Boost_LIBRARY_DIRS}/libboost_serialization.a;
+ ${Boost_LIBRARY_DIRS}/libboost_filesystem.a;${Boost_LIBRARY_DIRS}/libboost_locale.a;${Boost_LIBRARY_DIRS}/libboost_iostreams.a)
+ include_directories(${Boost_INCLUDE_DIR})
endif ()
message(STATUS "** Boost_INCLUDE_DIR: ${Boost_INCLUDE_DIR}")
@@ -76,18 +94,19 @@
find_package(Libevent 2.0.22 REQUIRED COMPONENTS)
if (LIBEVENT_FOUND)
include_directories(${LIBEVENT_INCLUDE_DIRS})
- message(STATUS "** libevent Include dir: ${LIBEVENT_INCLUDE_DIR}")
+ message(STATUS "** libevent Include dir: ${LIBEVENT_INCLUDE_DIRS}")
message(STATUS "** libevent Libraries: ${LIBEVENT_LIBRARIES}")
endif ()
else ()
#find_package(Libevent 2.0.22 REQUIRED COMPONENTS)
set(LIBEVENT_INCLUDE_DIRS ${PROJECT_SOURCE_DIR}/bin/include)
set(LIBEVENT_LIBRARIES_DIR ${PROJECT_SOURCE_DIR}/bin/lib)
- set(LIBEVENT_LIBRARIES ${LIBEVENT_LIBRARIES_DIR}/libevent.a;${LIBEVENT_LIBRARIES_DIR}/libevent_core.a;${LIBEVENT_LIBRARIES_DIR}/libevent_extra.a;${LIBEVENT_LIBRARIES_DIR}/libevent_pthreads.a)
+ set(LIBEVENT_LIBRARIES ${LIBEVENT_LIBRARIES_DIR}/libevent.a;${LIBEVENT_LIBRARIES_DIR}/libevent_core.a;${LIBEVENT_LIBRARIES_DIR}/libevent_extra.a;
+ ${LIBEVENT_LIBRARIES_DIR}/libevent_pthreads.a;${LIBEVENT_LIBRARIES_DIR}/libevent_openssl.a)
include_directories(${LIBEVENT_INCLUDE_DIRS})
endif ()
-message(STATUS "** LIBEVENT_INCLUDE_DIR: ${LIBEVENT_INCLUDE_DIR}")
+message(STATUS "** LIBEVENT_INCLUDE_DIR: ${LIBEVENT_INCLUDE_DIRS}")
message(STATUS "** LIBEVENT_LIBRARIES: ${LIBEVENT_LIBRARIES}")
option(JSONCPP_USE_STATIC_LIBS "only find jsoncpp static libs" ON) # only find static libs
@@ -187,8 +206,10 @@
message(STATUS "** ENABLE_LSAN: ${ENABLE_LSAN} Enable lsan reporting")
endif ()
- set(CMAKE_CXX_FLAGS_DEBUG "-O0 -DDEBUG")
+ set(CMAKE_CXX_FLAGS_DEBUG "-g -O0 -DDEBUG")
set(CMAKE_CXX_FLAGS_RELEASE "-O3 -DNDEBUG")
+ set(CMAKE_C_FLAGS_DEBUG "-g -O0 -DDEBUG")
+ set(CMAKE_C_FLAGS_RELEASE "-O3 -DNDEBUG")
# Declare deplibs, so we can use list in linker later. There's probably
diff --git a/build.sh b/build.sh
index 06530a7..4a76f9e 100755
--- a/build.sh
+++ b/build.sh
@@ -24,9 +24,12 @@
declare packet_dir="${basepath}/tmp_packet_dir"
declare install_lib_dir="${basepath}/bin"
declare static_package_dir="${basepath}/tmp_static_package_dir"
+declare fname_openssl="openssl*.tar.gz"
+
declare fname_libevent="libevent*.zip"
declare fname_jsoncpp="jsoncpp*.zip"
declare fname_boost="boost*.tar.gz"
+declare fname_openssl_down="openssl-1.1.1d.tar.gz"
declare fname_libevent_down="release-2.1.11-stable.zip"
declare fname_jsoncpp_down="0.10.7.zip"
declare fname_boost_down="1.58.0/boost_1_58_0.tar.gz"
@@ -44,24 +47,30 @@
elif test "$(uname)" = "Darwin" ; then
declare cpu_num=$(sysctl -n machdep.cpu.thread_count)
fi
-declare need_build_jsoncpp=1
+
+declare need_build_openssl=1
declare need_build_libevent=1
+declare need_build_jsoncpp=1
declare need_build_boost=1
declare enable_asan=0
declare enable_lsan=0
declare verbose=1
declare codecov=0
+declare debug=0
declare test=0
pasres_arguments() {
for var in "$@"; do
case "$var" in
- noJson)
- need_build_jsoncpp=0
+ noOpenSSL)
+ need_build_openssl=0
;;
noEvent)
need_build_libevent=0
;;
+ noJson)
+ need_build_jsoncpp=0
+ ;;
noBoost)
need_build_boost=0
;;
@@ -77,6 +86,9 @@
codecov)
codecov=1
;;
+ debug)
+ debug=1
+ ;;
test)
test=1
;;
@@ -88,16 +100,21 @@
PrintParams() {
echo "###########################################################################"
- if [ $need_build_libevent -eq 0 ]; then
- echo "no need build libevent lib"
+ if [ $need_build_openssl -eq 0 ]; then
+ echo "no need build openssl lib"
else
- echo "need build libevent lib"
+ echo "need build openssl lib"
fi
if [ $need_build_jsoncpp -eq 0 ]; then
echo "no need build jsoncpp lib"
else
echo "need build jsoncpp lib"
fi
+ if [ $need_build_libevent -eq 0 ]; then
+ echo "no need build libevent lib"
+ else
+ echo "need build libevent lib"
+ fi
if [ $need_build_boost -eq 0 ]; then
echo "no need build boost lib"
else
@@ -113,19 +130,26 @@
else
echo "disable lsan reporting"
fi
- if [ $test -eq 1 ]; then
- echo "build unit tests"
- else
- echo "without build unit tests"
- fi
- if [ $codecov -eq 1 ]; then
- echo "run unit tests with code coverage"
- fi
if [ $verbose -eq 0 ]; then
echo "no need print detail logs"
else
echo "need print detail logs"
fi
+ if [ $codecov -eq 1 ]; then
+ echo "run unit tests with code coverage"
+ else
+ echo "run unit tests without code coverage"
+ fi
+ if [ $debug -eq 1 ]; then
+ echo "enable debug"
+ else
+ echo "disable debug"
+ fi
+ if [ $test -eq 1 ]; then
+ echo "build unit tests"
+ else
+ echo "without build unit tests"
+ fi
echo "###########################################################################"
echo ""
@@ -141,6 +165,10 @@
fi
cd ${basepath}
+ if [ -e ${fname_openssl} ]; then
+ mv -f ${basepath}/${fname_openssl} ${down_dir}
+ fi
+
if [ -e ${fname_libevent} ]; then
mv -f ${basepath}/${fname_libevent} ${down_dir}
fi
@@ -174,6 +202,52 @@
fi
}
+BuildOpenSSL() {
+ if [ $need_build_openssl -eq 0 ]; then
+ echo "no need build openssl lib"
+ return 0
+ fi
+
+ cd ${down_dir}
+ if [ -e ${fname_openssl} ]; then
+ echo "${fname_openssl} is exist"
+ else
+ wget https://www.openssl.org/source/${fname_openssl_down} -O ${fname_openssl_down}
+ fi
+ tar -zxvf ${fname_openssl} &> unzipopenssl.txt
+ if [ $? -ne 0 ]; then
+ exit 1
+ fi
+
+ openssl_dir=$(ls | grep ^openssl | grep .*[^gz]$)
+ cd ${openssl_dir}
+ if [ $? -ne 0 ]; then
+ exit 1
+ fi
+ echo "build openssl static #####################"
+ if [ $verbose -eq 0 ]; then
+ ./config shared CFLAGS=-fPIC CPPFLAGS=-fPIC --prefix=${install_lib_dir} --openssldir=${install_lib_dir} &> opensslconfig.txt
+ else
+ ./config shared CFLAGS=-fPIC CPPFLAGS=-fPIC --prefix=${install_lib_dir} --openssldir=${install_lib_dir}
+ fi
+ if [ $? -ne 0 ]; then
+ exit 1
+ fi
+ if [ $verbose -eq 0 ]; then
+ echo "build openssl without detail log."
+ make depend &> opensslbuild.txt
+ make -j $cpu_num &> opensslbuild.txt
+ else
+ make depend
+ make -j $cpu_num
+ fi
+ if [ $? -ne 0 ]; then
+ exit 1
+ fi
+ make install
+ echo "build openssl success."
+}
+
BuildLibevent() {
if [ $need_build_libevent -eq 0 ]; then
echo "no need build libevent lib"
@@ -202,9 +276,9 @@
fi
echo "build libevent static #####################"
if [ $verbose -eq 0 ]; then
- ./configure --disable-openssl --enable-static=yes --enable-shared=no CFLAGS=-fPIC CPPFLAGS=-fPIC --prefix=${install_lib_dir} &> libeventconfig.txt
+ ./configure --enable-static=yes --enable-shared=no CFLAGS="-fPIC -I${install_lib_dir}/include" CPPFLAGS="-fPIC -I${install_lib_dir}/include" LDFLAGS="-L${install_lib_dir}/lib" --prefix=${install_lib_dir} &> libeventconfig.txt
else
- ./configure --disable-openssl --enable-static=yes --enable-shared=no CFLAGS=-fPIC CPPFLAGS=-fPIC --prefix=${install_lib_dir}
+ ./configure --enable-static=yes --enable-shared=no CFLAGS="-fPIC -I${install_lib_dir}/include" CPPFLAGS="-fPIC -I${install_lib_dir}/include" LDFLAGS="-L${install_lib_dir}/lib" --prefix=${install_lib_dir}
fi
if [ $? -ne 0 ]; then
exit 1
@@ -330,6 +404,11 @@
else
ROCKETMQ_CMAKE_FLAG=$ROCKETMQ_CMAKE_FLAG" -DENABLE_LSAN=OFF"
fi
+ if [ $debug -eq 1 ]; then
+ ROCKETMQ_CMAKE_FLAG=$ROCKETMQ_CMAKE_FLAG" -DCMAKE_BUILD_TYPE=Debug"
+ else
+ ROCKETMQ_CMAKE_FLAG=$ROCKETMQ_CMAKE_FLAG" -DCMAKE_BUILD_TYPE=Release"
+ fi
cmake .. $ROCKETMQ_CMAKE_FLAG
if [ $verbose -eq 0 ]; then
echo "build rocketmq without detail log."
@@ -426,22 +505,16 @@
cp -f ${install_lib_dir}/librocketmq.a .
echo "Md5 Hash RocketMQ Before:"
md5sum librocketmq.a
- local dir=`ls *.a | grep -v gtest | grep -v gmock `
+ local dir=`ls *.a | grep -E 'gtest|gmock'`
for i in $dir
do
- echo $i
- ar x $i
+ rm -rf $i
done
- echo "At last, ar libboost_filesystem"
- ar x libboost_filesystem.a
- ar cru librocketmq.a *.o
- ranlib librocketmq.a
+ libtool -no_warning_for_no_symbols -static -o librocketmq.a *.a
echo "Md5 Hash RocketMQ After:"
md5sum librocketmq.a
echo "Try to copy $(pwd)/librocketmq.a to ${install_lib_dir}/"
cp -f librocketmq.a ${install_lib_dir}/
- rm -rf *.o
- rm -rf __.*
cd ${basepath}
rm -rf ${static_package_dir}
fi
@@ -450,6 +523,7 @@
PrintParams
Prepare
+BuildOpenSSL
BuildLibevent
BuildJsonCPP
BuildBoost
diff --git a/example/CMakeLists.txt b/example/CMakeLists.txt
index aecec5c..daf9c3b 100755
--- a/example/CMakeLists.txt
+++ b/example/CMakeLists.txt
@@ -23,6 +23,7 @@
link_directories(${Boost_LIBRARY_DIRS})
link_directories(${LIBEVENT_LIBRARY})
link_directories(${JSONCPP_LIBRARY})
+link_directories(${OPENSSL_LIBRARIES_DIR})
#if (BUILD_ROCKETMQ_SHARED)
# set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DBOOST_ALL_DYN_LINK -shared ")
@@ -43,10 +44,10 @@
if (MSVC)
if (BUILD_ROCKETMQ_SHARED)
target_link_libraries (${basename} rocketmq_shared ${deplibs}
- ${Boost_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES})
+ ${Boost_LIBRARIES} ${OPENSSL_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES})
else()
target_link_libraries (${basename} rocketmq_static ${deplibs}
- ${Boost_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES})
+ ${Boost_LIBRARIES} ${OPENSSL_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES})
endif()
else()
if (BUILD_ROCKETMQ_SHARED)
diff --git a/example/common.h b/example/common.h
index 7dbd3d9..58d4b4e 100644
--- a/example/common.h
+++ b/example/common.h
@@ -17,9 +17,9 @@
#ifndef ROCKETMQ_CLIENT4CPP_EXAMPLE_COMMON_H_
#define ROCKETMQ_CLIENT4CPP_EXAMPLE_COMMON_H_
-#include <functional>
#include <atomic>
#include <chrono>
+#include <functional>
#include <iostream>
#include <memory>
#include <string>
diff --git a/include/DefaultMQProducer.h b/include/DefaultMQProducer.h
index 6ca52f9..9eab3a8 100644
--- a/include/DefaultMQProducer.h
+++ b/include/DefaultMQProducer.h
@@ -139,6 +139,12 @@
void setMessageTrace(bool messageTrace);
bool getMessageTrace() const;
+ void setEnableSsl(bool enableSsl);
+ bool getEnableSsl() const;
+
+ void setSslPropertyFile(const std::string& sslPropertyFile);
+ const std::string& getSslPropertyFile() const;
+
private:
DefaultMQProducerImpl* impl;
};
diff --git a/include/DefaultMQPullConsumer.h b/include/DefaultMQPullConsumer.h
index 6ed6c62..0883bca 100644
--- a/include/DefaultMQPullConsumer.h
+++ b/include/DefaultMQPullConsumer.h
@@ -63,6 +63,12 @@
const std::string& getGroupName() const;
void setGroupName(const std::string& groupname);
+ void setEnableSsl(bool enableSsl);
+ bool getEnableSsl() const;
+
+ void setSslPropertyFile(const std::string& sslPropertyFile);
+ const std::string& getSslPropertyFile() const;
+
/**
* Log configuration interface, default LOG_LEVEL is LOG_LEVEL_INFO, default
* log file num is 3, each log size is 100M
diff --git a/include/DefaultMQPushConsumer.h b/include/DefaultMQPushConsumer.h
index 46f76cf..5d3bc35 100644
--- a/include/DefaultMQPushConsumer.h
+++ b/include/DefaultMQPushConsumer.h
@@ -69,6 +69,12 @@
const std::string& getGroupName() const;
void setGroupName(const std::string& groupname);
+ void setEnableSsl(bool enableSsl);
+ bool getEnableSsl() const;
+
+ void setSslPropertyFile(const std::string& sslPropertyFile);
+ const std::string& getSslPropertyFile() const;
+
/**
* Log configuration interface, default LOG_LEVEL is LOG_LEVEL_INFO, default
* log file num is 3, each log size is 100M
diff --git a/package_rocketmq.mri b/package_rocketmq.mri
index fc5f690..f19383e 100644
--- a/package_rocketmq.mri
+++ b/package_rocketmq.mri
@@ -11,10 +11,13 @@
addlib ../bin/lib/libboost_system.a
addlib ../bin/lib/libboost_thread.a
addlib ../bin/lib/libboost_wserialization.a
+addlib ../bin/lib/libssl.a
+addlib ../bin/lib/libcrypto.a
addlib ../bin/lib/libevent.a
addlib ../bin/lib/libevent_core.a
addlib ../bin/lib/libevent_extra.a
addlib ../bin/lib/libevent_pthreads.a
+addlib ../bin/lib/libevent_openssl.a
addlib ../bin/lib/libjsoncpp.a
addlib ../bin/lib/libSignature.a
addlib ../bin/librocketmq.a
diff --git a/project/CMakeLists.txt b/project/CMakeLists.txt
index 60b5d0d..efd2a81 100755
--- a/project/CMakeLists.txt
+++ b/project/CMakeLists.txt
@@ -50,6 +50,7 @@
target_link_libraries(rocketmq_static ${JSONCPP_LIBRARIES})
target_link_libraries(rocketmq_static ${LIBEVENT_LIBRARIES})
target_link_libraries(rocketmq_static ${Boost_LIBRARIES})
+ target_link_libraries(rocketmq_static ${OPENSSL_LIBRARIES})
target_link_libraries(rocketmq_static ${deplibs})
endif ()
@@ -63,6 +64,7 @@
target_link_libraries(rocketmq_shared ${JSONCPP_LIBRARIES})
target_link_libraries(rocketmq_shared ${LIBEVENT_LIBRARIES})
target_link_libraries(rocketmq_shared ${Boost_LIBRARIES})
+ target_link_libraries(rocketmq_shared ${OPENSSL_LIBRARIES})
target_link_libraries(rocketmq_shared ${deplibs})
endif ()
diff --git a/src/MQClientAPIImpl.cpp b/src/MQClientAPIImpl.cpp
index 3b85d54..b76f6c3 100644
--- a/src/MQClientAPIImpl.cpp
+++ b/src/MQClientAPIImpl.cpp
@@ -27,15 +27,19 @@
namespace rocketmq {
//<!************************************************************************
-MQClientAPIImpl::MQClientAPIImpl(const string& mqClientId) : m_firstFetchNameSrv(true), m_mqClientId(mqClientId) {}
+MQClientAPIImpl::MQClientAPIImpl(const string& mqClientId, bool enableSsl, const std::string& sslPropertyFile)
+ : m_firstFetchNameSrv(true), m_mqClientId(mqClientId) {}
MQClientAPIImpl::MQClientAPIImpl(const string& mqClientId,
ClientRemotingProcessor* clientRemotingProcessor,
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
- string unitName)
+ string unitName,
+ bool enableSsl,
+ const std::string& sslPropertyFile)
: m_firstFetchNameSrv(true), m_mqClientId(mqClientId) {
- m_pRemotingClient.reset(new TcpRemotingClient(pullThreadNum, tcpConnectTimeout, tcpTransportTryLockTimeout));
+ m_pRemotingClient.reset(
+ new TcpRemotingClient(pullThreadNum, tcpConnectTimeout, tcpTransportTryLockTimeout, enableSsl, sslPropertyFile));
m_pRemotingClient->registerProcessor(CHECK_TRANSACTION_STATE, clientRemotingProcessor);
m_pRemotingClient->registerProcessor(RESET_CONSUMER_CLIENT_OFFSET, clientRemotingProcessor);
m_pRemotingClient->registerProcessor(GET_CONSUMER_STATUS_FROM_CLIENT, clientRemotingProcessor);
diff --git a/src/MQClientAPIImpl.h b/src/MQClientAPIImpl.h
index 0e47df0..b8bd883 100644
--- a/src/MQClientAPIImpl.h
+++ b/src/MQClientAPIImpl.h
@@ -42,13 +42,15 @@
//<!************************************************************************
class MQClientAPIImpl {
public:
- MQClientAPIImpl(const string& mqClientId);
+ MQClientAPIImpl(const string& mqClientId, bool enableSsl, const std::string& sslPropertyFile);
MQClientAPIImpl(const string& mqClientId,
ClientRemotingProcessor* clientRemotingProcessor,
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
- string unitName);
+ string unitName,
+ bool enableSsl,
+ const std::string& sslPropertyFile);
virtual ~MQClientAPIImpl();
virtual void stopAllTcpTransportThread();
virtual bool writeDataToFile(string filename, string data, bool isSync);
diff --git a/src/MQClientFactory.cpp b/src/MQClientFactory.cpp
index 7600d12..69d2543 100644
--- a/src/MQClientFactory.cpp
+++ b/src/MQClientFactory.cpp
@@ -30,14 +30,17 @@
namespace rocketmq {
//<!***************************************************************************
-MQClientFactory::MQClientFactory(const string& clientID) : m_bFetchNSService(true) {
+MQClientFactory::MQClientFactory(const string& clientID, bool enableSsl, const std::string& sslPropertyFile)
+ : m_bFetchNSService(true) {
m_clientId = clientID;
}
MQClientFactory::MQClientFactory(const string& clientID,
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
- string unitName)
+ string unitName,
+ bool enableSsl,
+ const std::string& sslPropertyFile)
: m_bFetchNSService(true) {
m_clientId = clientID;
// default Topic register;
@@ -45,7 +48,8 @@
m_topicPublishInfoTable[DEFAULT_TOPIC] = pDefaultTopicInfo;
m_pClientRemotingProcessor.reset(new ClientRemotingProcessor(this));
m_pClientAPIImpl.reset(new MQClientAPIImpl(m_clientId, m_pClientRemotingProcessor.get(), pullThreadNum,
- tcpConnectTimeout, tcpTransportTryLockTimeout, unitName));
+ tcpConnectTimeout, tcpTransportTryLockTimeout, unitName, enableSsl,
+ sslPropertyFile));
m_serviceState = CREATE_JUST;
LOG_DEBUG("MQClientFactory construct");
}
diff --git a/src/MQClientFactory.h b/src/MQClientFactory.h
index 51f554a..e071db2 100644
--- a/src/MQClientFactory.h
+++ b/src/MQClientFactory.h
@@ -45,8 +45,10 @@
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
- string unitName);
- MQClientFactory(const string& clientID);
+ string unitName,
+ bool enableSsl,
+ const std::string& sslPropertyFile);
+ MQClientFactory(const string& clientID, bool enableSsl, const std::string& sslPropertyFile);
virtual ~MQClientFactory();
virtual void start();
diff --git a/src/MQClientManager.cpp b/src/MQClientManager.cpp
index de35ddd..011ade2 100644
--- a/src/MQClientManager.cpp
+++ b/src/MQClientManager.cpp
@@ -34,13 +34,15 @@
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
- string unitName) {
+ string unitName,
+ bool enableSsl,
+ const std::string& sslPropertyFile) {
FTMAP::iterator it = m_factoryTable.find(clientId);
if (it != m_factoryTable.end()) {
return it->second;
} else {
- MQClientFactory* factory =
- new MQClientFactory(clientId, pullThreadNum, tcpConnectTimeout, tcpTransportTryLockTimeout, unitName);
+ MQClientFactory* factory = new MQClientFactory(clientId, pullThreadNum, tcpConnectTimeout,
+ tcpTransportTryLockTimeout, unitName, enableSsl, sslPropertyFile);
m_factoryTable[clientId] = factory;
return factory;
}
diff --git a/src/MQClientManager.h b/src/MQClientManager.h
index d846f96..d0ddd49 100644
--- a/src/MQClientManager.h
+++ b/src/MQClientManager.h
@@ -31,7 +31,9 @@
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
- string unitName);
+ string unitName,
+ bool enableSsl,
+ const std::string& sslPropertyFile);
void removeClientFactory(const string& clientId);
static MQClientManager* getInstance();
diff --git a/src/common/DefaultMQClient.cpp b/src/common/DefaultMQClient.cpp
index e381218..93ec906 100644
--- a/src/common/DefaultMQClient.cpp
+++ b/src/common/DefaultMQClient.cpp
@@ -147,7 +147,8 @@
void DefaultMQClient::start() {
if (getFactory() == NULL) {
m_clientFactory = MQClientManager::getInstance()->getMQClientFactory(
- getMQClientId(), m_pullThreadNum, m_tcpConnectTimeout, m_tcpTransportTryLockTimeout, m_unitName);
+ getMQClientId(), m_pullThreadNum, m_tcpConnectTimeout, m_tcpTransportTryLockTimeout, m_unitName, m_enableSsl,
+ m_sslPropertyFile);
}
LOG_INFO(
"MQClient "
@@ -236,6 +237,23 @@
const SessionCredentials& DefaultMQClient::getSessionCredentials() const {
return m_SessionCredentials;
}
+
+void DefaultMQClient::setEnableSsl(bool enableSsl) {
+ m_enableSsl = enableSsl;
+}
+
+bool DefaultMQClient::getEnableSsl() const {
+ return m_enableSsl;
+}
+
+void DefaultMQClient::setSslPropertyFile(const std::string& sslPropertyFile) {
+ m_sslPropertyFile = sslPropertyFile;
+}
+
+const std::string& DefaultMQClient::getSslPropertyFile() const {
+ return m_sslPropertyFile;
+}
+
void DefaultMQClient::showClientConfigs() {
// LOG_WARN("*****************************************************************************");
LOG_WARN("ClientID:%s", getMQClientId().c_str());
@@ -248,6 +266,8 @@
LOG_WARN("PullThreadNum:%d", m_pullThreadNum);
LOG_WARN("TcpConnectTimeout:%lld ms", m_tcpConnectTimeout);
LOG_WARN("TcpTransportTryLockTimeout:%lld s", m_tcpTransportTryLockTimeout);
+ LOG_WARN("EnableSsl:%s", m_enableSsl ? "true" : "false");
+ LOG_WARN("SslPropertyFile:%s", m_sslPropertyFile.c_str());
LOG_WARN("OpenMessageTrace:%s", m_messageTrace ? "true" : "false");
// LOG_WARN("*****************************************************************************");
}
diff --git a/src/common/UtilAll.cpp b/src/common/UtilAll.cpp
index 7122499..8faa8b1 100644
--- a/src/common/UtilAll.cpp
+++ b/src/common/UtilAll.cpp
@@ -353,4 +353,34 @@
return false;
#endif
}
+
+std::map<std::string, std::string> UtilAll::ReadProperties(const std::string& path) {
+ std::map<std::string, std::string> property_map;
+ std::ifstream property_file;
+ property_file.open(path);
+ std::string line_buffer;
+
+ if (property_file.is_open()) {
+ while (!property_file.eof()) {
+ std::getline(property_file, line_buffer);
+ std::size_t pos{0};
+ pos = line_buffer.find('#');
+ if (pos != string::npos) {
+ line_buffer = line_buffer.substr(0, pos);
+ }
+ if (line_buffer.empty()) {
+ continue;
+ }
+ pos = line_buffer.find('=');
+ if (pos != string::npos) {
+ std::string key = boost::trim_copy(line_buffer.substr(0, pos));
+ std::string value = boost::trim_copy(line_buffer.substr(pos + 1));
+ property_map[key] = value;
+ }
+ }
+ }
+
+ return property_map;
+}
+
} // namespace rocketmq
diff --git a/src/common/UtilAll.h b/src/common/UtilAll.h
index 6d021f8..4f34419 100644
--- a/src/common/UtilAll.h
+++ b/src/common/UtilAll.h
@@ -28,6 +28,7 @@
#include <sys/time.h>
#include <sys/types.h>
#endif
+#include <boost/algorithm/string/trim.hpp>
#include <boost/asio.hpp>
#include <boost/iostreams/copy.hpp>
#include <boost/iostreams/device/back_inserter.hpp>
@@ -37,14 +38,16 @@
#include <boost/lexical_cast.hpp>
#include <boost/locale/conversion.hpp>
#include <boost/locale/encoding.hpp>
+#include <fstream>
+#include <map>
#include <sstream>
#include "RocketMQClient.h"
using namespace std;
namespace rocketmq {
//<!************************************************************************
+const string null = "";
const string WHITESPACE = " \t\r\n";
-const int MASTER_ID = 0;
const string SUB_ALL = "*";
const string DEFAULT_TOPIC = "TBW102";
const string BENCHMARK_TOPIC = "BenchmarkTest";
@@ -58,22 +61,25 @@
const string ROCKETMQ_HOME_ENV = "ROCKETMQ_HOME";
const string ROCKETMQ_HOME_PROPERTY = "rocketmq.home.dir";
const string MESSAGE_COMPRESS_LEVEL = "rocketmq.message.compressLevel";
-const int POLL_NAMESERVER_INTEVAL = 1000 * 30;
-const int HEARTBEAT_BROKER_INTERVAL = 1000 * 30;
-const int PERSIST_CONSUMER_OFFSET_INTERVAL = 1000 * 5;
+const string DEFAULT_SSL_PROPERTY_FILE = "/etc/rocketmq/tls.properties";
+const string DEFAULT_CLIENT_KEY_PASSWD = null;
+const string DEFAULT_CLIENT_KEY_FILE = "/etc/rocketmq/client.key";
+const string DEFAULT_CLIENT_CERT_FILE = "/etc/rocketmq/client.pem";
+const string DEFAULT_CA_CERT_FILE = "/etc/rocketmq/ca.pem";
const string WS_ADDR =
"please set nameserver domain by setDomainName, there is no default "
"nameserver domain";
-
-const int LINE_SEPARATOR = 1; // rocketmq::UtilAll::charToString((char) 1);
-const int WORD_SEPARATOR = 2; // rocketmq::UtilAll::charToString((char) 2);
-
+const int POLL_NAMESERVER_INTEVAL = 1000 * 30;
+const int HEARTBEAT_BROKER_INTERVAL = 1000 * 30;
+const int PERSIST_CONSUMER_OFFSET_INTERVAL = 1000 * 5;
+const int LINE_SEPARATOR = 1; // rocketmq::UtilAll::charToString((char) 1);
+const int WORD_SEPARATOR = 2; // rocketmq::UtilAll::charToString((char) 2);
const int HTTP_TIMEOUT = 3000; // 3S
const int HTTP_CONFLICT = 409;
const int HTTP_OK = 200;
const int HTTP_NOTFOUND = 404;
const int CONNETERROR = -1;
-const string null = "";
+const int MASTER_ID = 0;
template <typename Type>
inline void deleteAndZero(Type& pointer) {
@@ -134,6 +140,8 @@
// Returns false on failure..
static bool ReplaceFile(const std::string& from_path, const std::string& to_path);
+ static std::map<std::string, std::string> ReadProperties(const std::string& path);
+
private:
static std::string s_localHostName;
static std::string s_localIpAddress;
diff --git a/src/consumer/DefaultMQPullConsumer.cpp b/src/consumer/DefaultMQPullConsumer.cpp
index 363c5cc..a5ccfc4 100644
--- a/src/consumer/DefaultMQPullConsumer.cpp
+++ b/src/consumer/DefaultMQPullConsumer.cpp
@@ -91,6 +91,22 @@
impl->setGroupName(groupName);
}
+void DefaultMQPullConsumer::setEnableSsl(bool enableSsl) {
+ impl->setEnableSsl(enableSsl);
+}
+
+bool DefaultMQPullConsumer::getEnableSsl() const {
+ return impl->getEnableSsl();
+}
+
+void DefaultMQPullConsumer::setSslPropertyFile(const std::string& sslPropertyFile) {
+ impl->setSslPropertyFile(sslPropertyFile);
+}
+
+const std::string& DefaultMQPullConsumer::getSslPropertyFile() const {
+ return impl->getSslPropertyFile();
+}
+
void DefaultMQPullConsumer::setLogLevel(elogLevel inputLevel) {
impl->setLogLevel(inputLevel);
}
diff --git a/src/consumer/DefaultMQPushConsumer.cpp b/src/consumer/DefaultMQPushConsumer.cpp
index e0cb5bf..6fe3d1f 100644
--- a/src/consumer/DefaultMQPushConsumer.cpp
+++ b/src/consumer/DefaultMQPushConsumer.cpp
@@ -163,6 +163,22 @@
impl->setGroupName(groupName);
}
+void DefaultMQPushConsumer::setEnableSsl(bool enableSsl) {
+ impl->setEnableSsl(enableSsl);
+}
+
+bool DefaultMQPushConsumer::getEnableSsl() const {
+ return impl->getEnableSsl();
+}
+
+void DefaultMQPushConsumer::setSslPropertyFile(const std::string& sslPropertyFile) {
+ impl->setSslPropertyFile(sslPropertyFile);
+}
+
+const std::string& DefaultMQPushConsumer::getSslPropertyFile() const {
+ return impl->getSslPropertyFile();
+}
+
void DefaultMQPushConsumer::setLogLevel(elogLevel inputLevel) {
impl->setLogLevel(inputLevel);
}
diff --git a/src/include/DefaultMQClient.h b/src/include/DefaultMQClient.h
index 32b6aed..92ce318 100644
--- a/src/include/DefaultMQClient.h
+++ b/src/include/DefaultMQClient.h
@@ -29,6 +29,7 @@
#include "QueryResult.h"
#include "RocketMQClient.h"
#include "SessionCredentials.h"
+#include "UtilAll.h"
namespace rocketmq {
class MQClientFactory;
@@ -170,8 +171,12 @@
virtual void setFactory(MQClientFactory*);
- bool getMessageTrace() const;
+ void setEnableSsl(bool enableSsl);
+ bool getEnableSsl() const;
+ void setSslPropertyFile(const std::string& sslPropertyFile);
+ const std::string& getSslPropertyFile() const;
+ bool getMessageTrace() const;
void setMessageTrace(bool mMessageTrace);
protected:
@@ -187,6 +192,8 @@
std::string m_instanceName;
std::string m_nameSpace;
std::string m_GroupName;
+ std::string m_sslPropertyFile{DEFAULT_SSL_PROPERTY_FILE};
+ bool m_enableSsl{false};
MQClientFactory* m_clientFactory;
int m_serviceState;
int m_pullThreadNum;
diff --git a/src/producer/DefaultMQProducer.cpp b/src/producer/DefaultMQProducer.cpp
index 34c8ac1..5b77030 100644
--- a/src/producer/DefaultMQProducer.cpp
+++ b/src/producer/DefaultMQProducer.cpp
@@ -238,4 +238,20 @@
impl->sendOneway(msg, selector, arg);
}
+void DefaultMQProducer::setEnableSsl(bool enableSsl) {
+ impl->setEnableSsl(enableSsl);
+}
+
+bool DefaultMQProducer::getEnableSsl() const {
+ return impl->getEnableSsl();
+}
+
+void DefaultMQProducer::setSslPropertyFile(const std::string& sslPropertyFile) {
+ impl->setSslPropertyFile(sslPropertyFile);
+}
+
+const std::string& DefaultMQProducer::getSslPropertyFile() const {
+ return impl->getSslPropertyFile();
+}
+
} // namespace rocketmq
\ No newline at end of file
diff --git a/src/transport/ClientRemotingProcessor.cpp b/src/transport/ClientRemotingProcessor.cpp
index 63736c5..2c67a0c 100644
--- a/src/transport/ClientRemotingProcessor.cpp
+++ b/src/transport/ClientRemotingProcessor.cpp
@@ -1,19 +1,19 @@
/*
-* 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.
-*/
+ * 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 "ClientRemotingProcessor.h"
#include "ClientRPCHook.h"
#include "ConsumerRunningInfo.h"
diff --git a/src/transport/ClientRemotingProcessor.h b/src/transport/ClientRemotingProcessor.h
index c88b8bb..2b26f28 100644
--- a/src/transport/ClientRemotingProcessor.h
+++ b/src/transport/ClientRemotingProcessor.h
@@ -1,19 +1,19 @@
/*
-* 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.
-*/
+ * 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 __CLIENTREMOTINGPROCESSOR_H__
#define __CLIENTREMOTINGPROCESSOR_H__
@@ -62,6 +62,6 @@
private:
std::map<MQMessageQueue, int64> m_offsetTable;
};
-}
+} // namespace rocketmq
#endif
diff --git a/src/transport/EventLoop.cpp b/src/transport/EventLoop.cpp
index 3d67405..e66ceb5 100644
--- a/src/transport/EventLoop.cpp
+++ b/src/transport/EventLoop.cpp
@@ -22,8 +22,9 @@
#include <event2/thread.h>
+#include <boost/filesystem.hpp>
+
#include "Logging.h"
-#include "UtilAll.h"
namespace rocketmq {
@@ -32,9 +33,7 @@
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
+EventLoop::EventLoop(const struct event_config* config, bool run_immediately) {
#ifdef WIN32
evthread_use_windows_threads();
#else
@@ -85,7 +84,7 @@
void EventLoop::stop() {
if (m_loopThread != nullptr /*&& m_loopThread.joinable()*/) {
- _is_running = false;
+ m_isRuning = false;
m_loopThread->join();
delete m_loopThread;
@@ -94,9 +93,9 @@
}
void EventLoop::runLoop() {
- _is_running = true;
+ m_isRuning = true;
- while (_is_running) {
+ while (m_isRuning) {
int ret;
ret = event_base_dispatch(m_eventBase);
@@ -109,11 +108,133 @@
}
}
+bool EventLoop::CreateSslContext(const std::string& ssl_property_file) {
+ ERR_load_crypto_strings();
+ SSL_load_error_strings();
+ SSL_library_init();
+ OpenSSL_add_all_algorithms();
+
+ m_sslCtx.reset(SSL_CTX_new(SSLv23_client_method()));
+ if (!m_sslCtx) {
+ LOG_ERROR("Failed to create ssl context!");
+ return false;
+ }
+
+ std::string client_key_file = DEFAULT_CLIENT_KEY_FILE;
+ std::string client_key_passwd = DEFAULT_CLIENT_KEY_PASSWD;
+ std::string client_cert_file = DEFAULT_CLIENT_CERT_FILE;
+ std::string ca_cert_file = DEFAULT_CA_CERT_FILE;
+ auto properties = UtilAll::ReadProperties(ssl_property_file);
+ if (!properties.empty()) {
+ if (properties.find("tls.client.keyPath") != properties.end()) {
+ client_key_file = properties["tls.client.keyPath"];
+ }
+ if (properties.find("tls.client.keyPassword") != properties.end()) {
+ client_key_passwd = properties["tls.client.keyPassword"];
+ }
+ if (properties.find("tls.client.certPath") != properties.end()) {
+ client_cert_file = properties["tls.client.certPath"];
+ }
+ if (properties.find("tls.client.trustCertPath") != properties.end()) {
+ ca_cert_file = properties["tls.client.trustCertPath"];
+ }
+ } else {
+ LOG_WARN(
+ "The tls properties file is not specified or empty. "
+ "Set it by modifying the api of setTlsPropertyFile and fill the configuration content.");
+ }
+
+ SSL_CTX_set_verify(m_sslCtx.get(), SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);
+ SSL_CTX_set_mode(m_sslCtx.get(), SSL_MODE_AUTO_RETRY);
+
+ if (client_key_passwd.empty()) {
+ LOG_WARN(
+ "The pass phrase is not specified. "
+ "Set it by adding the 'tls.client.keyPassword' property in configuration file.");
+ } else {
+ SSL_CTX_set_default_passwd_cb_userdata(m_sslCtx.get(), (void*)client_key_passwd.c_str());
+ }
+
+ bool check_flag{true};
+ if (!boost::filesystem::exists(ca_cert_file.c_str())) {
+ check_flag = false;
+ LOG_WARN(
+ "'%s' does not exist. Please make sure the 'tls.client.trustCertPath' property "
+ "in the configuration file is configured correctly.",
+ ca_cert_file.c_str());
+ } else if (SSL_CTX_load_verify_locations(m_sslCtx.get(), ca_cert_file.c_str(), NULL) <= 0) {
+ LOG_ERROR("SSL_CTX_load_verify_locations error!");
+ ERR_print_errors_fp(stderr);
+ return false;
+ }
+
+ if (!boost::filesystem::exists(client_cert_file.c_str())) {
+ check_flag = false;
+ LOG_WARN(
+ "'%s' does not exist. Please make sure the 'tls.client.certPath' property "
+ "in the configuration file is configured correctly.",
+ client_cert_file.c_str());
+ } else if (SSL_CTX_use_certificate_file(m_sslCtx.get(), client_cert_file.c_str(), SSL_FILETYPE_PEM) <= 0) {
+ LOG_ERROR("SSL_CTX_use_certificate_file error!");
+ ERR_print_errors_fp(stderr);
+ return false;
+ }
+
+ if (!boost::filesystem::exists(client_key_file.c_str())) {
+ check_flag = false;
+ LOG_WARN(
+ "'%s' does not exist. Please make sure the 'tls.client.keyPath' property "
+ "in the configuration file is configured correctly.",
+ client_key_file.c_str());
+ } else if (SSL_CTX_use_PrivateKey_file(m_sslCtx.get(), client_key_file.c_str(), SSL_FILETYPE_PEM) <= 0) {
+ LOG_ERROR("SSL_CTX_use_PrivateKey_file error!");
+ ERR_print_errors_fp(stderr);
+ return false;
+ }
+
+ if (check_flag && SSL_CTX_check_private_key(m_sslCtx.get()) <= 0) {
+ LOG_ERROR("SSL_CTX_check_private_key error!");
+ ERR_print_errors_fp(stderr);
+ return false;
+ }
+
+ return true;
+}
+
#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);
+BufferEvent* EventLoop::createBufferEvent(socket_t fd,
+ int options,
+ bool enable_ssl,
+ const std::string& ssl_property_file) {
+ struct bufferevent* event{nullptr};
+
+ if (enable_ssl) {
+ if (!m_sslCtx && !CreateSslContext(ssl_property_file)) {
+ LOG_ERROR("Failed to create ssl context!");
+ return nullptr;
+ }
+
+ SSL* ssl = SSL_new(m_sslCtx.get());
+ if (ssl == nullptr) {
+ LOG_ERROR("Failed to create ssl handle!");
+ return nullptr;
+ }
+
+ // create ssl bufferevent
+ event = bufferevent_openssl_socket_new(m_eventBase, fd, ssl, BUFFEREVENT_SSL_CONNECTING, options);
+
+ /* create filter ssl bufferevent
+ struct bufferevent *bev = bufferevent_socket_new(m_eventBase, fd, options);
+ event = bufferevent_openssl_filter_new(m_eventBase, bev, ssl,
+ BUFFEREVENT_SSL_CONNECTING, options);
+ */
+ } else {
+ event = bufferevent_socket_new(m_eventBase, fd, options);
+ }
+
if (event == nullptr) {
+ LOG_ERROR("Failed to create bufferevent!");
return nullptr;
}
diff --git a/src/transport/EventLoop.h b/src/transport/EventLoop.h
index c974479..24eb9ba 100644
--- a/src/transport/EventLoop.h
+++ b/src/transport/EventLoop.h
@@ -22,8 +22,12 @@
#include <event2/buffer.h>
#include <event2/bufferevent.h>
+#include <event2/bufferevent_ssl.h>
#include <event2/event.h>
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+#include "UtilAll.h"
#include "noncopyable.h"
using socket_t = evutil_socket_t;
@@ -43,16 +47,18 @@
void start();
void stop();
- BufferEvent* createBufferEvent(socket_t fd, int options);
+ BufferEvent* createBufferEvent(socket_t fd, int options, bool enable_ssl, const std::string& ssl_property_file);
private:
void runLoop();
+ bool CreateSslContext(const std::string& ssl_property_file);
private:
- struct event_base* m_eventBase;
- std::thread* m_loopThread;
-
- bool _is_running; // aotmic is unnecessary
+ struct event_base* m_eventBase{nullptr};
+ std::thread* m_loopThread{nullptr};
+ using SSL_CTX_ptr = std::unique_ptr<SSL_CTX, decltype(::SSL_CTX_free)&>;
+ SSL_CTX_ptr m_sslCtx{nullptr, ::SSL_CTX_free};
+ bool m_isRuning{false}; // aotmic is unnecessary
};
class TcpTransport;
diff --git a/src/transport/SocketUtil.cpp b/src/transport/SocketUtil.cpp
index daa369c..3e8f269 100644
--- a/src/transport/SocketUtil.cpp
+++ b/src/transport/SocketUtil.cpp
@@ -86,4 +86,4 @@
uint64 n2hll(uint64 v) {
return swapll(v);
}
-} //<!end namespace;
+} // namespace rocketmq
diff --git a/src/transport/SocketUtil.h b/src/transport/SocketUtil.h
index 536b486..e6156a6 100644
--- a/src/transport/SocketUtil.h
+++ b/src/transport/SocketUtil.h
@@ -19,8 +19,8 @@
#ifdef WIN32
#include <WS2tcpip.h>
-#include <Winsock2.h>
#include <Windows.h>
+#include <Winsock2.h>
#pragma comment(lib, "ws2_32.lib")
#else
#include <arpa/inet.h>
@@ -34,7 +34,6 @@
#include <sys/ioctl.h>
#include <sys/select.h>
#include <sys/socket.h>
-#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
@@ -45,8 +44,8 @@
namespace rocketmq {
//<!************************************************************************
/**
-* IP:PORT
-*/
+ * IP:PORT
+ */
sockaddr IPPort2socketAddress(int host, int port);
string socketAddress2IPPort(sockaddr addr);
void socketAddress2IPPort(sockaddr addr, int& host, int& port);
@@ -58,6 +57,6 @@
uint64 n2hll(uint64 v);
//<!************************************************************************
-} //<!end namespace;
+} // namespace rocketmq
#endif
diff --git a/src/transport/TcpRemotingClient.cpp b/src/transport/TcpRemotingClient.cpp
index 294c319..fa2bfc8 100644
--- a/src/transport/TcpRemotingClient.cpp
+++ b/src/transport/TcpRemotingClient.cpp
@@ -28,13 +28,22 @@
namespace rocketmq {
//<!************************************************************************
-TcpRemotingClient::TcpRemotingClient()
- : m_dispatchServiceWork(m_dispatchService), m_handleServiceWork(m_handleService) {}
-TcpRemotingClient::TcpRemotingClient(int pullThreadNum, uint64_t tcpConnectTimeout, uint64_t tcpTransportTryLockTimeout)
+TcpRemotingClient::TcpRemotingClient(bool enableSsl, const std::string& sslPropertyFile)
+ : m_enableSsl(enableSsl),
+ m_sslPropertyFile(sslPropertyFile),
+ m_dispatchServiceWork(m_dispatchService),
+ m_handleServiceWork(m_handleService) {}
+TcpRemotingClient::TcpRemotingClient(int pullThreadNum,
+ uint64_t tcpConnectTimeout,
+ uint64_t tcpTransportTryLockTimeout,
+ bool enableSsl,
+ const std::string& sslPropertyFile)
: m_dispatchThreadNum(1),
m_pullThreadNum(pullThreadNum),
m_tcpConnectTimeout(tcpConnectTimeout),
m_tcpTransportTryLockTimeout(tcpTransportTryLockTimeout),
+ m_enableSsl(enableSsl),
+ m_sslPropertyFile(sslPropertyFile),
m_namesrvIndex(0),
m_dispatchServiceWork(m_dispatchService),
m_handleServiceWork(m_handleService) {
@@ -326,7 +335,7 @@
//<!callback;
TcpTransportReadCallback callback = needResponse ? &TcpRemotingClient::static_messageReceived : nullptr;
- tts = TcpTransport::CreateTransport(this, callback);
+ tts = TcpTransport::CreateTransport(this, m_enableSsl, m_sslPropertyFile, callback);
TcpConnectStatus connectStatus = tts->connect(addr, 0); // use non-block
if (connectStatus != TCP_CONNECT_STATUS_WAIT) {
LOG_WARN("can not connect to:%s", addr.c_str());
diff --git a/src/transport/TcpRemotingClient.h b/src/transport/TcpRemotingClient.h
index 856026a..4ec3d87 100644
--- a/src/transport/TcpRemotingClient.h
+++ b/src/transport/TcpRemotingClient.h
@@ -36,8 +36,12 @@
class TcpRemotingClient {
public:
- TcpRemotingClient();
- TcpRemotingClient(int pullThreadNum, uint64_t tcpConnectTimeout, uint64_t tcpTransportTryLockTimeout);
+ TcpRemotingClient(bool enableSsl, const string& sslPropertyFile);
+ TcpRemotingClient(int pullThreadNum,
+ uint64_t tcpConnectTimeout,
+ uint64_t tcpTransportTryLockTimeout,
+ bool enableSsl,
+ const string& sslPropertyFile);
virtual ~TcpRemotingClient();
virtual void stopAllTcpTransportThread();
@@ -109,6 +113,9 @@
uint64_t m_tcpConnectTimeout; // ms
uint64_t m_tcpTransportTryLockTimeout; // s
+ bool m_enableSsl;
+ std::string m_sslPropertyFile;
+
//<! NameServer
std::timed_mutex m_namesrvLock;
vector<string> m_namesrvAddrList;
diff --git a/src/transport/TcpTransport.cpp b/src/transport/TcpTransport.cpp
index 641c516..503acb0 100644
--- a/src/transport/TcpTransport.cpp
+++ b/src/transport/TcpTransport.cpp
@@ -31,13 +31,18 @@
namespace rocketmq {
//<!************************************************************************
-TcpTransport::TcpTransport(TcpRemotingClient* pTcpRemointClient, TcpTransportReadCallback handle)
+TcpTransport::TcpTransport(TcpRemotingClient* pTcpRemointClient,
+ bool enableSsl,
+ const std::string& sslPropertyFile,
+ TcpTransportReadCallback handle)
: m_event(nullptr),
m_tcpConnectStatus(TCP_CONNECT_STATUS_INIT),
m_connectEventLock(),
m_connectEvent(),
m_readCallback(handle),
- m_tcpRemotingClient(pTcpRemointClient) {
+ m_tcpRemotingClient(pTcpRemointClient),
+ m_enableSsl(enableSsl),
+ m_sslPropertyFile(sslPropertyFile) {
m_startTime = UtilAll::currentTimeMillis();
}
@@ -155,7 +160,8 @@
sin.sin_addr.s_addr = getInetAddr(hostname);
sin.sin_port = htons(port);
- m_event.reset(EventLoop::GetDefaultEventLoop()->createBufferEvent(-1, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE));
+ m_event.reset(EventLoop::GetDefaultEventLoop()->createBufferEvent(-1, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE,
+ m_enableSsl, m_sslPropertyFile));
m_event->setCallback(readNextMessageIntCallback, nullptr, eventCallback, shared_from_this());
m_event->setWatermark(EV_READ, 4, 0);
m_event->enable(EV_READ | EV_WRITE);
diff --git a/src/transport/TcpTransport.h b/src/transport/TcpTransport.h
old mode 100755
new mode 100644
index bff23dd..79b14d7
--- a/src/transport/TcpTransport.h
+++ b/src/transport/TcpTransport.h
@@ -41,9 +41,11 @@
class TcpTransport : public std::enable_shared_from_this<TcpTransport> {
public:
static std::shared_ptr<TcpTransport> CreateTransport(TcpRemotingClient* pTcpRemotingClient,
+ bool enableSsl,
+ const std::string& sslPropertyFile,
TcpTransportReadCallback handle = nullptr) {
// transport must be managed by smart pointer
- std::shared_ptr<TcpTransport> transport(new TcpTransport(pTcpRemotingClient, handle));
+ std::shared_ptr<TcpTransport> transport(new TcpTransport(pTcpRemotingClient, enableSsl, sslPropertyFile, handle));
return transport;
}
@@ -59,7 +61,10 @@
const uint64_t getStartTime() const;
private:
- TcpTransport(TcpRemotingClient* pTcpRemotingClient, TcpTransportReadCallback handle = nullptr);
+ TcpTransport(TcpRemotingClient* pTcpRemotingClient,
+ bool enableSsl,
+ const std::string& sslPropertyFile,
+ TcpTransportReadCallback handle = nullptr);
static void readNextMessageIntCallback(BufferEvent* event, TcpTransport* transport);
static void eventCallback(BufferEvent* event, short what, TcpTransport* transport);
@@ -85,6 +90,9 @@
//<! read data callback
TcpTransportReadCallback m_readCallback;
TcpRemotingClient* m_tcpRemotingClient;
+
+ bool m_enableSsl;
+ std::string m_sslPropertyFile;
};
//<!************************************************************************
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 68085c7..ce3845d 100755
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -45,14 +45,13 @@
message(STATUS "** Gmock_LIBRARIES: ${Gmock_LIBRARIES}")
link_directories(${Boost_LIBRARY_DIRS})
+link_directories(${OPENSSL_LIBRARIES_DIR})
link_directories(${LIBEVENT_LIBRARY})
link_directories(${JSONCPP_LIBRARY})
set(ROCKETMQ_LIBRARIES ${CMAKE_SOURCE_DIR}/bin/librocketmq.a)
message(STATUS "** ROCKETMQ_LIBRARIES ${ROCKETMQ_LIBRARIES}")
-set(CMAKE_BUILD_TYPE "Debug")
-
function(compile files)
foreach (file ${files})
get_filename_component(basename ${file} NAME_WE)
@@ -69,10 +68,10 @@
if (MSVC)
if (BUILD_ROCKETMQ_SHARED)
target_link_libraries(${basename} rocketmq_shared ${deplibs}
- ${Boost_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES} ${x`})
+ ${Boost_LIBRARIES} ${OPENSSL_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES} ${x`})
else ()
target_link_libraries(${basename} rocketmq_static ${deplibs}
- ${Boost_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES} ${Gtest_LIBRARIES})
+ ${Boost_LIBRARIES} ${OPENSSL_LIBRARIES} ${LIBEVENT_LIBRARIES} ${JSONCPP_LIBRARIES} ${Gtest_LIBRARIES})
endif ()
else ()
target_link_libraries(${basename} rocketmq_shared ${deplibs})
diff --git a/test/src/MQClientAPIImpTest.cpp b/test/src/MQClientAPIImpTest.cpp
index b334af3..d736870 100644
--- a/test/src/MQClientAPIImpTest.cpp
+++ b/test/src/MQClientAPIImpTest.cpp
@@ -35,7 +35,7 @@
class MockTcpRemotingClient : public TcpRemotingClient {
public:
- MockTcpRemotingClient() : TcpRemotingClient() {}
+ MockTcpRemotingClient() : TcpRemotingClient(true, DEFAULT_SSL_PROPERTY_FILE) {}
MOCK_METHOD3(invokeSync, RemotingCommand*(const string&, RemotingCommand&, int));
MOCK_METHOD6(invokeAsync, bool(const string&, RemotingCommand&, std::shared_ptr<AsyncCallbackWrap>, int64, int, int));
@@ -48,7 +48,7 @@
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
string unitName)
- : MQClientAPIImpl(mqClientId) {}
+ : MQClientAPIImpl(mqClientId, true, DEFAULT_SSL_PROPERTY_FILE) {}
void reInitRemoteClient(TcpRemotingClient* client) { m_pRemotingClient.reset(client); }
};
diff --git a/test/src/MQClientFactoryTest.cpp b/test/src/MQClientFactoryTest.cpp
index eb78cf7..cab4f32 100644
--- a/test/src/MQClientFactoryTest.cpp
+++ b/test/src/MQClientFactoryTest.cpp
@@ -50,7 +50,7 @@
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
string unitName)
- : MQClientAPIImpl(mqClientId) {}
+ : MQClientAPIImpl(mqClientId, true, DEFAULT_SSL_PROPERTY_FILE) {}
MOCK_METHOD5(getMinOffset, int64(const string&, const string&, int, int, const SessionCredentials&));
MOCK_METHOD3(getTopicRouteInfoFromNameServer, TopicRouteData*(const string&, int, const SessionCredentials&));
@@ -62,7 +62,7 @@
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
string unitName)
- : MQClientFactory(mqClientId) {}
+ : MQClientFactory(mqClientId, true, DEFAULT_SSL_PROPERTY_FILE) {}
void reInitClientImpl(MQClientAPIImpl* pImpl) { m_pClientAPIImpl.reset(pImpl); }
void addTestConsumer(const string& consumerName, MQConsumer* pMQConsumer) {
addConsumerToTable(consumerName, pMQConsumer);
diff --git a/test/src/MQClientManagerTest.cpp b/test/src/MQClientManagerTest.cpp
index 78b336b..208770e 100644
--- a/test/src/MQClientManagerTest.cpp
+++ b/test/src/MQClientManagerTest.cpp
@@ -33,8 +33,10 @@
TEST(MQClientManagerTest, getClientFactory) {
string clientId = "testClientId";
string unitName = "central";
- MQClientFactory* factory = MQClientManager::getInstance()->getMQClientFactory(clientId, 1, 1000, 3000, unitName);
- MQClientFactory* factory2 = MQClientManager::getInstance()->getMQClientFactory(clientId, 1, 1000, 3000, unitName);
+ MQClientFactory* factory = MQClientManager::getInstance()->getMQClientFactory(clientId, 1, 1000, 3000, unitName, true,
+ DEFAULT_SSL_PROPERTY_FILE);
+ MQClientFactory* factory2 = MQClientManager::getInstance()->getMQClientFactory(clientId, 1, 1000, 3000, unitName,
+ true, DEFAULT_SSL_PROPERTY_FILE);
EXPECT_EQ(factory, factory2);
factory->shutdown();
diff --git a/test/src/producer/DefaultMQProducerImplTest.cpp b/test/src/producer/DefaultMQProducerImplTest.cpp
index 3e42e69..a86041e 100644
--- a/test/src/producer/DefaultMQProducerImplTest.cpp
+++ b/test/src/producer/DefaultMQProducerImplTest.cpp
@@ -48,7 +48,7 @@
};
class MockMQClientFactory : public MQClientFactory {
public:
- MockMQClientFactory(const string& mqClientId) : MQClientFactory(mqClientId) {}
+ MockMQClientFactory(const string& mqClientId) : MQClientFactory(mqClientId, true, DEFAULT_SSL_PROPERTY_FILE) {}
MOCK_METHOD0(start, void());
MOCK_METHOD0(shutdown, void());
MOCK_METHOD0(sendHeartbeatToAllBroker, void());
@@ -61,7 +61,7 @@
};
class MockMQClientAPIImpl : public MQClientAPIImpl {
public:
- MockMQClientAPIImpl() : MQClientAPIImpl("testMockMQClientAPIImpl") {}
+ MockMQClientAPIImpl() : MQClientAPIImpl("testMockMQClientAPIImpl", true, DEFAULT_SSL_PROPERTY_FILE) {}
MOCK_METHOD9(sendMessage,
SendResult(const string&,
diff --git a/test/src/transport/ClientRemotingProcessorTest.cpp b/test/src/transport/ClientRemotingProcessorTest.cpp
index 3664d89..716150d 100644
--- a/test/src/transport/ClientRemotingProcessorTest.cpp
+++ b/test/src/transport/ClientRemotingProcessorTest.cpp
@@ -80,7 +80,13 @@
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
string unitName)
- : MQClientFactory(clientID, pullThreadNum, tcpConnectTimeout, tcpTransportTryLockTimeout, unitName) {}
+ : MQClientFactory(clientID,
+ pullThreadNum,
+ tcpConnectTimeout,
+ tcpTransportTryLockTimeout,
+ unitName,
+ true,
+ rocketmq::DEFAULT_SSL_PROPERTY_FILE) {}
MOCK_METHOD3(resetOffset,
void(const string& group, const string& topic, const map<MQMessageQueue, int64>& offsetTable));
