diff --git a/environment/os/centos-7.sh b/environment/os/centos-7.sh index b5fa816c1..b8ffede4a 100755 --- a/environment/os/centos-7.sh +++ b/environment/os/centos-7.sh @@ -49,6 +49,7 @@ MEMGRAPH_BUILD_DEPS=( doxygen graphviz # source documentation generators which mono-complete dotnet-sdk-3.1 golang nodejs zip unzip java-11-openjdk-devel # for driver tests autoconf # for jemalloc code generation + libtool # for protobuf code generation ) list() { diff --git a/environment/os/centos-8.sh b/environment/os/centos-8.sh index d8fcfe561..994fc0e88 100755 --- a/environment/os/centos-8.sh +++ b/environment/os/centos-8.sh @@ -48,6 +48,7 @@ MEMGRAPH_BUILD_DEPS=( which mono-complete dotnet-sdk-3.1 nodejs golang zip unzip java-11-openjdk-devel # for driver tests sbcl # for custom Lisp C++ preprocessing autoconf # for jemalloc code generation + libtool # for protobuf code generation ) list() { diff --git a/environment/os/debian-10.sh b/environment/os/debian-10.sh index 1580dbfa4..50ca5e8cd 100755 --- a/environment/os/debian-10.sh +++ b/environment/os/debian-10.sh @@ -46,6 +46,7 @@ MEMGRAPH_BUILD_DEPS=( mono-runtime mono-mcs zip unzip default-jdk-headless # for driver tests dotnet-sdk-3.1 golang nodejs npm autoconf # for jemalloc code generation + libtool # for protobuf code generation ) list() { diff --git a/environment/os/debian-11.sh b/environment/os/debian-11.sh index bc79c1f5f..8e73688e5 100755 --- a/environment/os/debian-11.sh +++ b/environment/os/debian-11.sh @@ -47,6 +47,7 @@ MEMGRAPH_BUILD_DEPS=( mono-runtime mono-mcs zip unzip default-jdk-headless # for driver tests dotnet-sdk-3.1 golang nodejs npm autoconf # for jemalloc code generation + libtool # for protobuf code generation ) list() { diff --git a/environment/os/debian-9.sh b/environment/os/debian-9.sh index 4bde3db0f..1f94adcee 100755 --- a/environment/os/debian-9.sh +++ b/environment/os/debian-9.sh @@ -44,6 +44,7 @@ MEMGRAPH_BUILD_DEPS=( doxygen graphviz # source documentation generators mono-runtime mono-mcs nodejs zip unzip default-jdk-headless # for driver tests autoconf # for jemalloc code generation + libtool # for protobuf code generation ) list() { diff --git a/environment/os/ubuntu-18.04.sh b/environment/os/ubuntu-18.04.sh index 03f094ec2..6856039d1 100755 --- a/environment/os/ubuntu-18.04.sh +++ b/environment/os/ubuntu-18.04.sh @@ -45,6 +45,7 @@ MEMGRAPH_BUILD_DEPS=( doxygen graphviz # source documentation generators mono-runtime mono-mcs nodejs zip unzip default-jdk-headless # driver tests autoconf # for jemalloc code generation + libtool # for protobuf code generation ) list() { diff --git a/environment/os/ubuntu-20.04.sh b/environment/os/ubuntu-20.04.sh index 21a7049e1..6b9e643cf 100755 --- a/environment/os/ubuntu-20.04.sh +++ b/environment/os/ubuntu-20.04.sh @@ -46,6 +46,7 @@ MEMGRAPH_BUILD_DEPS=( mono-runtime mono-mcs zip unzip default-jdk-headless # for driver tests dotnet-sdk-3.1 golang nodejs npm autoconf # for jemalloc code generation + libtool # for protobuf code generation ) list() { diff --git a/libs/.gitignore b/libs/.gitignore index 08bd2de9f..0be557228 100644 --- a/libs/.gitignore +++ b/libs/.gitignore @@ -5,3 +5,4 @@ !CMakeLists.txt !__main.cpp !jemalloc.cmake +!pulsar.patch diff --git a/libs/CMakeLists.txt b/libs/CMakeLists.txt index eab36adf4..1cd2dcd72 100644 --- a/libs/CMakeLists.txt +++ b/libs/CMakeLists.txt @@ -48,7 +48,7 @@ endfunction(import_library) # INSTALL_COMMAND arguments. function(add_external_project name) set(options NO_C_COMPILER) - set(one_value_kwargs SOURCE_DIR) + set(one_value_kwargs SOURCE_DIR BUILD_IN_SOURCE) set(multi_value_kwargs CMAKE_ARGS DEPENDS INSTALL_COMMAND BUILD_COMMAND CONFIGURE_COMMAND) cmake_parse_arguments(KW "${options}" "${one_value_kwargs}" "${multi_value_kwargs}" ${ARGN}) @@ -56,11 +56,16 @@ function(add_external_project name) if (KW_SOURCE_DIR) set(source_dir ${KW_SOURCE_DIR}) endif() + set(build_in_source 0) + if (KW_BUILD_IN_SOURCE) + set(build_in_source ${KW_BUILD_IN_SOURCE}) + endif() if (NOT KW_NO_C_COMPILER) set(KW_CMAKE_ARGS -DCMAKE_C_COMPILER=${CMAKE_C_COMPILER} ${KW_CMAKE_ARGS}) endif() ExternalProject_Add(${name}-proj DEPENDS ${KW_DEPENDS} PREFIX ${source_dir} SOURCE_DIR ${source_dir} + BUILD_IN_SOURCE ${build_in_source} CONFIGURE_COMMAND ${KW_CONFIGURE_COMMAND} CMAKE_ARGS -DCMAKE_BUILD_TYPE=Release -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER} @@ -169,9 +174,12 @@ import_external_library(bzip2 STATIC INSTALL_COMMAND true) # Setup zlib +set(ZLIB_ROOT ${CMAKE_CURRENT_SOURCE_DIR}/zlib) +set(ZLIB_LIBRARIES ${ZLIB_ROOT}/lib/libz.a) +set(ZLIB_INCLUDE_DIRS ${ZLIB_ROOT}/include) import_external_library(zlib STATIC - ${CMAKE_CURRENT_SOURCE_DIR}/zlib/lib/libz.a - ${CMAKE_CURRENT_SOURCE_DIR}/zlib + ${ZLIB_LIBRARIES} + ${ZLIB_INCLUDE_DIRS} CMAKE_ARGS -DCMAKE_SKIP_INSTALL_ALL_DEPENDENCY=true BUILD_COMMAND $(MAKE) zlibstatic) @@ -231,6 +239,7 @@ import_external_library(librdkafka STATIC CMAKE_ARGS -DRDKAFKA_BUILD_STATIC=ON -DRDKAFKA_BUILD_EXAMPLES=OFF -DRDKAFKA_BUILD_TESTS=OFF + -DENABLE_LZ4_EXT=OFF -DCMAKE_INSTALL_LIBDIR=lib -DWITH_SSL=ON # If we want SASL, we need to install it on build machines @@ -242,3 +251,36 @@ import_library(librdkafka++ STATIC ${CMAKE_CURRENT_SOURCE_DIR}/librdkafka/include ) target_link_libraries(librdkafka++ INTERFACE librdkafka) + +set(PROTOBUF_ROOT ${CMAKE_CURRENT_SOURCE_DIR}/protobuf/lib) +import_external_library(protobuf STATIC + ${PROTOBUF_ROOT}/lib/libprotobuf.a + ${PROTOBUF_ROOT}/include + BUILD_IN_SOURCE 1 + CONFIGURE_COMMAND true) + +set(BOOST_ROOT ${CMAKE_CURRENT_SOURCE_DIR}/boost/lib) + +import_external_library(pulsar STATIC + ${CMAKE_CURRENT_SOURCE_DIR}/pulsar/pulsar-client-cpp/lib/libpulsarwithdeps.a + ${CMAKE_CURRENT_SOURCE_DIR}/pulsar/install/include + BUILD_IN_SOURCE 1 + CONFIGURE_COMMAND cmake pulsar-client-cpp + -DCMAKE_INSTALL_PREFIX=${CMAKE_CURRENT_SOURCE_DIR}/pulsar/install + -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER} + -DCMAKE_C_COMPILER=${CMAKE_C_COMPILER} + -DBUILD_DYNAMIC_LIB=OFF + -DBUILD_STATIC_LIB=ON + -DBUILD_TESTS=OFF + -DLINK_STATIC=ON + -DPROTOC_PATH=${PROTOBUF_ROOT}/bin/protoc + -DBOOST_ROOT=${BOOST_ROOT} + -DCMAKE_PREFIX_PATH=${PROTOBUF_ROOT} + -DProtobuf_INCLUDE_DIRS=${PROTOBUF_ROOT}/include + -DZLIB_LIBRARIES=${ZLIB_LIBRARIES} + -DZLIB_INCLUDE_DIRS=${ZLIB_INCLUDE_DIRS} + -DBUILD_PYTHON_WRAPPER=OFF + -DBUILD_PERF_TOOLS=OFF + -DUSE_LOG4CXX=OFF + BUILD_COMMAND $(MAKE) pulsarStaticWithDeps) +add_dependencies(pulsar-proj protobuf zlib) diff --git a/libs/pulsar.patch b/libs/pulsar.patch new file mode 100644 index 000000000..870e54375 --- /dev/null +++ b/libs/pulsar.patch @@ -0,0 +1,1508 @@ +diff --git a/pulsar-client-cpp/CMakeLists.txt b/pulsar-client-cpp/CMakeLists.txt +index 5cbbea6..17f82dc 100644 +--- a/pulsar-client-cpp/CMakeLists.txt ++++ b/pulsar-client-cpp/CMakeLists.txt +@@ -112,19 +112,11 @@ find_package(OpenSSL REQUIRED) + set(RECORD_OPENSSL_SSL_LIBRARY ${OPENSSL_SSL_LIBRARY}) + set(RECORD_OPENSSL_CRYPTO_LIBRARY ${OPENSSL_CRYPTO_LIBRARY}) + +-unset(OPENSSL_FOUND CACHE) +-unset(OPENSSL_INCLUDE_DIR CACHE) +-unset(OPENSSL_CRYPTO_LIBRARY CACHE) +-unset(OPENSSL_CRYPTO_LIBRARIES CACHE) +-unset(OPENSSL_SSL_LIBRARY CACHE) +-unset(OPENSSL_SSL_LIBRARIES CACHE) +-unset(OPENSSL_LIBRARIES CACHE) +-unset(OPENSSL_VERSION CACHE) +- + if (LINK_STATIC) + find_library(ZLIB_LIBRARIES REQUIRED NAMES libz.a z zlib) + find_library(Protobuf_LITE_LIBRARIES NAMES libprotobuf-lite.a libprotobuf-lite) +- find_library(CURL_LIBRARIES NAMES libcurl.a curl curl_a libcurl_a) ++ find_package(CURL REQUIRED) ++ set(COMMON_LIBS ${COMMON_LIBS} CURL::libcurl) + find_library(LIB_ZSTD NAMES libzstd.a) + find_library(LIB_SNAPPY NAMES libsnappy.a) + message(STATUS "Protobuf_LITE_LIBRARIES: ${Protobuf_LITE_LIBRARIES}") +@@ -157,7 +149,6 @@ if (LINK_STATIC) + endif() + + SET(Boost_USE_STATIC_LIBS ON) +- SET(OPENSSL_USE_STATIC_LIBS TRUE) + else() + # Link to shared libraries + find_package(ZLIB REQUIRED) +diff --git a/pulsar-client-cpp/lib/CompressionCodecLZ4.cc b/pulsar-client-cpp/lib/CompressionCodecLZ4.cc +index 508e4f4..87c5f2a 100644 +--- a/pulsar-client-cpp/lib/CompressionCodecLZ4.cc ++++ b/pulsar-client-cpp/lib/CompressionCodecLZ4.cc +@@ -25,11 +25,11 @@ namespace pulsar { + + SharedBuffer CompressionCodecLZ4::encode(const SharedBuffer& raw) { + // Get the max size of the compressed data and allocate a buffer to hold it +- int maxCompressedSize = LZ4_compressBound(raw.readableBytes()); ++ int maxCompressedSize = PULSAR_LZ4_compressBound(raw.readableBytes()); + SharedBuffer compressed = SharedBuffer::allocate(maxCompressedSize); + + int compressedSize = +- LZ4_compress_default(raw.data(), compressed.mutableData(), raw.readableBytes(), maxCompressedSize); ++ PULSAR_LZ4_compress_default(raw.data(), compressed.mutableData(), raw.readableBytes(), maxCompressedSize); + assert(compressedSize > 0); + compressed.bytesWritten(compressedSize); + +@@ -40,7 +40,7 @@ bool CompressionCodecLZ4::decode(const SharedBuffer& encoded, uint32_t uncompres + SharedBuffer& decoded) { + SharedBuffer decompressed = SharedBuffer::allocate(uncompressedSize); + +- int result = LZ4_decompress_fast(encoded.data(), decompressed.mutableData(), uncompressedSize); ++ int result = PULSAR_LZ4_decompress_fast(encoded.data(), decompressed.mutableData(), uncompressedSize); + if (result > 0) { + decompressed.bytesWritten(uncompressedSize); + decoded = decompressed; +diff --git a/pulsar-client-cpp/lib/lz4/lz4.c b/pulsar-client-cpp/lib/lz4/lz4.c +index 08cf6b5..d74c287 100644 +--- a/pulsar-client-cpp/lib/lz4/lz4.c ++++ b/pulsar-client-cpp/lib/lz4/lz4.c +@@ -45,7 +45,7 @@ + + /* + * ACCELERATION_DEFAULT : +- * Select "acceleration" for LZ4_compress_fast() when parameter value <= 0 ++ * Select "acceleration" for PULSAR_LZ4_compress_fast() when parameter value <= 0 + */ + #define ACCELERATION_DEFAULT 1 + +@@ -135,25 +135,25 @@ + + static unsigned LZ4_64bits(void) { return sizeof(void*)==8; } + +-static unsigned LZ4_isLittleEndian(void) ++static unsigned PULSAR_LZ4_isLittleEndian(void) + { + const union { U32 i; BYTE c[4]; } one = { 1 }; /* don't use static : performance detrimental */ + return one.c[0]; + } + + +-static U16 LZ4_read16(const void* memPtr) ++static U16 PULSAR_LZ4_read16(const void* memPtr) + { + U16 val16; + memcpy(&val16, memPtr, 2); + return val16; + } + +-static U16 LZ4_readLE16(const void* memPtr) ++static U16 PULSAR_LZ4_readLE16(const void* memPtr) + { +- if (LZ4_isLittleEndian()) ++ if (PULSAR_LZ4_isLittleEndian()) + { +- return LZ4_read16(memPtr); ++ return PULSAR_LZ4_read16(memPtr); + } + else + { +@@ -162,9 +162,9 @@ static U16 LZ4_readLE16(const void* memPtr) + } + } + +-static void LZ4_writeLE16(void* memPtr, U16 value) ++static void PULSAR_LZ4_writeLE16(void* memPtr, U16 value) + { +- if (LZ4_isLittleEndian()) ++ if (PULSAR_LZ4_isLittleEndian()) + { + memcpy(memPtr, &value, 2); + } +@@ -176,40 +176,40 @@ static void LZ4_writeLE16(void* memPtr, U16 value) + } + } + +-static U32 LZ4_read32(const void* memPtr) ++static U32 PULSAR_LZ4_read32(const void* memPtr) + { + U32 val32; + memcpy(&val32, memPtr, 4); + return val32; + } + +-static U64 LZ4_read64(const void* memPtr) ++static U64 PULSAR_LZ4_read64(const void* memPtr) + { + U64 val64; + memcpy(&val64, memPtr, 8); + return val64; + } + +-static size_t LZ4_read_ARCH(const void* p) ++static size_t PULSAR_LZ4_read_ARCH(const void* p) + { + if (LZ4_64bits()) +- return (size_t)LZ4_read64(p); ++ return (size_t)PULSAR_LZ4_read64(p); + else +- return (size_t)LZ4_read32(p); ++ return (size_t)PULSAR_LZ4_read32(p); + } + + +-static void LZ4_copy4(void* dstPtr, const void* srcPtr) { memcpy(dstPtr, srcPtr, 4); } ++static void PULSAR_LZ4_copy4(void* dstPtr, const void* srcPtr) { memcpy(dstPtr, srcPtr, 4); } + +-static void LZ4_copy8(void* dstPtr, const void* srcPtr) { memcpy(dstPtr, srcPtr, 8); } ++static void PULSAR_LZ4_copy8(void* dstPtr, const void* srcPtr) { memcpy(dstPtr, srcPtr, 8); } + + /* customized version of memcpy, which may overwrite up to 7 bytes beyond dstEnd */ +-static void LZ4_wildCopy(void* dstPtr, const void* srcPtr, void* dstEnd) ++static void PULSAR_LZ4_wildCopy(void* dstPtr, const void* srcPtr, void* dstEnd) + { + BYTE* d = (BYTE*)dstPtr; + const BYTE* s = (const BYTE*)srcPtr; + BYTE* e = (BYTE*)dstEnd; +- do { LZ4_copy8(d,s); d+=8; s+=8; } while (d compression run slower on incompressible data */ ++static const U32 PULSAR_LZ4_skipTrigger = 6; /* Increase this value ==> compression run slower on incompressible data */ + + + /************************************** +@@ -353,7 +353,7 @@ typedef struct { + const BYTE* dictionary; + BYTE* bufferStart; /* obsolete, used for slideInputBuffer */ + U32 dictSize; +-} LZ4_stream_t_internal; ++} PULSAR_LZ4_stream_t_internal; + + typedef enum { notLimited = 0, limitedOutput = 1 } limitedOutput_directive; + typedef enum { byPtr, byU32, byU16 } tableType_t; +@@ -368,9 +368,9 @@ typedef enum { full = 0, partial = 1 } earlyEnd_directive; + /************************************** + * Local Utils + **************************************/ +-int LZ4_versionNumber (void) { return LZ4_VERSION_NUMBER; } +-int LZ4_compressBound(int isize) { return LZ4_COMPRESSBOUND(isize); } +-int LZ4_sizeofState() { return LZ4_STREAMSIZE; } ++int PULSAR_LZ4_versionNumber (void) { return LZ4_VERSION_NUMBER; } ++int PULSAR_LZ4_compressBound(int isize) { return LZ4_COMPRESSBOUND(isize); } ++int PULSAR_LZ4_sizeofState() { return LZ4_STREAMSIZE; } + + + +@@ -378,7 +378,7 @@ int LZ4_sizeofState() { return LZ4_STREAMSIZE; } + * Compression functions + ********************************/ + +-static U32 LZ4_hashSequence(U32 sequence, tableType_t const tableType) ++static U32 PULSAR_LZ4_hashSequence(U32 sequence, tableType_t const tableType) + { + if (tableType == byU16) + return (((sequence) * 2654435761U) >> ((MINMATCH*8)-(LZ4_HASHLOG+1))); +@@ -387,23 +387,23 @@ static U32 LZ4_hashSequence(U32 sequence, tableType_t const tableType) + } + + static const U64 prime5bytes = 889523592379ULL; +-static U32 LZ4_hashSequence64(size_t sequence, tableType_t const tableType) ++static U32 PULSAR_LZ4_hashSequence64(size_t sequence, tableType_t const tableType) + { + const U32 hashLog = (tableType == byU16) ? LZ4_HASHLOG+1 : LZ4_HASHLOG; + const U32 hashMask = (1<> (40 - hashLog)) & hashMask; + } + +-static U32 LZ4_hashSequenceT(size_t sequence, tableType_t const tableType) ++static U32 PULSAR_LZ4_hashSequenceT(size_t sequence, tableType_t const tableType) + { + if (LZ4_64bits()) +- return LZ4_hashSequence64(sequence, tableType); +- return LZ4_hashSequence((U32)sequence, tableType); ++ return PULSAR_LZ4_hashSequence64(sequence, tableType); ++ return PULSAR_LZ4_hashSequence((U32)sequence, tableType); + } + +-static U32 LZ4_hashPosition(const void* p, tableType_t tableType) { return LZ4_hashSequenceT(LZ4_read_ARCH(p), tableType); } ++static U32 PULSAR_LZ4_hashPosition(const void* p, tableType_t tableType) { return PULSAR_LZ4_hashSequenceT(PULSAR_LZ4_read_ARCH(p), tableType); } + +-static void LZ4_putPositionOnHash(const BYTE* p, U32 h, void* tableBase, tableType_t const tableType, const BYTE* srcBase) ++static void PULSAR_LZ4_putPositionOnHash(const BYTE* p, U32 h, void* tableBase, tableType_t const tableType, const BYTE* srcBase) + { + switch (tableType) + { +@@ -413,26 +413,26 @@ static void LZ4_putPositionOnHash(const BYTE* p, U32 h, void* tableBase, tableTy + } + } + +-static void LZ4_putPosition(const BYTE* p, void* tableBase, tableType_t tableType, const BYTE* srcBase) ++static void PULSAR_LZ4_putPosition(const BYTE* p, void* tableBase, tableType_t tableType, const BYTE* srcBase) + { +- U32 h = LZ4_hashPosition(p, tableType); +- LZ4_putPositionOnHash(p, h, tableBase, tableType, srcBase); ++ U32 h = PULSAR_LZ4_hashPosition(p, tableType); ++ PULSAR_LZ4_putPositionOnHash(p, h, tableBase, tableType, srcBase); + } + +-static const BYTE* LZ4_getPositionOnHash(U32 h, void* tableBase, tableType_t tableType, const BYTE* srcBase) ++static const BYTE* PULSAR_LZ4_getPositionOnHash(U32 h, void* tableBase, tableType_t tableType, const BYTE* srcBase) + { + if (tableType == byPtr) { const BYTE** hashTable = (const BYTE**) tableBase; return hashTable[h]; } + if (tableType == byU32) { U32* hashTable = (U32*) tableBase; return hashTable[h] + srcBase; } + { U16* hashTable = (U16*) tableBase; return hashTable[h] + srcBase; } /* default, to ensure a return */ + } + +-static const BYTE* LZ4_getPosition(const BYTE* p, void* tableBase, tableType_t tableType, const BYTE* srcBase) ++static const BYTE* PULSAR_LZ4_getPosition(const BYTE* p, void* tableBase, tableType_t tableType, const BYTE* srcBase) + { +- U32 h = LZ4_hashPosition(p, tableType); +- return LZ4_getPositionOnHash(h, tableBase, tableType, srcBase); ++ U32 h = PULSAR_LZ4_hashPosition(p, tableType); ++ return PULSAR_LZ4_getPositionOnHash(h, tableBase, tableType, srcBase); + } + +-FORCE_INLINE int LZ4_compress_generic( ++FORCE_INLINE int PULSAR_LZ4_compress_generic( + void* const ctx, + const char* const source, + char* const dest, +@@ -444,7 +444,7 @@ FORCE_INLINE int LZ4_compress_generic( + const dictIssue_directive dictIssue, + const U32 acceleration) + { +- LZ4_stream_t_internal* const dictPtr = (LZ4_stream_t_internal*)ctx; ++ PULSAR_LZ4_stream_t_internal* const dictPtr = (PULSAR_LZ4_stream_t_internal*)ctx; + + const BYTE* ip = (const BYTE*) source; + const BYTE* base; +@@ -483,11 +483,11 @@ FORCE_INLINE int LZ4_compress_generic( + break; + } + if ((tableType == byU16) && (inputSize>=LZ4_64Klimit)) return 0; /* Size too large (not within 64K limit) */ +- if (inputSize> LZ4_skipTrigger); ++ step = (searchMatchNb++ >> PULSAR_LZ4_skipTrigger); + + if (unlikely(forwardIp > mflimit)) goto _last_literals; + +- match = LZ4_getPositionOnHash(h, ctx, tableType, base); ++ match = PULSAR_LZ4_getPositionOnHash(h, ctx, tableType, base); + if (dict==usingExtDict) + { + if (match<(const BYTE*)source) +@@ -522,12 +522,12 @@ FORCE_INLINE int LZ4_compress_generic( + lowLimit = (const BYTE*)source; + } + } +- forwardH = LZ4_hashPosition(forwardIp, tableType); +- LZ4_putPositionOnHash(ip, h, ctx, tableType, base); ++ forwardH = PULSAR_LZ4_hashPosition(forwardIp, tableType); ++ PULSAR_LZ4_putPositionOnHash(ip, h, ctx, tableType, base); + + } while ( ((dictIssue==dictSmall) ? (match < lowRefLimit) : 0) + || ((tableType==byU16) ? 0 : (match + MAX_DISTANCE < ip)) +- || (LZ4_read32(match+refDelta) != LZ4_read32(ip)) ); ++ || (PULSAR_LZ4_read32(match+refDelta) != PULSAR_LZ4_read32(ip)) ); + } + + /* Catch up */ +@@ -549,13 +549,13 @@ FORCE_INLINE int LZ4_compress_generic( + else *token = (BYTE)(litLength< matchlimit) limit = matchlimit; +- matchLength = LZ4_count(ip+MINMATCH, match+MINMATCH, limit); ++ matchLength = PULSAR_LZ4_count(ip+MINMATCH, match+MINMATCH, limit); + ip += MINMATCH + matchLength; + if (ip==limit) + { +- unsigned more = LZ4_count(ip, (const BYTE*)source, matchlimit); ++ unsigned more = PULSAR_LZ4_count(ip, (const BYTE*)source, matchlimit); + matchLength += more; + ip += more; + } + } + else + { +- matchLength = LZ4_count(ip+MINMATCH, match+MINMATCH, matchlimit); ++ matchLength = PULSAR_LZ4_count(ip+MINMATCH, match+MINMATCH, matchlimit); + ip += MINMATCH + matchLength; + } + +@@ -601,10 +601,10 @@ _next_match: + if (ip > mflimit) break; + + /* Fill table */ +- LZ4_putPosition(ip-2, ctx, tableType, base); ++ PULSAR_LZ4_putPosition(ip-2, ctx, tableType, base); + + /* Test next position */ +- match = LZ4_getPosition(ip, ctx, tableType, base); ++ match = PULSAR_LZ4_getPosition(ip, ctx, tableType, base); + if (dict==usingExtDict) + { + if (match<(const BYTE*)source) +@@ -618,14 +618,14 @@ _next_match: + lowLimit = (const BYTE*)source; + } + } +- LZ4_putPosition(ip, ctx, tableType, base); ++ PULSAR_LZ4_putPosition(ip, ctx, tableType, base); + if ( ((dictIssue==dictSmall) ? (match>=lowRefLimit) : 1) + && (match+MAX_DISTANCE>=ip) +- && (LZ4_read32(match+refDelta)==LZ4_read32(ip)) ) ++ && (PULSAR_LZ4_read32(match+refDelta)==PULSAR_LZ4_read32(ip)) ) + { token=op++; *token=0; goto _next_match; } + + /* Prepare next loop */ +- forwardH = LZ4_hashPosition(++ip, tableType); ++ forwardH = PULSAR_LZ4_hashPosition(++ip, tableType); + } + + _last_literals: +@@ -654,38 +654,38 @@ _last_literals: + } + + +-int LZ4_compress_fast_extState(void* state, const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) ++int PULSAR_LZ4_compress_fast_extState(void* state, const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) + { +- LZ4_resetStream((LZ4_stream_t*)state); ++ PULSAR_LZ4_resetStream((PULSAR_LZ4_stream_t*)state); + if (acceleration < 1) acceleration = ACCELERATION_DEFAULT; + +- if (maxOutputSize >= LZ4_compressBound(inputSize)) ++ if (maxOutputSize >= PULSAR_LZ4_compressBound(inputSize)) + { + if (inputSize < LZ4_64Klimit) +- return LZ4_compress_generic(state, source, dest, inputSize, 0, notLimited, byU16, noDict, noDictIssue, acceleration); ++ return PULSAR_LZ4_compress_generic(state, source, dest, inputSize, 0, notLimited, byU16, noDict, noDictIssue, acceleration); + else +- return LZ4_compress_generic(state, source, dest, inputSize, 0, notLimited, LZ4_64bits() ? byU32 : byPtr, noDict, noDictIssue, acceleration); ++ return PULSAR_LZ4_compress_generic(state, source, dest, inputSize, 0, notLimited, LZ4_64bits() ? byU32 : byPtr, noDict, noDictIssue, acceleration); + } + else + { + if (inputSize < LZ4_64Klimit) +- return LZ4_compress_generic(state, source, dest, inputSize, maxOutputSize, limitedOutput, byU16, noDict, noDictIssue, acceleration); ++ return PULSAR_LZ4_compress_generic(state, source, dest, inputSize, maxOutputSize, limitedOutput, byU16, noDict, noDictIssue, acceleration); + else +- return LZ4_compress_generic(state, source, dest, inputSize, maxOutputSize, limitedOutput, LZ4_64bits() ? byU32 : byPtr, noDict, noDictIssue, acceleration); ++ return PULSAR_LZ4_compress_generic(state, source, dest, inputSize, maxOutputSize, limitedOutput, LZ4_64bits() ? byU32 : byPtr, noDict, noDictIssue, acceleration); + } + } + + +-int LZ4_compress_fast(const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) ++int PULSAR_LZ4_compress_fast(const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) + { + #if (HEAPMODE) +- void* ctxPtr = ALLOCATOR(1, sizeof(LZ4_stream_t)); /* malloc-calloc always properly aligned */ ++ void* ctxPtr = ALLOCATOR(1, sizeof(PULSAR_LZ4_stream_t)); /* malloc-calloc always properly aligned */ + #else +- LZ4_stream_t ctx; ++ PULSAR_LZ4_stream_t ctx; + void* ctxPtr = &ctx; + #endif + +- int result = LZ4_compress_fast_extState(ctxPtr, source, dest, inputSize, maxOutputSize, acceleration); ++ int result = PULSAR_LZ4_compress_fast_extState(ctxPtr, source, dest, inputSize, maxOutputSize, acceleration); + + #if (HEAPMODE) + FREEMEM(ctxPtr); +@@ -694,24 +694,24 @@ int LZ4_compress_fast(const char* source, char* dest, int inputSize, int maxOutp + } + + +-int LZ4_compress_default(const char* source, char* dest, int inputSize, int maxOutputSize) ++int PULSAR_LZ4_compress_default(const char* source, char* dest, int inputSize, int maxOutputSize) + { +- return LZ4_compress_fast(source, dest, inputSize, maxOutputSize, 1); ++ return PULSAR_LZ4_compress_fast(source, dest, inputSize, maxOutputSize, 1); + } + + + /* hidden debug function */ + /* strangely enough, gcc generates faster code when this function is uncommented, even if unused */ +-int LZ4_compress_fast_force(const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) ++int PULSAR_LZ4_compress_fast_force(const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) + { +- LZ4_stream_t ctx; ++ PULSAR_LZ4_stream_t ctx; + +- LZ4_resetStream(&ctx); ++ PULSAR_LZ4_resetStream(&ctx); + + if (inputSize < LZ4_64Klimit) +- return LZ4_compress_generic(&ctx, source, dest, inputSize, maxOutputSize, limitedOutput, byU16, noDict, noDictIssue, acceleration); ++ return PULSAR_LZ4_compress_generic(&ctx, source, dest, inputSize, maxOutputSize, limitedOutput, byU16, noDict, noDictIssue, acceleration); + else +- return LZ4_compress_generic(&ctx, source, dest, inputSize, maxOutputSize, limitedOutput, LZ4_64bits() ? byU32 : byPtr, noDict, noDictIssue, acceleration); ++ return PULSAR_LZ4_compress_generic(&ctx, source, dest, inputSize, maxOutputSize, limitedOutput, LZ4_64bits() ? byU32 : byPtr, noDict, noDictIssue, acceleration); + } + + +@@ -719,7 +719,7 @@ int LZ4_compress_fast_force(const char* source, char* dest, int inputSize, int m + * destSize variant + ********************************/ + +-static int LZ4_compress_destSize_generic( ++static int PULSAR_LZ4_compress_destSize_generic( + void* const ctx, + const char* const src, + char* const dst, +@@ -748,12 +748,12 @@ static int LZ4_compress_destSize_generic( + if (targetDstSize < 1) return 0; /* Impossible to store anything */ + if ((U32)*srcSizePtr > (U32)LZ4_MAX_INPUT_SIZE) return 0; /* Unsupported input size, too large (or negative) */ + if ((tableType == byU16) && (*srcSizePtr>=LZ4_64Klimit)) return 0; /* Size too large (not within 64K limit) */ +- if (*srcSizePtr> LZ4_skipTrigger); ++ step = (searchMatchNb++ >> PULSAR_LZ4_skipTrigger); + + if (unlikely(forwardIp > mflimit)) + goto _last_literals; + +- match = LZ4_getPositionOnHash(h, ctx, tableType, base); +- forwardH = LZ4_hashPosition(forwardIp, tableType); +- LZ4_putPositionOnHash(ip, h, ctx, tableType, base); ++ match = PULSAR_LZ4_getPositionOnHash(h, ctx, tableType, base); ++ forwardH = PULSAR_LZ4_hashPosition(forwardIp, tableType); ++ PULSAR_LZ4_putPositionOnHash(ip, h, ctx, tableType, base); + + } while ( ((tableType==byU16) ? 0 : (match + MAX_DISTANCE < ip)) +- || (LZ4_read32(match) != LZ4_read32(ip)) ); ++ || (PULSAR_LZ4_read32(match) != PULSAR_LZ4_read32(ip)) ); + } + + /* Catch up */ +@@ -806,19 +806,19 @@ static int LZ4_compress_destSize_generic( + else *token = (BYTE)(litLength< oMaxMatch) + { +@@ -845,17 +845,17 @@ _next_match: + if (op > oMaxSeq) break; + + /* Fill table */ +- LZ4_putPosition(ip-2, ctx, tableType, base); ++ PULSAR_LZ4_putPosition(ip-2, ctx, tableType, base); + + /* Test next position */ +- match = LZ4_getPosition(ip, ctx, tableType, base); +- LZ4_putPosition(ip, ctx, tableType, base); ++ match = PULSAR_LZ4_getPosition(ip, ctx, tableType, base); ++ PULSAR_LZ4_putPosition(ip, ctx, tableType, base); + if ( (match+MAX_DISTANCE>=ip) +- && (LZ4_read32(match)==LZ4_read32(ip)) ) ++ && (PULSAR_LZ4_read32(match)==PULSAR_LZ4_read32(ip)) ) + { token=op++; *token=0; goto _next_match; } + + /* Prepare next loop */ +- forwardH = LZ4_hashPosition(++ip, tableType); ++ forwardH = PULSAR_LZ4_hashPosition(++ip, tableType); + } + + _last_literals: +@@ -891,34 +891,34 @@ _last_literals: + } + + +-static int LZ4_compress_destSize_extState (void* state, const char* src, char* dst, int* srcSizePtr, int targetDstSize) ++static int PULSAR_LZ4_compress_destSize_extState (void* state, const char* src, char* dst, int* srcSizePtr, int targetDstSize) + { +- LZ4_resetStream((LZ4_stream_t*)state); ++ PULSAR_LZ4_resetStream((PULSAR_LZ4_stream_t*)state); + +- if (targetDstSize >= LZ4_compressBound(*srcSizePtr)) /* compression success is guaranteed */ ++ if (targetDstSize >= PULSAR_LZ4_compressBound(*srcSizePtr)) /* compression success is guaranteed */ + { +- return LZ4_compress_fast_extState(state, src, dst, *srcSizePtr, targetDstSize, 1); ++ return PULSAR_LZ4_compress_fast_extState(state, src, dst, *srcSizePtr, targetDstSize, 1); + } + else + { + if (*srcSizePtr < LZ4_64Klimit) +- return LZ4_compress_destSize_generic(state, src, dst, srcSizePtr, targetDstSize, byU16); ++ return PULSAR_LZ4_compress_destSize_generic(state, src, dst, srcSizePtr, targetDstSize, byU16); + else +- return LZ4_compress_destSize_generic(state, src, dst, srcSizePtr, targetDstSize, LZ4_64bits() ? byU32 : byPtr); ++ return PULSAR_LZ4_compress_destSize_generic(state, src, dst, srcSizePtr, targetDstSize, LZ4_64bits() ? byU32 : byPtr); + } + } + + +-int LZ4_compress_destSize(const char* src, char* dst, int* srcSizePtr, int targetDstSize) ++int PULSAR_LZ4_compress_destSize(const char* src, char* dst, int* srcSizePtr, int targetDstSize) + { + #if (HEAPMODE) +- void* ctx = ALLOCATOR(1, sizeof(LZ4_stream_t)); /* malloc-calloc always properly aligned */ ++ void* ctx = ALLOCATOR(1, sizeof(PULSAR_LZ4_stream_t)); /* malloc-calloc always properly aligned */ + #else +- LZ4_stream_t ctxBody; ++ PULSAR_LZ4_stream_t ctxBody; + void* ctx = &ctxBody; + #endif + +- int result = LZ4_compress_destSize_extState(ctx, src, dst, srcSizePtr, targetDstSize); ++ int result = PULSAR_LZ4_compress_destSize_extState(ctx, src, dst, srcSizePtr, targetDstSize); + + #if (HEAPMODE) + FREEMEM(ctx); +@@ -932,36 +932,36 @@ int LZ4_compress_destSize(const char* src, char* dst, int* srcSizePtr, int targe + * Streaming functions + ********************************/ + +-LZ4_stream_t* LZ4_createStream(void) ++PULSAR_LZ4_stream_t* PULSAR_LZ4_createStream(void) + { +- LZ4_stream_t* lz4s = (LZ4_stream_t*)ALLOCATOR(8, LZ4_STREAMSIZE_U64); +- LZ4_STATIC_ASSERT(LZ4_STREAMSIZE >= sizeof(LZ4_stream_t_internal)); /* A compilation error here means LZ4_STREAMSIZE is not large enough */ +- LZ4_resetStream(lz4s); ++ PULSAR_LZ4_stream_t* lz4s = (PULSAR_LZ4_stream_t*)ALLOCATOR(8, LZ4_STREAMSIZE_U64); ++ LZ4_STATIC_ASSERT(LZ4_STREAMSIZE >= sizeof(PULSAR_LZ4_stream_t_internal)); /* A compilation error here means LZ4_STREAMSIZE is not large enough */ ++ PULSAR_LZ4_resetStream(lz4s); + return lz4s; + } + +-void LZ4_resetStream (LZ4_stream_t* LZ4_stream) ++void PULSAR_LZ4_resetStream (PULSAR_LZ4_stream_t* PULSAR_LZ4_stream) + { +- MEM_INIT(LZ4_stream, 0, sizeof(LZ4_stream_t)); ++ MEM_INIT(PULSAR_LZ4_stream, 0, sizeof(PULSAR_LZ4_stream_t)); + } + +-int LZ4_freeStream (LZ4_stream_t* LZ4_stream) ++int PULSAR_LZ4_freeStream (PULSAR_LZ4_stream_t* PULSAR_LZ4_stream) + { +- FREEMEM(LZ4_stream); ++ FREEMEM(PULSAR_LZ4_stream); + return (0); + } + + + #define HASH_UNIT sizeof(size_t) +-int LZ4_loadDict (LZ4_stream_t* LZ4_dict, const char* dictionary, int dictSize) ++int PULSAR_LZ4_loadDict (PULSAR_LZ4_stream_t* PULSAR_LZ4_dict, const char* dictionary, int dictSize) + { +- LZ4_stream_t_internal* dict = (LZ4_stream_t_internal*) LZ4_dict; ++ PULSAR_LZ4_stream_t_internal* dict = (PULSAR_LZ4_stream_t_internal*) PULSAR_LZ4_dict; + const BYTE* p = (const BYTE*)dictionary; + const BYTE* const dictEnd = p + dictSize; + const BYTE* base; + + if ((dict->initCheck) || (dict->currentOffset > 1 GB)) /* Uninitialized structure, or reuse overflow */ +- LZ4_resetStream(LZ4_dict); ++ PULSAR_LZ4_resetStream(PULSAR_LZ4_dict); + + if (dictSize < (int)HASH_UNIT) + { +@@ -979,7 +979,7 @@ int LZ4_loadDict (LZ4_stream_t* LZ4_dict, const char* dictionary, int dictSize) + + while (p <= dictEnd-HASH_UNIT) + { +- LZ4_putPosition(p, dict->hashTable, byU32, base); ++ PULSAR_LZ4_putPosition(p, dict->hashTable, byU32, base); + p+=3; + } + +@@ -987,36 +987,36 @@ int LZ4_loadDict (LZ4_stream_t* LZ4_dict, const char* dictionary, int dictSize) + } + + +-static void LZ4_renormDictT(LZ4_stream_t_internal* LZ4_dict, const BYTE* src) ++static void PULSAR_LZ4_renormDictT(PULSAR_LZ4_stream_t_internal* PULSAR_LZ4_dict, const BYTE* src) + { +- if ((LZ4_dict->currentOffset > 0x80000000) || +- ((size_t)LZ4_dict->currentOffset > (size_t)src)) /* address space overflow */ ++ if ((PULSAR_LZ4_dict->currentOffset > 0x80000000) || ++ ((size_t)PULSAR_LZ4_dict->currentOffset > (size_t)src)) /* address space overflow */ + { + /* rescale hash table */ +- U32 delta = LZ4_dict->currentOffset - 64 KB; +- const BYTE* dictEnd = LZ4_dict->dictionary + LZ4_dict->dictSize; ++ U32 delta = PULSAR_LZ4_dict->currentOffset - 64 KB; ++ const BYTE* dictEnd = PULSAR_LZ4_dict->dictionary + PULSAR_LZ4_dict->dictSize; + int i; + for (i=0; ihashTable[i] < delta) LZ4_dict->hashTable[i]=0; +- else LZ4_dict->hashTable[i] -= delta; ++ if (PULSAR_LZ4_dict->hashTable[i] < delta) PULSAR_LZ4_dict->hashTable[i]=0; ++ else PULSAR_LZ4_dict->hashTable[i] -= delta; + } +- LZ4_dict->currentOffset = 64 KB; +- if (LZ4_dict->dictSize > 64 KB) LZ4_dict->dictSize = 64 KB; +- LZ4_dict->dictionary = dictEnd - LZ4_dict->dictSize; ++ PULSAR_LZ4_dict->currentOffset = 64 KB; ++ if (PULSAR_LZ4_dict->dictSize > 64 KB) PULSAR_LZ4_dict->dictSize = 64 KB; ++ PULSAR_LZ4_dict->dictionary = dictEnd - PULSAR_LZ4_dict->dictSize; + } + } + + +-int LZ4_compress_fast_continue (LZ4_stream_t* LZ4_stream, const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) ++int PULSAR_LZ4_compress_fast_continue (PULSAR_LZ4_stream_t* PULSAR_LZ4_stream, const char* source, char* dest, int inputSize, int maxOutputSize, int acceleration) + { +- LZ4_stream_t_internal* streamPtr = (LZ4_stream_t_internal*)LZ4_stream; ++ PULSAR_LZ4_stream_t_internal* streamPtr = (PULSAR_LZ4_stream_t_internal*)PULSAR_LZ4_stream; + const BYTE* const dictEnd = streamPtr->dictionary + streamPtr->dictSize; + + const BYTE* smallest = (const BYTE*) source; + if (streamPtr->initCheck) return 0; /* Uninitialized structure detected */ + if ((streamPtr->dictSize>0) && (smallest>dictEnd)) smallest = dictEnd; +- LZ4_renormDictT(streamPtr, smallest); ++ PULSAR_LZ4_renormDictT(streamPtr, smallest); + if (acceleration < 1) acceleration = ACCELERATION_DEFAULT; + + /* Check overlapping input/dictionary space */ +@@ -1036,9 +1036,9 @@ int LZ4_compress_fast_continue (LZ4_stream_t* LZ4_stream, const char* source, ch + { + int result; + if ((streamPtr->dictSize < 64 KB) && (streamPtr->dictSize < streamPtr->currentOffset)) +- result = LZ4_compress_generic(LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, withPrefix64k, dictSmall, acceleration); ++ result = PULSAR_LZ4_compress_generic(PULSAR_LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, withPrefix64k, dictSmall, acceleration); + else +- result = LZ4_compress_generic(LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, withPrefix64k, noDictIssue, acceleration); ++ result = PULSAR_LZ4_compress_generic(PULSAR_LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, withPrefix64k, noDictIssue, acceleration); + streamPtr->dictSize += (U32)inputSize; + streamPtr->currentOffset += (U32)inputSize; + return result; +@@ -1048,9 +1048,9 @@ int LZ4_compress_fast_continue (LZ4_stream_t* LZ4_stream, const char* source, ch + { + int result; + if ((streamPtr->dictSize < 64 KB) && (streamPtr->dictSize < streamPtr->currentOffset)) +- result = LZ4_compress_generic(LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, usingExtDict, dictSmall, acceleration); ++ result = PULSAR_LZ4_compress_generic(PULSAR_LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, usingExtDict, dictSmall, acceleration); + else +- result = LZ4_compress_generic(LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, usingExtDict, noDictIssue, acceleration); ++ result = PULSAR_LZ4_compress_generic(PULSAR_LZ4_stream, source, dest, inputSize, maxOutputSize, limitedOutput, byU32, usingExtDict, noDictIssue, acceleration); + streamPtr->dictionary = (const BYTE*)source; + streamPtr->dictSize = (U32)inputSize; + streamPtr->currentOffset += (U32)inputSize; +@@ -1060,17 +1060,17 @@ int LZ4_compress_fast_continue (LZ4_stream_t* LZ4_stream, const char* source, ch + + + /* Hidden debug function, to force external dictionary mode */ +-int LZ4_compress_forceExtDict (LZ4_stream_t* LZ4_dict, const char* source, char* dest, int inputSize) ++int PULSAR_LZ4_compress_forceExtDict (PULSAR_LZ4_stream_t* PULSAR_LZ4_dict, const char* source, char* dest, int inputSize) + { +- LZ4_stream_t_internal* streamPtr = (LZ4_stream_t_internal*)LZ4_dict; ++ PULSAR_LZ4_stream_t_internal* streamPtr = (PULSAR_LZ4_stream_t_internal*)PULSAR_LZ4_dict; + int result; + const BYTE* const dictEnd = streamPtr->dictionary + streamPtr->dictSize; + + const BYTE* smallest = dictEnd; + if (smallest > (const BYTE*) source) smallest = (const BYTE*) source; +- LZ4_renormDictT((LZ4_stream_t_internal*)LZ4_dict, smallest); ++ PULSAR_LZ4_renormDictT((PULSAR_LZ4_stream_t_internal*)PULSAR_LZ4_dict, smallest); + +- result = LZ4_compress_generic(LZ4_dict, source, dest, inputSize, 0, notLimited, byU32, usingExtDict, noDictIssue, 1); ++ result = PULSAR_LZ4_compress_generic(PULSAR_LZ4_dict, source, dest, inputSize, 0, notLimited, byU32, usingExtDict, noDictIssue, 1); + + streamPtr->dictionary = (const BYTE*)source; + streamPtr->dictSize = (U32)inputSize; +@@ -1080,9 +1080,9 @@ int LZ4_compress_forceExtDict (LZ4_stream_t* LZ4_dict, const char* source, char* + } + + +-int LZ4_saveDict (LZ4_stream_t* LZ4_dict, char* safeBuffer, int dictSize) ++int PULSAR_LZ4_saveDict (PULSAR_LZ4_stream_t* PULSAR_LZ4_dict, char* safeBuffer, int dictSize) + { +- LZ4_stream_t_internal* dict = (LZ4_stream_t_internal*) LZ4_dict; ++ PULSAR_LZ4_stream_t_internal* dict = (PULSAR_LZ4_stream_t_internal*) PULSAR_LZ4_dict; + const BYTE* previousDictEnd = dict->dictionary + dict->dictSize; + + if ((U32)dictSize > 64 KB) dictSize = 64 KB; /* useless to define a dictionary > 64 KB */ +@@ -1107,7 +1107,7 @@ int LZ4_saveDict (LZ4_stream_t* LZ4_dict, char* safeBuffer, int dictSize) + * Note that it is essential this generic function is really inlined, + * in order to remove useless branches during compilation optimization. + */ +-FORCE_INLINE int LZ4_decompress_generic( ++FORCE_INLINE int PULSAR_LZ4_decompress_generic( + const char* const source, + char* const dest, + int inputSize, +@@ -1188,11 +1188,11 @@ FORCE_INLINE int LZ4_decompress_generic( + op += length; + break; /* Necessarily EOF, due to parsing restrictions */ + } +- LZ4_wildCopy(op, ip, cpy); ++ PULSAR_LZ4_wildCopy(op, ip, cpy); + ip += length; op = cpy; + + /* get offset */ +- match = cpy - LZ4_readLE16(ip); ip+=2; ++ match = cpy - PULSAR_LZ4_readLE16(ip); ip+=2; + if ((checkOffset) && (unlikely(match < lowLimit))) goto _output_error; /* Error : offset outside destination buffer */ + + /* get matchlength */ +@@ -1253,23 +1253,23 @@ FORCE_INLINE int LZ4_decompress_generic( + op[2] = match[2]; + op[3] = match[3]; + match += dec32table[op-match]; +- LZ4_copy4(op+4, match); ++ PULSAR_LZ4_copy4(op+4, match); + op += 8; match -= dec64; +- } else { LZ4_copy8(op, match); op+=8; match+=8; } ++ } else { PULSAR_LZ4_copy8(op, match); op+=8; match+=8; } + + if (unlikely(cpy>oend-12)) + { + if (cpy > oend-LASTLITERALS) goto _output_error; /* Error : last LASTLITERALS bytes must be literals */ + if (op < oend-8) + { +- LZ4_wildCopy(op, match, oend-8); ++ PULSAR_LZ4_wildCopy(op, match, oend-8); + match += (oend-8) - op; + op = oend-8; + } + while (opprefixSize = (size_t) dictSize; + lz4sd->prefixEnd = (const BYTE*) dictionary + dictSize; + lz4sd->externalDict = NULL; +@@ -1350,16 +1350,16 @@ int LZ4_setStreamDecode (LZ4_streamDecode_t* LZ4_streamDecode, const char* dicti + These decoding functions allow decompression of multiple blocks in "streaming" mode. + Previously decoded blocks must still be available at the memory position where they were decoded. + If it's not possible, save the relevant part of decoded data into a safe buffer, +- and indicate where it stands using LZ4_setStreamDecode() ++ and indicate where it stands using PULSAR_LZ4_setStreamDecode() + */ +-int LZ4_decompress_safe_continue (LZ4_streamDecode_t* LZ4_streamDecode, const char* source, char* dest, int compressedSize, int maxOutputSize) ++int PULSAR_LZ4_decompress_safe_continue (PULSAR_LZ4_streamDecode_t* PULSAR_LZ4_streamDecode, const char* source, char* dest, int compressedSize, int maxOutputSize) + { +- LZ4_streamDecode_t_internal* lz4sd = (LZ4_streamDecode_t_internal*) LZ4_streamDecode; ++ PULSAR_LZ4_streamDecode_t_internal* lz4sd = (PULSAR_LZ4_streamDecode_t_internal*) PULSAR_LZ4_streamDecode; + int result; + + if (lz4sd->prefixEnd == (BYTE*)dest) + { +- result = LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, ++ result = PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, + endOnInputSize, full, 0, + usingExtDict, lz4sd->prefixEnd - lz4sd->prefixSize, lz4sd->externalDict, lz4sd->extDictSize); + if (result <= 0) return result; +@@ -1370,7 +1370,7 @@ int LZ4_decompress_safe_continue (LZ4_streamDecode_t* LZ4_streamDecode, const ch + { + lz4sd->extDictSize = lz4sd->prefixSize; + lz4sd->externalDict = lz4sd->prefixEnd - lz4sd->extDictSize; +- result = LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, ++ result = PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, + endOnInputSize, full, 0, + usingExtDict, (BYTE*)dest, lz4sd->externalDict, lz4sd->extDictSize); + if (result <= 0) return result; +@@ -1381,14 +1381,14 @@ int LZ4_decompress_safe_continue (LZ4_streamDecode_t* LZ4_streamDecode, const ch + return result; + } + +-int LZ4_decompress_fast_continue (LZ4_streamDecode_t* LZ4_streamDecode, const char* source, char* dest, int originalSize) ++int PULSAR_LZ4_decompress_fast_continue (PULSAR_LZ4_streamDecode_t* PULSAR_LZ4_streamDecode, const char* source, char* dest, int originalSize) + { +- LZ4_streamDecode_t_internal* lz4sd = (LZ4_streamDecode_t_internal*) LZ4_streamDecode; ++ PULSAR_LZ4_streamDecode_t_internal* lz4sd = (PULSAR_LZ4_streamDecode_t_internal*) PULSAR_LZ4_streamDecode; + int result; + + if (lz4sd->prefixEnd == (BYTE*)dest) + { +- result = LZ4_decompress_generic(source, dest, 0, originalSize, ++ result = PULSAR_LZ4_decompress_generic(source, dest, 0, originalSize, + endOnOutputSize, full, 0, + usingExtDict, lz4sd->prefixEnd - lz4sd->prefixSize, lz4sd->externalDict, lz4sd->extDictSize); + if (result <= 0) return result; +@@ -1399,7 +1399,7 @@ int LZ4_decompress_fast_continue (LZ4_streamDecode_t* LZ4_streamDecode, const ch + { + lz4sd->extDictSize = lz4sd->prefixSize; + lz4sd->externalDict = (BYTE*)dest - lz4sd->extDictSize; +- result = LZ4_decompress_generic(source, dest, 0, originalSize, ++ result = PULSAR_LZ4_decompress_generic(source, dest, 0, originalSize, + endOnOutputSize, full, 0, + usingExtDict, (BYTE*)dest, lz4sd->externalDict, lz4sd->extDictSize); + if (result <= 0) return result; +@@ -1418,33 +1418,33 @@ Advanced decoding functions : + the dictionary must be explicitly provided within parameters + */ + +-FORCE_INLINE int LZ4_decompress_usingDict_generic(const char* source, char* dest, int compressedSize, int maxOutputSize, int safe, const char* dictStart, int dictSize) ++FORCE_INLINE int PULSAR_LZ4_decompress_usingDict_generic(const char* source, char* dest, int compressedSize, int maxOutputSize, int safe, const char* dictStart, int dictSize) + { + if (dictSize==0) +- return LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, noDict, (BYTE*)dest, NULL, 0); ++ return PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, noDict, (BYTE*)dest, NULL, 0); + if (dictStart+dictSize == dest) + { + if (dictSize >= (int)(64 KB - 1)) +- return LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, withPrefix64k, (BYTE*)dest-64 KB, NULL, 0); +- return LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, noDict, (BYTE*)dest-dictSize, NULL, 0); ++ return PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, withPrefix64k, (BYTE*)dest-64 KB, NULL, 0); ++ return PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, noDict, (BYTE*)dest-dictSize, NULL, 0); + } +- return LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, usingExtDict, (BYTE*)dest, (const BYTE*)dictStart, dictSize); ++ return PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, safe, full, 0, usingExtDict, (BYTE*)dest, (const BYTE*)dictStart, dictSize); + } + +-int LZ4_decompress_safe_usingDict(const char* source, char* dest, int compressedSize, int maxOutputSize, const char* dictStart, int dictSize) ++int PULSAR_LZ4_decompress_safe_usingDict(const char* source, char* dest, int compressedSize, int maxOutputSize, const char* dictStart, int dictSize) + { +- return LZ4_decompress_usingDict_generic(source, dest, compressedSize, maxOutputSize, 1, dictStart, dictSize); ++ return PULSAR_LZ4_decompress_usingDict_generic(source, dest, compressedSize, maxOutputSize, 1, dictStart, dictSize); + } + +-int LZ4_decompress_fast_usingDict(const char* source, char* dest, int originalSize, const char* dictStart, int dictSize) ++int PULSAR_LZ4_decompress_fast_usingDict(const char* source, char* dest, int originalSize, const char* dictStart, int dictSize) + { +- return LZ4_decompress_usingDict_generic(source, dest, 0, originalSize, 0, dictStart, dictSize); ++ return PULSAR_LZ4_decompress_usingDict_generic(source, dest, 0, originalSize, 0, dictStart, dictSize); + } + + /* debug function */ +-int LZ4_decompress_safe_forceExtDict(const char* source, char* dest, int compressedSize, int maxOutputSize, const char* dictStart, int dictSize) ++int PULSAR_LZ4_decompress_safe_forceExtDict(const char* source, char* dest, int compressedSize, int maxOutputSize, const char* dictStart, int dictSize) + { +- return LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, endOnInputSize, full, 0, usingExtDict, (BYTE*)dest, (const BYTE*)dictStart, dictSize); ++ return PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, endOnInputSize, full, 0, usingExtDict, (BYTE*)dest, (const BYTE*)dictStart, dictSize); + } + + +@@ -1452,64 +1452,64 @@ int LZ4_decompress_safe_forceExtDict(const char* source, char* dest, int compres + * Obsolete Functions + ***************************************************/ + /* obsolete compression functions */ +-int LZ4_compress_limitedOutput(const char* source, char* dest, int inputSize, int maxOutputSize) { return LZ4_compress_default(source, dest, inputSize, maxOutputSize); } +-int LZ4_compress(const char* source, char* dest, int inputSize) { return LZ4_compress_default(source, dest, inputSize, LZ4_compressBound(inputSize)); } +-int LZ4_compress_limitedOutput_withState (void* state, const char* src, char* dst, int srcSize, int dstSize) { return LZ4_compress_fast_extState(state, src, dst, srcSize, dstSize, 1); } +-int LZ4_compress_withState (void* state, const char* src, char* dst, int srcSize) { return LZ4_compress_fast_extState(state, src, dst, srcSize, LZ4_compressBound(srcSize), 1); } +-int LZ4_compress_limitedOutput_continue (LZ4_stream_t* LZ4_stream, const char* src, char* dst, int srcSize, int maxDstSize) { return LZ4_compress_fast_continue(LZ4_stream, src, dst, srcSize, maxDstSize, 1); } +-int LZ4_compress_continue (LZ4_stream_t* LZ4_stream, const char* source, char* dest, int inputSize) { return LZ4_compress_fast_continue(LZ4_stream, source, dest, inputSize, LZ4_compressBound(inputSize), 1); } ++int PULSAR_LZ4_compress_limitedOutput(const char* source, char* dest, int inputSize, int maxOutputSize) { return PULSAR_LZ4_compress_default(source, dest, inputSize, maxOutputSize); } ++int PULSAR_LZ4_compress(const char* source, char* dest, int inputSize) { return PULSAR_LZ4_compress_default(source, dest, inputSize, PULSAR_LZ4_compressBound(inputSize)); } ++int PULSAR_LZ4_compress_limitedOutput_withState (void* state, const char* src, char* dst, int srcSize, int dstSize) { return PULSAR_LZ4_compress_fast_extState(state, src, dst, srcSize, dstSize, 1); } ++int PULSAR_LZ4_compress_withState (void* state, const char* src, char* dst, int srcSize) { return PULSAR_LZ4_compress_fast_extState(state, src, dst, srcSize, PULSAR_LZ4_compressBound(srcSize), 1); } ++int PULSAR_LZ4_compress_limitedOutput_continue (PULSAR_LZ4_stream_t* PULSAR_LZ4_stream, const char* src, char* dst, int srcSize, int maxDstSize) { return PULSAR_LZ4_compress_fast_continue(PULSAR_LZ4_stream, src, dst, srcSize, maxDstSize, 1); } ++int PULSAR_LZ4_compress_continue (PULSAR_LZ4_stream_t* PULSAR_LZ4_stream, const char* source, char* dest, int inputSize) { return PULSAR_LZ4_compress_fast_continue(PULSAR_LZ4_stream, source, dest, inputSize, PULSAR_LZ4_compressBound(inputSize), 1); } + + /* + These function names are deprecated and should no longer be used. + They are only provided here for compatibility with older user programs. +-- LZ4_uncompress is totally equivalent to LZ4_decompress_fast +-- LZ4_uncompress_unknownOutputSize is totally equivalent to LZ4_decompress_safe ++- PULSAR_LZ4_uncompress is totally equivalent to PULSAR_LZ4_decompress_fast ++- PULSAR_LZ4_uncompress_unknownOutputSize is totally equivalent to PULSAR_LZ4_decompress_safe + */ +-int LZ4_uncompress (const char* source, char* dest, int outputSize) { return LZ4_decompress_fast(source, dest, outputSize); } +-int LZ4_uncompress_unknownOutputSize (const char* source, char* dest, int isize, int maxOutputSize) { return LZ4_decompress_safe(source, dest, isize, maxOutputSize); } ++int PULSAR_LZ4_uncompress (const char* source, char* dest, int outputSize) { return PULSAR_LZ4_decompress_fast(source, dest, outputSize); } ++int PULSAR_LZ4_uncompress_unknownOutputSize (const char* source, char* dest, int isize, int maxOutputSize) { return PULSAR_LZ4_decompress_safe(source, dest, isize, maxOutputSize); } + + + /* Obsolete Streaming functions */ + +-int LZ4_sizeofStreamState() { return LZ4_STREAMSIZE; } ++int PULSAR_LZ4_sizeofStreamState() { return LZ4_STREAMSIZE; } + +-static void LZ4_init(LZ4_stream_t_internal* lz4ds, BYTE* base) ++static void PULSAR_LZ4_init(PULSAR_LZ4_stream_t_internal* lz4ds, BYTE* base) + { + MEM_INIT(lz4ds, 0, LZ4_STREAMSIZE); + lz4ds->bufferStart = base; + } + +-int LZ4_resetStreamState(void* state, char* inputBuffer) ++int PULSAR_LZ4_resetStreamState(void* state, char* inputBuffer) + { + if ((((size_t)state) & 3) != 0) return 1; /* Error : pointer is not aligned on 4-bytes boundary */ +- LZ4_init((LZ4_stream_t_internal*)state, (BYTE*)inputBuffer); ++ PULSAR_LZ4_init((PULSAR_LZ4_stream_t_internal*)state, (BYTE*)inputBuffer); + return 0; + } + +-void* LZ4_create (char* inputBuffer) ++void* PULSAR_LZ4_create (char* inputBuffer) + { + void* lz4ds = ALLOCATOR(8, LZ4_STREAMSIZE_U64); +- LZ4_init ((LZ4_stream_t_internal*)lz4ds, (BYTE*)inputBuffer); ++ PULSAR_LZ4_init ((PULSAR_LZ4_stream_t_internal*)lz4ds, (BYTE*)inputBuffer); + return lz4ds; + } + +-char* LZ4_slideInputBuffer (void* LZ4_Data) ++char* PULSAR_LZ4_slideInputBuffer (void* LZ4_Data) + { +- LZ4_stream_t_internal* ctx = (LZ4_stream_t_internal*)LZ4_Data; +- int dictSize = LZ4_saveDict((LZ4_stream_t*)LZ4_Data, (char*)ctx->bufferStart, 64 KB); ++ PULSAR_LZ4_stream_t_internal* ctx = (PULSAR_LZ4_stream_t_internal*)LZ4_Data; ++ int dictSize = PULSAR_LZ4_saveDict((PULSAR_LZ4_stream_t*)LZ4_Data, (char*)ctx->bufferStart, 64 KB); + return (char*)(ctx->bufferStart + dictSize); + } + + /* Obsolete streaming decompression functions */ + +-int LZ4_decompress_safe_withPrefix64k(const char* source, char* dest, int compressedSize, int maxOutputSize) ++int PULSAR_LZ4_decompress_safe_withPrefix64k(const char* source, char* dest, int compressedSize, int maxOutputSize) + { +- return LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, endOnInputSize, full, 0, withPrefix64k, (BYTE*)dest - 64 KB, NULL, 64 KB); ++ return PULSAR_LZ4_decompress_generic(source, dest, compressedSize, maxOutputSize, endOnInputSize, full, 0, withPrefix64k, (BYTE*)dest - 64 KB, NULL, 64 KB); + } + +-int LZ4_decompress_fast_withPrefix64k(const char* source, char* dest, int originalSize) ++int PULSAR_LZ4_decompress_fast_withPrefix64k(const char* source, char* dest, int originalSize) + { +- return LZ4_decompress_generic(source, dest, 0, originalSize, endOnOutputSize, full, 0, withPrefix64k, (BYTE*)dest - 64 KB, NULL, 64 KB); ++ return PULSAR_LZ4_decompress_generic(source, dest, 0, originalSize, endOnOutputSize, full, 0, withPrefix64k, (BYTE*)dest - 64 KB, NULL, 64 KB); + } + + #endif /* LZ4_COMMONDEFS_ONLY */ +diff --git a/pulsar-client-cpp/lib/lz4/lz4.h b/pulsar-client-cpp/lib/lz4/lz4.h +index c68232a..df593fe 100644 +--- a/pulsar-client-cpp/lib/lz4/lz4.h ++++ b/pulsar-client-cpp/lib/lz4/lz4.h +@@ -51,7 +51,7 @@ extern "C" { + #define LZ4_VERSION_MINOR 7 /* for new (non-breaking) interface capabilities */ + #define LZ4_VERSION_RELEASE 1 /* for tweaks, bug-fixes, or development */ + #define LZ4_VERSION_NUMBER (LZ4_VERSION_MAJOR * 100 * 100 + LZ4_VERSION_MINOR * 100 + LZ4_VERSION_RELEASE) +-int LZ4_versionNumber(void); ++int PULSAR_LZ4_versionNumber(void); + + /************************************** + * Tuning parameter +@@ -69,14 +69,14 @@ int LZ4_versionNumber(void); + * Simple Functions + **************************************/ + +-int LZ4_compress_default(const char* source, char* dest, int sourceSize, int maxDestSize); +-int LZ4_decompress_safe(const char* source, char* dest, int compressedSize, int maxDecompressedSize); ++int PULSAR_LZ4_compress_default(const char* source, char* dest, int sourceSize, int maxDestSize); ++int PULSAR_LZ4_decompress_safe(const char* source, char* dest, int compressedSize, int maxDecompressedSize); + + /* +-LZ4_compress_default() : ++PULSAR_LZ4_compress_default() : + Compresses 'sourceSize' bytes from buffer 'source' + into already allocated 'dest' buffer of size 'maxDestSize'. +- Compression is guaranteed to succeed if 'maxDestSize' >= LZ4_compressBound(sourceSize). ++ Compression is guaranteed to succeed if 'maxDestSize' >= PULSAR_LZ4_compressBound(sourceSize). + It also runs faster, so it's a recommended setting. + If the function cannot compress 'source' into a more limited 'dest' budget, + compression stops *immediately*, and the function result is zero. +@@ -87,7 +87,7 @@ LZ4_compress_default() : + return : the number of bytes written into buffer 'dest' (necessarily <= maxOutputSize) + or 0 if compression fails + +-LZ4_decompress_safe() : ++PULSAR_LZ4_decompress_safe() : + compressedSize : is the precise full size of the compressed block. + maxDecompressedSize : is the size of destination buffer, which must be already allocated. + return : the number of bytes decompressed into destination buffer (necessarily <= maxDecompressedSize) +@@ -106,42 +106,42 @@ result. + ((unsigned)(isize) > (unsigned)LZ4_MAX_INPUT_SIZE ? 0 : (isize) + ((isize) / 255) + 16) + + /* +-LZ4_compressBound() : ++PULSAR_LZ4_compressBound() : + Provides the maximum size that LZ4 compression may output in a "worst case" scenario (input data not + compressible) + This function is primarily useful for memory allocation purposes (destination buffer size). + Macro LZ4_COMPRESSBOUND() is also provided for compilation-time evaluation (stack memory allocation for + example). +- Note that LZ4_compress_default() compress faster when dest buffer size is >= LZ4_compressBound(srcSize) ++ Note that PULSAR_LZ4_compress_default() compress faster when dest buffer size is >= PULSAR_LZ4_compressBound(srcSize) + inputSize : max supported value is LZ4_MAX_INPUT_SIZE + return : maximum output size in a "worst case" scenario + or 0, if input size is too large ( > LZ4_MAX_INPUT_SIZE) + */ +-int LZ4_compressBound(int inputSize); ++int PULSAR_LZ4_compressBound(int inputSize); + + /* +-LZ4_compress_fast() : +- Same as LZ4_compress_default(), but allows to select an "acceleration" factor. ++PULSAR_LZ4_compress_fast() : ++ Same as PULSAR_LZ4_compress_default(), but allows to select an "acceleration" factor. + The larger the acceleration value, the faster the algorithm, but also the lesser the compression. + It's a trade-off. It can be fine tuned, with each successive value providing roughly +~3% to speed. +- An acceleration value of "1" is the same as regular LZ4_compress_default() ++ An acceleration value of "1" is the same as regular PULSAR_LZ4_compress_default() + Values <= 0 will be replaced by ACCELERATION_DEFAULT (see lz4.c), which is 1. + */ +-int LZ4_compress_fast(const char* source, char* dest, int sourceSize, int maxDestSize, int acceleration); ++int PULSAR_LZ4_compress_fast(const char* source, char* dest, int sourceSize, int maxDestSize, int acceleration); + + /* +-LZ4_compress_fast_extState() : ++PULSAR_LZ4_compress_fast_extState() : + Same compression function, just using an externally allocated memory space to store compression state. +- Use LZ4_sizeofState() to know how much memory must be allocated, ++ Use PULSAR_LZ4_sizeofState() to know how much memory must be allocated, + and allocate it on 8-bytes boundaries (using malloc() typically). + Then, provide it as 'void* state' to compression function. + */ +-int LZ4_sizeofState(void); +-int LZ4_compress_fast_extState(void* state, const char* source, char* dest, int inputSize, int maxDestSize, ++int PULSAR_LZ4_sizeofState(void); ++int PULSAR_LZ4_compress_fast_extState(void* state, const char* source, char* dest, int inputSize, int maxDestSize, + int acceleration); + + /* +-LZ4_compress_destSize() : ++PULSAR_LZ4_compress_destSize() : + Reverse the logic, by compressing as much data as possible from 'source' buffer + into already allocated buffer 'dest' of size 'targetDestSize'. + This function either compresses the entire 'source' content into 'dest' if it's large enough, +@@ -151,25 +151,25 @@ LZ4_compress_destSize() : + return : Nb bytes written into 'dest' (necessarily <= targetDestSize) + or 0 if compression fails + */ +-int LZ4_compress_destSize(const char* source, char* dest, int* sourceSizePtr, int targetDestSize); ++int PULSAR_LZ4_compress_destSize(const char* source, char* dest, int* sourceSizePtr, int targetDestSize); + + /* +-LZ4_decompress_fast() : ++PULSAR_LZ4_decompress_fast() : + originalSize : is the original and therefore uncompressed size + return : the number of bytes read from the source buffer (in other words, the compressed size) + If the source stream is detected malformed, the function will stop decoding and return a negative + result. + Destination buffer must be already allocated. Its size must be a minimum of 'originalSize' bytes. + note : This function fully respect memory boundaries for properly formed compressed data. +- It is a bit faster than LZ4_decompress_safe(). ++ It is a bit faster than PULSAR_LZ4_decompress_safe(). + However, it does not provide any protection against intentionally modified data stream (malicious + input). + Use this function in trusted environment only (data to decode comes from a trusted source). + */ +-int LZ4_decompress_fast(const char* source, char* dest, int originalSize); ++int PULSAR_LZ4_decompress_fast(const char* source, char* dest, int originalSize); + + /* +-LZ4_decompress_safe_partial() : ++PULSAR_LZ4_decompress_safe_partial() : + This function decompress a compressed block of size 'compressedSize' at position 'source' + into destination buffer 'dest' of size 'maxDecompressedSize'. + The function tries to stop decompressing operation as soon as 'targetOutputSize' has been reached, +@@ -182,7 +182,7 @@ result. + This function never writes outside of output buffer, and never reads outside of input buffer. It + is therefore protected against malicious data packets + */ +-int LZ4_decompress_safe_partial(const char* source, char* dest, int compressedSize, int targetOutputSize, ++int PULSAR_LZ4_decompress_safe_partial(const char* source, char* dest, int compressedSize, int targetOutputSize, + int maxDecompressedSize); + + /*********************************************** +@@ -191,7 +191,7 @@ int LZ4_decompress_safe_partial(const char* source, char* dest, int compressedSi + #define LZ4_STREAMSIZE_U64 ((1 << (LZ4_MEMORY_USAGE - 3)) + 4) + #define LZ4_STREAMSIZE (LZ4_STREAMSIZE_U64 * sizeof(long long)) + /* +- * LZ4_stream_t ++ * PULSAR_LZ4_stream_t + * information structure to track an LZ4 stream. + * important : init this structure content before first use ! + * note : only allocated directly the structure if you are statically linking LZ4 +@@ -200,55 +200,55 @@ int LZ4_decompress_safe_partial(const char* source, char* dest, int compressedSi + // clang-format off + typedef struct { + long long table[LZ4_STREAMSIZE_U64]; +-} LZ4_stream_t; ++} PULSAR_LZ4_stream_t; + // clang-format on + + /* +- * LZ4_resetStream +- * Use this function to init an allocated LZ4_stream_t structure ++ * PULSAR_LZ4_resetStream ++ * Use this function to init an allocated PULSAR_LZ4_stream_t structure + */ +-void LZ4_resetStream(LZ4_stream_t* streamPtr); ++void PULSAR_LZ4_resetStream(PULSAR_LZ4_stream_t* streamPtr); + + /* +- * LZ4_createStream will allocate and initialize an LZ4_stream_t structure +- * LZ4_freeStream releases its memory. ++ * PULSAR_LZ4_createStream will allocate and initialize an PULSAR_LZ4_stream_t structure ++ * PULSAR_LZ4_freeStream releases its memory. + * In the context of a DLL (liblz4), please use these methods rather than the static struct. +- * They are more future proof, in case of a change of LZ4_stream_t size. ++ * They are more future proof, in case of a change of PULSAR_LZ4_stream_t size. + */ +-LZ4_stream_t* LZ4_createStream(void); +-int LZ4_freeStream(LZ4_stream_t* streamPtr); ++PULSAR_LZ4_stream_t* PULSAR_LZ4_createStream(void); ++int PULSAR_LZ4_freeStream(PULSAR_LZ4_stream_t* streamPtr); + + /* +- * LZ4_loadDict +- * Use this function to load a static dictionary into LZ4_stream. ++ * PULSAR_LZ4_loadDict ++ * Use this function to load a static dictionary into PULSAR_LZ4_stream. + * Any previous data will be forgotten, only 'dictionary' will remain in memory. + * Loading a size of 0 is allowed. + * Return : dictionary size, in bytes (necessarily <= 64 KB) + */ +-int LZ4_loadDict(LZ4_stream_t* streamPtr, const char* dictionary, int dictSize); ++int PULSAR_LZ4_loadDict(PULSAR_LZ4_stream_t* streamPtr, const char* dictionary, int dictSize); + + /* +- * LZ4_compress_fast_continue ++ * PULSAR_LZ4_compress_fast_continue + * Compress buffer content 'src', using data from previously compressed blocks as dictionary to improve + * compression ratio. + * Important : Previous data blocks are assumed to still be present and unmodified ! + * 'dst' buffer must be already allocated. +- * If maxDstSize >= LZ4_compressBound(srcSize), compression is guaranteed to succeed, and runs faster. ++ * If maxDstSize >= PULSAR_LZ4_compressBound(srcSize), compression is guaranteed to succeed, and runs faster. + * If not, and if compressed data cannot fit into 'dst' buffer size, compression stops, and function returns a + * zero. + */ +-int LZ4_compress_fast_continue(LZ4_stream_t* streamPtr, const char* src, char* dst, int srcSize, ++int PULSAR_LZ4_compress_fast_continue(PULSAR_LZ4_stream_t* streamPtr, const char* src, char* dst, int srcSize, + int maxDstSize, int acceleration); + + /* +- * LZ4_saveDict ++ * PULSAR_LZ4_saveDict + * If previously compressed data block is not guaranteed to remain available at its memory location + * save it into a safer place (char* safeBuffer) +- * Note : you don't need to call LZ4_loadDict() afterwards, +- * dictionary is immediately usable, you can therefore call LZ4_compress_fast_continue() ++ * Note : you don't need to call PULSAR_LZ4_loadDict() afterwards, ++ * dictionary is immediately usable, you can therefore call PULSAR_LZ4_compress_fast_continue() + * Return : saved dictionary size in bytes (necessarily <= dictSize), or 0 if error + */ +-int LZ4_saveDict(LZ4_stream_t* streamPtr, char* safeBuffer, int dictSize); ++int PULSAR_LZ4_saveDict(PULSAR_LZ4_stream_t* streamPtr, char* safeBuffer, int dictSize); + + /************************************************ + * Streaming Decompression Functions +@@ -259,29 +259,29 @@ int LZ4_saveDict(LZ4_stream_t* streamPtr, char* safeBuffer, int dictSize); + // clang-format off + typedef struct { + unsigned long long table[LZ4_STREAMDECODESIZE_U64]; +-} LZ4_streamDecode_t; ++} PULSAR_LZ4_streamDecode_t; + // clang-format on + + /* +- * LZ4_streamDecode_t ++ * PULSAR_LZ4_streamDecode_t + * information structure to track an LZ4 stream. +- * init this structure content using LZ4_setStreamDecode or memset() before first use ! ++ * init this structure content using PULSAR_LZ4_setStreamDecode or memset() before first use ! + * + * In the context of a DLL (liblz4) please prefer usage of construction methods below. +- * They are more future proof, in case of a change of LZ4_streamDecode_t size in the future. +- * LZ4_createStreamDecode will allocate and initialize an LZ4_streamDecode_t structure +- * LZ4_freeStreamDecode releases its memory. ++ * They are more future proof, in case of a change of PULSAR_LZ4_streamDecode_t size in the future. ++ * PULSAR_LZ4_createStreamDecode will allocate and initialize an PULSAR_LZ4_streamDecode_t structure ++ * PULSAR_LZ4_freeStreamDecode releases its memory. + */ +-LZ4_streamDecode_t* LZ4_createStreamDecode(void); +-int LZ4_freeStreamDecode(LZ4_streamDecode_t* LZ4_stream); ++PULSAR_LZ4_streamDecode_t* PULSAR_LZ4_createStreamDecode(void); ++int PULSAR_LZ4_freeStreamDecode(PULSAR_LZ4_streamDecode_t* PULSAR_LZ4_stream); + + /* +- * LZ4_setStreamDecode ++ * PULSAR_LZ4_setStreamDecode + * Use this function to instruct where to find the dictionary. + * Setting a size of 0 is allowed (same effect as reset). + * Return : 1 if OK, 0 if error + */ +-int LZ4_setStreamDecode(LZ4_streamDecode_t* LZ4_streamDecode, const char* dictionary, int dictSize); ++int PULSAR_LZ4_setStreamDecode(PULSAR_LZ4_streamDecode_t* PULSAR_LZ4_streamDecode, const char* dictionary, int dictSize); + + /* + *_continue() : +@@ -301,23 +301,23 @@ block. + In which case, encoding and decoding buffers do not need to be synchronized, + and encoding ring buffer can have any size, including larger than decoding buffer. + Whenever these conditions are not possible, save the last 64KB of decoded data into a safe buffer, +- and indicate where it is saved using LZ4_setStreamDecode() ++ and indicate where it is saved using PULSAR_LZ4_setStreamDecode() + */ +-int LZ4_decompress_safe_continue(LZ4_streamDecode_t* LZ4_streamDecode, const char* source, char* dest, ++int PULSAR_LZ4_decompress_safe_continue(PULSAR_LZ4_streamDecode_t* PULSAR_LZ4_streamDecode, const char* source, char* dest, + int compressedSize, int maxDecompressedSize); +-int LZ4_decompress_fast_continue(LZ4_streamDecode_t* LZ4_streamDecode, const char* source, char* dest, ++int PULSAR_LZ4_decompress_fast_continue(PULSAR_LZ4_streamDecode_t* PULSAR_LZ4_streamDecode, const char* source, char* dest, + int originalSize); + + /* + Advanced decoding functions : + *_usingDict() : + These decoding functions work the same as +- a combination of LZ4_setStreamDecode() followed by LZ4_decompress_x_continue() +- They are stand-alone. They don't need nor update an LZ4_streamDecode_t structure. ++ a combination of PULSAR_LZ4_setStreamDecode() followed by PULSAR_LZ4_decompress_x_continue() ++ They are stand-alone. They don't need nor update an PULSAR_LZ4_streamDecode_t structure. + */ +-int LZ4_decompress_safe_usingDict(const char* source, char* dest, int compressedSize, int maxDecompressedSize, ++int PULSAR_LZ4_decompress_safe_usingDict(const char* source, char* dest, int compressedSize, int maxDecompressedSize, + const char* dictStart, int dictSize); +-int LZ4_decompress_fast_usingDict(const char* source, char* dest, int originalSize, const char* dictStart, ++int PULSAR_LZ4_decompress_fast_usingDict(const char* source, char* dest, int originalSize, const char* dictStart, + int dictSize); + + /************************************** +@@ -346,36 +346,36 @@ int LZ4_decompress_fast_usingDict(const char* source, char* dest, int originalSi + + /* Obsolete compression functions */ + /* These functions are planned to start generate warnings by r131 approximately */ +-int LZ4_compress(const char* source, char* dest, int sourceSize); +-int LZ4_compress_limitedOutput(const char* source, char* dest, int sourceSize, int maxOutputSize); +-int LZ4_compress_withState(void* state, const char* source, char* dest, int inputSize); +-int LZ4_compress_limitedOutput_withState(void* state, const char* source, char* dest, int inputSize, ++int PULSAR_LZ4_compress(const char* source, char* dest, int sourceSize); ++int PULSAR_LZ4_compress_limitedOutput(const char* source, char* dest, int sourceSize, int maxOutputSize); ++int PULSAR_LZ4_compress_withState(void* state, const char* source, char* dest, int inputSize); ++int PULSAR_LZ4_compress_limitedOutput_withState(void* state, const char* source, char* dest, int inputSize, + int maxOutputSize); +-int LZ4_compress_continue(LZ4_stream_t* LZ4_streamPtr, const char* source, char* dest, int inputSize); +-int LZ4_compress_limitedOutput_continue(LZ4_stream_t* LZ4_streamPtr, const char* source, char* dest, ++int PULSAR_LZ4_compress_continue(PULSAR_LZ4_stream_t* PULSAR_LZ4_streamPtr, const char* source, char* dest, int inputSize); ++int PULSAR_LZ4_compress_limitedOutput_continue(PULSAR_LZ4_stream_t* PULSAR_LZ4_streamPtr, const char* source, char* dest, + int inputSize, int maxOutputSize); + + /* Obsolete decompression functions */ + /* These function names are completely deprecated and must no longer be used. + They are only provided here for compatibility with older programs. +- - LZ4_uncompress is the same as LZ4_decompress_fast +- - LZ4_uncompress_unknownOutputSize is the same as LZ4_decompress_safe ++ - PULSAR_LZ4_uncompress is the same as PULSAR_LZ4_decompress_fast ++ - PULSAR_LZ4_uncompress_unknownOutputSize is the same as PULSAR_LZ4_decompress_safe + These function prototypes are now disabled; uncomment them only if you really need them. + It is highly recommended to stop using these prototypes and migrate to maintained ones */ +-/* int LZ4_uncompress (const char* source, char* dest, int outputSize); */ +-/* int LZ4_uncompress_unknownOutputSize (const char* source, char* dest, int isize, int maxOutputSize); */ ++/* int PULSAR_LZ4_uncompress (const char* source, char* dest, int outputSize); */ ++/* int PULSAR_LZ4_uncompress_unknownOutputSize (const char* source, char* dest, int isize, int maxOutputSize); */ + + /* Obsolete streaming functions; use new streaming interface whenever possible */ +-LZ4_DEPRECATED("use LZ4_createStream() instead") void* LZ4_create(char* inputBuffer); +-LZ4_DEPRECATED("use LZ4_createStream() instead") int LZ4_sizeofStreamState(void); +-LZ4_DEPRECATED("use LZ4_resetStream() instead") int LZ4_resetStreamState(void* state, char* inputBuffer); +-LZ4_DEPRECATED("use LZ4_saveDict() instead") char* LZ4_slideInputBuffer(void* state); ++LZ4_DEPRECATED("use PULSAR_LZ4_createStream() instead") void* PULSAR_LZ4_create(char* inputBuffer); ++LZ4_DEPRECATED("use PULSAR_LZ4_createStream() instead") int PULSAR_LZ4_sizeofStreamState(void); ++LZ4_DEPRECATED("use PULSAR_LZ4_resetStream() instead") int PULSAR_LZ4_resetStreamState(void* state, char* inputBuffer); ++LZ4_DEPRECATED("use PULSAR_LZ4_saveDict() instead") char* PULSAR_LZ4_slideInputBuffer(void* state); + + /* Obsolete streaming decoding functions */ +-LZ4_DEPRECATED("use LZ4_decompress_safe_usingDict() instead") +-int LZ4_decompress_safe_withPrefix64k(const char* src, char* dst, int compressedSize, int maxDstSize); +-LZ4_DEPRECATED("use LZ4_decompress_fast_usingDict() instead") +-int LZ4_decompress_fast_withPrefix64k(const char* src, char* dst, int originalSize); ++LZ4_DEPRECATED("use PULSAR_LZ4_decompress_safe_usingDict() instead") ++int PULSAR_LZ4_decompress_safe_withPrefix64k(const char* src, char* dst, int compressedSize, int maxDstSize); ++LZ4_DEPRECATED("use PULSAR_LZ4_decompress_fast_usingDict() instead") ++int PULSAR_LZ4_decompress_fast_withPrefix64k(const char* src, char* dst, int originalSize); + + #if defined(__cplusplus) + } diff --git a/libs/setup.sh b/libs/setup.sh index e20783188..282dfaf52 100755 --- a/libs/setup.sh +++ b/libs/setup.sh @@ -14,39 +14,49 @@ clone () { local git_repo=$1 local dir_name=$2 local checkout_id=$3 - shift 3 + local shallow=$4 + shift 4 # Clone if there's no repo. if [[ ! -d "$dir_name" ]]; then echo "Cloning from $git_repo" # If the clone fails, it doesn't make sense to continue with the function # execution but the whole script should continue executing because we might # clone the same repo from a different source. - git clone "$git_repo" "$dir_name" || return 1 + + if [ "$shallow" = true ]; then + git clone --depth 1 --branch "$checkout_id" "$git_repo" "$dir_name" || return 1 + else + git clone "$git_repo" "$dir_name" || return 1 + fi fi pushd "$dir_name" - # Just fetch new commits from remote repository. Don't merge/pull them in, so - # that we don't clobber local modifications. - git fetch # Check whether we have any local changes which need to be preserved. local local_changes=true if git diff --no-ext-diff --quiet && git diff --no-ext-diff --cached --quiet; then local_changes=false fi - # Stash regardless of local_changes, so that a user gets a message on stdout. - git stash - # Checkout the primary commit (there's no need to pull/merge). - # The checkout fail should exit this script immediately because the target - # commit is not there and that will most likely create build-time errors. - git checkout "$checkout_id" || exit 1 - # Apply any optional cherry pick fixes. - while [[ $# -ne 0 ]]; do - local cherry_pick_id=$1 - shift - # The cherry-pick fail should exit this script immediately because the - # target commit is not there and that will most likely create build-time - # errors. - git cherry-pick -n "$cherry_pick_id" || exit 1 - done + + if [ "$shallow" = false ]; then + # Stash regardless of local_changes, so that a user gets a message on stdout. + git stash + # Just fetch new commits from remote repository. Don't merge/pull them in, so + # that we don't clobber local modifications. + git fetch + # Checkout the primary commit (there's no need to pull/merge). + # The checkout fail should exit this script immediately because the target + # commit is not there and that will most likely create build-time errors. + git checkout "$checkout_id" || exit 1 + # Apply any optional cherry pick fixes. + while [[ $# -ne 0 ]]; do + local cherry_pick_id=$1 + shift + # The cherry-pick fail should exit this script immediately because the + # target commit is not there and that will most likely create build-time + # errors. + git cherry-pick -n "$cherry_pick_id" || exit 1 + done + fi + # Reapply any local changes. if [[ $local_changes == true ]]; then git stash pop @@ -70,12 +80,13 @@ repo_clone_try_double () { secondary_url="$2" folder_name="$3" ref="$4" + shallow="${5:-false}" echo "Cloning primary from $primary_url secondary from $secondary_url" if [ -z "$primary_url" ]; then echo "Primary should not be empty." && exit 1; fi if [ -z "$secondary_url" ]; then echo "Secondary should not be empty." && exit 1; fi if [ -z "$folder_name" ]; then echo "Clone folder should not be empty." && exit 1; fi if [ -z "$ref" ]; then echo "Git clone ref should not be empty." && exit 1; fi - clone "$primary_url" "$folder_name" "$ref" || clone "$secondary_url" "$folder_name" "$ref" || exit 1 + clone "$primary_url" "$folder_name" "$ref" "$shallow" || clone "$secondary_url" "$folder_name" "$ref" "$shallow" || exit 1 echo "" } @@ -113,6 +124,9 @@ declare -A primary_urls=( ["nlohmann"]="http://$local_cache_host/file/nlohmann/json/b3e5cb7f20dcc5c806e418df34324eca60d17d4e/single_include/nlohmann/json.hpp" ["neo4j"]="http://$local_cache_host/file/neo4j-community-3.2.3-unix.tar.gz" ["librdkafka"]="http://$local_cache_host/git/librdkafka.git" + ["protobuf"]="http://$local_cache_host/git/protobuf.git" + ["boost"]="https://boostorg.jfrog.io/artifactory/main/release/1.77.0/source/boost_1_77_0.tar.gz" + ["pulsar"]="https://github.com/apache/pulsar.git" ) # The goal of secondary urls is to have links to the "source of truth" of @@ -140,13 +154,16 @@ declare -A secondary_urls=( ["nlohmann"]="https://raw.githubusercontent.com/nlohmann/json/b3e5cb7f20dcc5c806e418df34324eca60d17d4e/single_include/nlohmann/json.hpp" ["neo4j"]="https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/neo4j-community-3.2.3-unix.tar.gz" ["librdkafka"]="https://github.com/edenhill/librdkafka.git" + ["protobuf"]="https://github.com/protocolbuffers/protobuf.git" + ["boost"]="https://boostorg.jfrog.io/artifactory/main/release/1.77.0/source/boost_1_77_0.tar.gz" + ["pulsar"]="https://github.com/apache/pulsar.git" ) # antlr file_get_try_double "${primary_urls[antlr4-generator]}" "${secondary_urls[antlr4-generator]}" -antlr4_tag="5e5b6d35b4183fd330102c40947b95c4b5c6abb5" # v4.9.2 -repo_clone_try_double "${primary_urls[antlr4-code]}" "${secondary_urls[antlr4-code]}" "antlr4" "$antlr4_tag" +antlr4_tag="4.9.2" # v4.9.2 +repo_clone_try_double "${primary_urls[antlr4-code]}" "${secondary_urls[antlr4-code]}" "antlr4" "$antlr4_tag" true # remove shared library from install dependencies sed -i 's/install(TARGETS antlr4_shared/install(TARGETS antlr4_shared OPTIONAL/' antlr4/runtime/Cpp/runtime/CMakeLists.txt # fix issue https://github.com/antlr/antlr4/issues/3194 - should update Antlr commit once the PR related to the issue gets merged @@ -161,20 +178,20 @@ cppitertools_ref="cb3635456bdb531121b82b4d2e3afc7ae1f56d47" repo_clone_try_double "${primary_urls[cppitertools]}" "${secondary_urls[cppitertools]}" "cppitertools" "$cppitertools_ref" # fmt -fmt_tag="7bdf0628b1276379886c7f6dda2cef2b3b374f0b" # (2020-11-25) -repo_clone_try_double "${primary_urls[fmt]}" "${secondary_urls[fmt]}" "fmt" "$fmt_tag" +fmt_tag="7.1.3" # (2020-11-25) +repo_clone_try_double "${primary_urls[fmt]}" "${secondary_urls[fmt]}" "fmt" "$fmt_tag" true # rapidcheck rapidcheck_tag="7bc7d302191a4f3d0bf005692677126136e02f60" # (2020-05-04) repo_clone_try_double "${primary_urls[rapidcheck]}" "${secondary_urls[rapidcheck]}" "rapidcheck" "$rapidcheck_tag" # google benchmark -benchmark_tag="4f8bfeae470950ef005327973f15b0044eceaceb" # v1.1.0 -repo_clone_try_double "${primary_urls[gbenchmark]}" "${secondary_urls[gbenchmark]}" "benchmark" "$benchmark_tag" +benchmark_tag="v1.1.0" +repo_clone_try_double "${primary_urls[gbenchmark]}" "${secondary_urls[gbenchmark]}" "benchmark" "$benchmark_tag" true # google test -googletest_tag="ec44c6c1675c25b9827aacd08c02433cccde7780" # v1.8.0 -repo_clone_try_double "${primary_urls[gtest]}" "${secondary_urls[gtest]}" "googletest" "$googletest_tag" +googletest_tag="release-1.8.0" +repo_clone_try_double "${primary_urls[gtest]}" "${secondary_urls[gtest]}" "googletest" "$googletest_tag" true # google flags gflags_tag="b37ceb03a0e56c9f15ce80409438a555f8a67b7c" # custom version (May 6, 2017) @@ -201,19 +218,19 @@ cd .. bzip2_tag="0405487e2b1de738e7f1c8afb50d19cf44e8d580" # v1.0.6 (May 26, 2011) repo_clone_try_double "${primary_urls[bzip2]}" "${secondary_urls[bzip2]}" "bzip2" "$bzip2_tag" -zlib_tag="cacf7f1d4e3d44d871b605da3b647f07d718623f" # v1.2.11. -repo_clone_try_double "${primary_urls[zlib]}" "${secondary_urls[zlib]}" "zlib" "$zlib_tag" +zlib_tag="v1.2.11" # v1.2.11. +repo_clone_try_double "${primary_urls[zlib]}" "${secondary_urls[zlib]}" "zlib" "$zlib_tag" true # remove shared library from install dependencies sed -i 's/install(TARGETS zlib zlibstatic/install(TARGETS zlibstatic/g' zlib/CMakeLists.txt -rocksdb_tag="f3e33549c151f30ac4eb7c22356c6d0331f37652" # (2020-10-14) -repo_clone_try_double "${primary_urls[rocksdb]}" "${secondary_urls[rocksdb]}" "rocksdb" "$rocksdb_tag" +rocksdb_tag="v6.14.6" # (2020-10-14) +repo_clone_try_double "${primary_urls[rocksdb]}" "${secondary_urls[rocksdb]}" "rocksdb" "$rocksdb_tag" true # remove shared library from install dependencies sed -i 's/TARGETS ${ROCKSDB_SHARED_LIB}/TARGETS ${ROCKSDB_SHARED_LIB} OPTIONAL/' rocksdb/CMakeLists.txt # mgclient mgclient_tag="v1.3.0" # (2021-09-23) -repo_clone_try_double "${primary_urls[mgclient]}" "${secondary_urls[mgclient]}" "mgclient" "$mgclient_tag" +repo_clone_try_double "${primary_urls[mgclient]}" "${secondary_urls[mgclient]}" "mgclient" "$mgclient_tag" true sed -i 's/\${CMAKE_INSTALL_LIBDIR}/lib/' mgclient/src/CMakeLists.txt # pymgclient @@ -222,10 +239,10 @@ repo_clone_try_double "${primary_urls[pymgclient]}" "${secondary_urls[pymgclient # mgconsole mgconsole_tag="v1.1.0" # (2021-10-07) -repo_clone_try_double "${primary_urls[mgconsole]}" "${secondary_urls[mgconsole]}" "mgconsole" "$mgconsole_tag" +repo_clone_try_double "${primary_urls[mgconsole]}" "${secondary_urls[mgconsole]}" "mgconsole" "$mgconsole_tag" true -spdlog_tag="46d418164dd4cd9822cf8ca62a116a3f71569241" # (2020-12-01) -repo_clone_try_double "${primary_urls[spdlog]}" "${secondary_urls[spdlog]}" "spdlog" "$spdlog_tag" +spdlog_tag="v1.8.2" # (2020-12-01) +repo_clone_try_double "${primary_urls[spdlog]}" "${secondary_urls[spdlog]}" "spdlog" "$spdlog_tag" true jemalloc_tag="ea6b3e973b477b8061e0076bb257dbd7f3faa756" # (2021-02-11) repo_clone_try_double "${primary_urls[jemalloc]}" "${secondary_urls[jemalloc]}" "jemalloc" "$jemalloc_tag" @@ -248,4 +265,27 @@ popd # librdkafka librdkafka_tag="v1.7.0" # (2021-05-06) -repo_clone_try_double "${primary_urls[librdkafka]}" "${secondary_urls[librdkafka]}" "librdkafka" "$librdkafka_tag" +repo_clone_try_double "${primary_urls[librdkafka]}" "${secondary_urls[librdkafka]}" "librdkafka" "$librdkafka_tag" true + +# protobuf +protobuf_tag="v3.12.4" +repo_clone_try_double "${primary_urls[protobuf]}" "${secondary_urls[protobuf]}" "protobuf" "$protobuf_tag" true +pushd protobuf +./autogen.sh && ./configure --prefix=$(pwd)/lib +popd + +# boost +file_get_try_double "${primary_urls[boost]}" "${secondary_urls[boost]}" +tar -xzf boost_1_77_0.tar.gz +mv boost_1_77_0 boost +pushd boost +./bootstrap.sh --prefix=$(pwd)/lib --with-libraries="system,regex" +./b2 -j$(nproc) install +popd + +#pulsar +pulsar_tag="v2.8.1" +repo_clone_try_double "${primary_urls[pulsar]}" "${secondary_urls[pulsar]}" "pulsar" "$pulsar_tag" true +pushd pulsar +git apply ../pulsar.patch +popd diff --git a/src/integrations/CMakeLists.txt b/src/integrations/CMakeLists.txt index f7f7449ff..969d711cd 100644 --- a/src/integrations/CMakeLists.txt +++ b/src/integrations/CMakeLists.txt @@ -1 +1,2 @@ add_subdirectory(kafka) +add_subdirectory(pulsar) diff --git a/src/integrations/constants.hpp b/src/integrations/constants.hpp new file mode 100644 index 000000000..c1500dba0 --- /dev/null +++ b/src/integrations/constants.hpp @@ -0,0 +1,22 @@ +// Copyright 2021 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#pragma once +#include + +namespace integrations { +constexpr std::chrono::milliseconds kDefaultBatchInterval{100}; +constexpr int64_t kDefaultBatchSize{1000}; +constexpr int64_t kDefaultCheckBatchLimit{1}; +constexpr std::chrono::milliseconds kDefaultCheckTimeout{30000}; +constexpr std::chrono::milliseconds kMinimumInterval{1}; +constexpr int64_t kMinimumSize{1}; +} // namespace integrations diff --git a/src/integrations/kafka/consumer.cpp b/src/integrations/kafka/consumer.cpp index 0ec8f0e94..b7a287450 100644 --- a/src/integrations/kafka/consumer.cpp +++ b/src/integrations/kafka/consumer.cpp @@ -19,6 +19,8 @@ #include #include + +#include "integrations/constants.hpp" #include "integrations/kafka/exceptions.hpp" #include "utils/exceptions.hpp" #include "utils/logging.hpp" @@ -27,13 +29,6 @@ namespace integrations::kafka { -constexpr std::chrono::milliseconds kDefaultBatchInterval{100}; -constexpr int64_t kDefaultBatchSize = 1000; -constexpr int64_t kDefaultCheckBatchLimit = 1; -constexpr std::chrono::milliseconds kDefaultCheckTimeout{30000}; -constexpr std::chrono::milliseconds kMinimumInterval{1}; -constexpr int64_t kMinimumSize{1}; - namespace { utils::BasicResult> GetBatch(RdKafka::KafkaConsumer &consumer, const ConsumerInfo &info, @@ -114,8 +109,9 @@ int64_t Message::Offset() const { return c_message->offset; } -Consumer::Consumer(const std::string &bootstrap_servers, ConsumerInfo info, ConsumerFunction consumer_function) +Consumer::Consumer(ConsumerInfo info, ConsumerFunction consumer_function) : info_{std::move(info)}, consumer_function_(std::move(consumer_function)), cb_(info_.consumer_name) { + MG_ASSERT(consumer_function_, "Empty consumer function for Kafka consumer"); // NOLINTNEXTLINE (modernize-use-nullptr) if (info_.batch_interval.value_or(kMinimumInterval) < kMinimumInterval) { @@ -148,7 +144,7 @@ Consumer::Consumer(const std::string &bootstrap_servers, ConsumerInfo info, Cons throw ConsumerFailedToInitializeException(info_.consumer_name, error); } - if (conf->set("bootstrap.servers", bootstrap_servers, error) != RdKafka::Conf::CONF_OK) { + if (conf->set("bootstrap.servers", info_.bootstrap_servers, error) != RdKafka::Conf::CONF_OK) { throw ConsumerFailedToInitializeException(info_.consumer_name, error); } diff --git a/src/integrations/kafka/consumer.hpp b/src/integrations/kafka/consumer.hpp index fc4e30061..630963a87 100644 --- a/src/integrations/kafka/consumer.hpp +++ b/src/integrations/kafka/consumer.hpp @@ -83,6 +83,7 @@ struct ConsumerInfo { std::string consumer_name; std::vector topics; std::string consumer_group; + std::string bootstrap_servers; std::optional batch_interval; std::optional batch_size; }; @@ -97,7 +98,7 @@ class Consumer final : public RdKafka::EventCb { /// /// @throws ConsumerFailedToInitializeException if the consumer can't connect /// to the Kafka endpoint. - Consumer(const std::string &bootstrap_servers, ConsumerInfo info, ConsumerFunction consumer_function); + Consumer(ConsumerInfo info, ConsumerFunction consumer_function); ~Consumer() override; Consumer(const Consumer &other) = delete; diff --git a/src/integrations/pulsar/CMakeLists.txt b/src/integrations/pulsar/CMakeLists.txt new file mode 100644 index 000000000..20376fd7f --- /dev/null +++ b/src/integrations/pulsar/CMakeLists.txt @@ -0,0 +1,8 @@ +set(integrations_pulsar_src_files + consumer.cpp +) + +find_package(CURL REQUIRED) + +add_library(mg-integrations-pulsar STATIC ${integrations_pulsar_src_files}) +target_link_libraries(mg-integrations-pulsar mg-utils pulsar Threads::Threads ${CURL_LIBRARIES}) diff --git a/src/integrations/pulsar/consumer.cpp b/src/integrations/pulsar/consumer.cpp new file mode 100644 index 000000000..224aca9f8 --- /dev/null +++ b/src/integrations/pulsar/consumer.cpp @@ -0,0 +1,292 @@ +// Copyright 2021 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#include "integrations/pulsar/consumer.hpp" + +#include +#include +#include + +#include +#include + +#include "integrations/constants.hpp" +#include "integrations/pulsar/exceptions.hpp" +#include "utils/concepts.hpp" +#include "utils/logging.hpp" +#include "utils/on_scope_exit.hpp" +#include "utils/result.hpp" +#include "utils/thread.hpp" + +namespace integrations::pulsar { + +namespace { + +template +concept PulsarConsumer = utils::SameAsAnyOf; + +pulsar_client::Result ConsumeMessage(pulsar_client::Consumer &consumer, pulsar_client::Message &message, + int remaining_timeout_in_ms) { + return consumer.receive(message, remaining_timeout_in_ms); +} + +pulsar_client::Result ConsumeMessage(pulsar_client::Reader &reader, pulsar_client::Message &message, + int remaining_timeout_in_ms) { + return reader.readNext(message, remaining_timeout_in_ms); +} + +template +utils::BasicResult> GetBatch(TConsumer &consumer, const ConsumerInfo &info, + std::atomic &is_running) { + std::vector batch{}; + + const auto batch_size = info.batch_size.value_or(kDefaultBatchSize); + batch.reserve(batch_size); + + auto remaining_timeout_in_ms = info.batch_interval.value_or(kDefaultBatchInterval).count(); + auto start = std::chrono::steady_clock::now(); + + for (int64_t i = 0; remaining_timeout_in_ms > 0 && i < batch_size && is_running.load(); ++i) { + pulsar_client::Message message; + const auto result = ConsumeMessage(consumer, message, remaining_timeout_in_ms); + switch (result) { + case pulsar_client::Result::ResultTimeout: + return std::move(batch); + case pulsar_client::Result::ResultOk: + batch.emplace_back(Message{std::move(message)}); + break; + default: + spdlog::warn(fmt::format("Unexpected error while consuming message from consumer {}, error: {}", + info.consumer_name, result)); + return {pulsar_client::strResult(result)}; + } + + auto now = std::chrono::steady_clock::now(); + auto took = std::chrono::duration_cast(now - start); + remaining_timeout_in_ms = remaining_timeout_in_ms - took.count(); + start = now; + } + + return std::move(batch); +} + +class SpdlogLogger : public pulsar_client::Logger { + bool isEnabled(Level /*level*/) override { return spdlog::should_log(spdlog::level::trace); } + + void log(Level /*level*/, int /*line*/, const std::string &message) override { + spdlog::trace("[Pulsar] {}", message); + } +}; + +class SpdlogLoggerFactory : public pulsar_client::LoggerFactory { + pulsar_client::Logger *getLogger(const std::string & /*file_name*/) override { + if (!logger_) { + logger_ = std::make_unique(); + } + return logger_.get(); + } + + private: + std::unique_ptr logger_; +}; + +pulsar_client::Client CreateClient(const std::string &service_url) { + static SpdlogLoggerFactory logger_factory; + pulsar_client::ClientConfiguration conf; + conf.setLogger(&logger_factory); + return {service_url, conf}; +} +} // namespace + +Message::Message(pulsar_client::Message &&message) : message_{std::move(message)} {} + +std::span Message::Payload() const { + return {static_cast(message_.getData()), message_.getLength()}; +} + +Consumer::Consumer(ConsumerInfo info, ConsumerFunction consumer_function) + : info_{std::move(info)}, + client_{CreateClient(info_.service_url)}, + consumer_function_{std::move(consumer_function)} { + pulsar_client::ConsumerConfiguration config; + config.setSubscriptionInitialPosition(pulsar_client::InitialPositionLatest); + config.setConsumerType(pulsar_client::ConsumerType::ConsumerExclusive); + if (pulsar_client::Result result = client_.subscribe(info_.topics, info_.consumer_name, config, consumer_); + result != pulsar_client::ResultOk) { + throw ConsumerFailedToInitializeException(info_.consumer_name, pulsar_client::strResult(result)); + } +} +Consumer::~Consumer() { + StopIfRunning(); + consumer_.close(); + client_.close(); +} + +bool Consumer::IsRunning() const { return is_running_; } + +const ConsumerInfo &Consumer::Info() const { return info_; } + +void Consumer::Start() { + if (is_running_) { + throw ConsumerRunningException(info_.consumer_name); + } + + StartConsuming(); +} + +void Consumer::Stop() { + if (!is_running_) { + throw ConsumerStoppedException(info_.consumer_name); + } + StopConsuming(); +} + +void Consumer::StopIfRunning() { + if (is_running_) { + StopConsuming(); + } + + if (thread_.joinable()) { + thread_.join(); + } +} + +void Consumer::Check(std::optional timeout, std::optional limit_batches, + const ConsumerFunction &check_consumer_function) const { + // NOLINTNEXTLINE (modernize-use-nullptr) + if (timeout.value_or(kMinimumInterval) < kMinimumInterval) { + throw ConsumerCheckFailedException(info_.consumer_name, "Timeout has to be positive!"); + } + if (limit_batches.value_or(kMinimumSize) < kMinimumSize) { + throw ConsumerCheckFailedException(info_.consumer_name, "Batch limit has to be positive!"); + } + // The implementation of this function is questionable: it is const qualified, though it changes the inner state of + // PulsarConsumer. Though it changes the inner state, it saves the current assignment for future Check/Start calls to + // restore the current state, so the changes made by this function shouldn't be visible for the users of the class. It + // also passes a non const reference of PulsarConsumer to GetBatch function. That means the object is bitwise const + // (PulsarConsumer is stored in unique_ptr) and internally mostly synchronized. Mostly, because as Start/Stop requires + // exclusive access to consumer, so we don't have to deal with simultaneous calls to those functions. The only concern + // in this function is to prevent executing this function on multiple threads simultaneously. + if (is_running_.exchange(true)) { + throw ConsumerRunningException(info_.consumer_name); + } + + utils::OnScopeExit restore_is_running([this] { is_running_.store(false); }); + + const auto num_of_batches = limit_batches.value_or(kDefaultCheckBatchLimit); + const auto timeout_to_use = timeout.value_or(kDefaultCheckTimeout); + const auto start = std::chrono::steady_clock::now(); + + if (info_.topics.size() != 1) { + throw ConsumerCheckFailedException(info_.consumer_name, "Check cannot be used for consumers with multiple topics."); + } + + std::vector partitions; + const auto &topic = info_.topics.front(); + client_.getPartitionsForTopic(topic, partitions); + if (partitions.size() > 1) { + throw ConsumerCheckFailedException(info_.consumer_name, "Check cannot be used for topics with multiple partitions"); + } + pulsar_client::Reader reader; + client_.createReader(topic, last_message_id_, {}, reader); + for (int64_t i = 0; i < num_of_batches;) { + const auto now = std::chrono::steady_clock::now(); + // NOLINTNEXTLINE (modernize-use-nullptr) + if (now - start >= timeout_to_use) { + throw ConsumerCheckFailedException(info_.consumer_name, "Timeout reached"); + } + + auto maybe_batch = GetBatch(reader, info_, is_running_); + + if (maybe_batch.HasError()) { + throw ConsumerCheckFailedException(info_.consumer_name, maybe_batch.GetError()); + } + + const auto &batch = maybe_batch.GetValue(); + + if (batch.empty()) { + continue; + } + ++i; + + try { + check_consumer_function(batch); + } catch (const std::exception &e) { + spdlog::warn("Pulsar consumer {} check failed with error {}", info_.consumer_name, e.what()); + throw ConsumerCheckFailedException(info_.consumer_name, e.what()); + } + } + reader.close(); +} + +void Consumer::StartConsuming() { + MG_ASSERT(!is_running_, "Cannot start already running consumer!"); + if (thread_.joinable()) { + thread_.join(); + } + + is_running_.store(true); + + thread_ = std::thread([this] { + constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize(); + const auto full_thread_name = "Cons#" + info_.consumer_name; + + utils::ThreadSetName(full_thread_name.substr(0, kMaxThreadNameSize)); + + while (is_running_) { + auto maybe_batch = GetBatch(consumer_, info_, is_running_); + + if (maybe_batch.HasError()) { + spdlog::warn("Error happened in consumer {} while fetching messages: {}!", info_.consumer_name, + maybe_batch.GetError()); + break; + } + + const auto &batch = maybe_batch.GetValue(); + + if (batch.empty()) { + continue; + } + + spdlog::info("Pulsar consumer {} is processing a batch", info_.consumer_name); + + try { + consumer_function_(batch); + + if (std::any_of(batch.begin(), batch.end(), [&](const auto &message) { + if (const auto result = consumer_.acknowledge(message.message_); result != pulsar_client::ResultOk) { + spdlog::warn("Acknowledging a message of consumer {} failed: {}", info_.consumer_name, result); + return true; + } + last_message_id_ = message.message_.getMessageId(); + return false; + })) { + break; + } + } catch (const std::exception &e) { + spdlog::warn("Error happened in consumer {} while processing a batch: {}!", info_.consumer_name, e.what()); + break; + } + + spdlog::info("Pulsar consumer {} finished processing", info_.consumer_name); + } + is_running_.store(false); + }); +} + +void Consumer::StopConsuming() { + is_running_.store(false); + if (thread_.joinable()) { + thread_.join(); + } +} + +} // namespace integrations::pulsar diff --git a/src/integrations/pulsar/consumer.hpp b/src/integrations/pulsar/consumer.hpp new file mode 100644 index 000000000..d291af88e --- /dev/null +++ b/src/integrations/pulsar/consumer.hpp @@ -0,0 +1,81 @@ +// Copyright 2021 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#pragma once +#include +#include +#include +#include + +#include + +namespace integrations::pulsar { + +namespace pulsar_client = ::pulsar; + +class Consumer; + +class Message final { + public: + explicit Message(pulsar_client::Message &&message); + + std::span Payload() const; + + private: + pulsar_client::Message message_; + + friend Consumer; +}; + +using ConsumerFunction = std::function &)>; + +struct ConsumerInfo { + std::optional batch_size; + std::optional batch_interval; + std::vector topics; + std::string consumer_name; + std::string service_url; +}; + +class Consumer final { + public: + Consumer(ConsumerInfo info, ConsumerFunction consumer_function); + ~Consumer(); + + Consumer(const Consumer &) = delete; + Consumer(Consumer &&) noexcept = delete; + Consumer &operator=(const Consumer &) = delete; + Consumer &operator=(Consumer &&) = delete; + + bool IsRunning() const; + void Start(); + void Stop(); + void StopIfRunning(); + + void Check(std::optional timeout, std::optional limit_batches, + const ConsumerFunction &check_consumer_function) const; + + const ConsumerInfo &Info() const; + + private: + void StartConsuming(); + void StopConsuming(); + + ConsumerInfo info_; + mutable pulsar_client::Client client_; + pulsar_client::Consumer consumer_; + ConsumerFunction consumer_function_; + + mutable std::atomic is_running_{false}; + pulsar_client::MessageId last_message_id_{pulsar_client::MessageId::earliest()}; + std::thread thread_; +}; +} // namespace integrations::pulsar diff --git a/src/integrations/pulsar/exceptions.hpp b/src/integrations/pulsar/exceptions.hpp new file mode 100644 index 000000000..dbe1a7212 --- /dev/null +++ b/src/integrations/pulsar/exceptions.hpp @@ -0,0 +1,58 @@ +// Copyright 2021 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#pragma once + +#include + +#include "utils/exceptions.hpp" + +namespace integrations::pulsar { +class PulsarStreamException : public utils::BasicException { + using utils::BasicException::BasicException; +}; + +class ConsumerFailedToInitializeException : public PulsarStreamException { + public: + ConsumerFailedToInitializeException(const std::string &consumer_name, const std::string &error) + : PulsarStreamException("Failed to initialize Pulsar consumer {} : {}", consumer_name, error) {} +}; + +class ConsumerRunningException : public PulsarStreamException { + public: + explicit ConsumerRunningException(const std::string &consumer_name) + : PulsarStreamException("Pulsar consumer {} is already running", consumer_name) {} +}; + +class ConsumerStoppedException : public PulsarStreamException { + public: + explicit ConsumerStoppedException(const std::string &consumer_name) + : PulsarStreamException("Pulsar consumer {} is already stopped", consumer_name) {} +}; + +class ConsumerCheckFailedException : public PulsarStreamException { + public: + explicit ConsumerCheckFailedException(const std::string &consumer_name, const std::string &error) + : PulsarStreamException("Pulsar consumer {} check failed: {}", consumer_name, error) {} +}; + +class ConsumerStartFailedException : public PulsarStreamException { + public: + explicit ConsumerStartFailedException(const std::string &consumer_name, const std::string &error) + : PulsarStreamException("Starting Pulsar consumer {} failed: {}", consumer_name, error) {} +}; + +class TopicNotFoundException : public PulsarStreamException { + public: + TopicNotFoundException(const std::string &consumer_name, const std::string &topic_name) + : PulsarStreamException("Pulsar consumer {} cannot find topic {}", consumer_name, topic_name) {} +}; +} // namespace integrations::pulsar diff --git a/src/memgraph.cpp b/src/memgraph.cpp index 2f2567d47..e4c9ce120 100644 --- a/src/memgraph.cpp +++ b/src/memgraph.cpp @@ -1119,11 +1119,11 @@ int main(int argc, char **argv) { } storage::Storage db(db_config); - query::InterpreterContext interpreter_context{ - &db, - {.query = {.allow_load_csv = FLAGS_allow_load_csv}, .execution_timeout_sec = FLAGS_query_execution_timeout_sec}, - FLAGS_data_directory, - FLAGS_kafka_bootstrap_servers}; + query::InterpreterContext interpreter_context{&db, + {.query = {.allow_load_csv = FLAGS_allow_load_csv}, + .execution_timeout_sec = FLAGS_query_execution_timeout_sec, + .default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers}, + FLAGS_data_directory}; #ifdef MG_ENTERPRISE SessionData session_data{&db, &interpreter_context, &auth, &audit_log}; #else @@ -1133,9 +1133,6 @@ int main(int argc, char **argv) { query::procedure::gModuleRegistry.SetModulesDirectory(query_modules_directories); query::procedure::gModuleRegistry.UnloadAndLoadModulesFromDirectories(); - // As the Stream transformations are using modules, they have to be restored after the query modules are loaded. - interpreter_context.streams.RestoreStreams(); - AuthQueryHandler auth_handler(&auth, FLAGS_auth_user_or_role_name_regex); AuthChecker auth_checker{&auth}; interpreter_context.auth = &auth_handler; @@ -1151,6 +1148,9 @@ int main(int argc, char **argv) { interpreter_context.auth_checker); } + // As the Stream transformations are using modules, they have to be restored after the query modules are loaded. + interpreter_context.streams.RestoreStreams(); + ServerContext context; std::string service_name = "Bolt"; if (!FLAGS_bolt_key_file.empty() && !FLAGS_bolt_cert_file.empty()) { diff --git a/src/memory/new_delete.cpp b/src/memory/new_delete.cpp index 4636557d5..4264c29dd 100644 --- a/src/memory/new_delete.cpp +++ b/src/memory/new_delete.cpp @@ -19,13 +19,12 @@ #include #endif -#include "utils/likely.hpp" #include "utils/memory_tracker.hpp" namespace { void *newImpl(const std::size_t size) { auto *ptr = malloc(size); - if (LIKELY(ptr != nullptr)) { + if (ptr != nullptr) [[likely]] { return ptr; } @@ -34,7 +33,7 @@ void *newImpl(const std::size_t size) { void *newImpl(const std::size_t size, const std::align_val_t align) { auto *ptr = aligned_alloc(static_cast(align), size); - if (LIKELY(ptr != nullptr)) { + if (ptr != nullptr) [[likely]] { return ptr; } @@ -47,14 +46,22 @@ void *newNoExcept(const std::size_t size, const std::align_val_t align) noexcept } #if USE_JEMALLOC -void deleteImpl(void *ptr) noexcept { dallocx(ptr, 0); } +void deleteImpl(void *ptr) noexcept { + if (ptr == nullptr) [[unlikely]] { + return; + } + dallocx(ptr, 0); +} void deleteImpl(void *ptr, const std::align_val_t align) noexcept { + if (ptr == nullptr) [[unlikely]] { + return; + } dallocx(ptr, MALLOCX_ALIGN(align)); // NOLINT(hicpp-signed-bitwise) } void deleteSized(void *ptr, const std::size_t size) noexcept { - if (UNLIKELY(ptr == nullptr)) { + if (ptr == nullptr) [[unlikely]] { return; } @@ -62,7 +69,7 @@ void deleteSized(void *ptr, const std::size_t size) noexcept { } void deleteSized(void *ptr, const std::size_t size, const std::align_val_t align) noexcept { - if (UNLIKELY(ptr == nullptr)) { + if (ptr == nullptr) [[unlikely]] { return; } @@ -81,7 +88,7 @@ void deleteSized(void *ptr, const std::size_t /*unused*/, const std::align_val_t void TrackMemory(std::size_t size) { #if USE_JEMALLOC - if (LIKELY(size != 0)) { + if (size != 0) [[likely]] { size = nallocx(size, 0); } #endif @@ -90,7 +97,7 @@ void TrackMemory(std::size_t size) { void TrackMemory(std::size_t size, const std::align_val_t align) { #if USE_JEMALLOC - if (LIKELY(size != 0)) { + if (size != 0) [[likely]] { size = nallocx(size, MALLOCX_ALIGN(align)); // NOLINT(hicpp-signed-bitwise) } #endif @@ -120,7 +127,7 @@ bool TrackMemoryNoExcept(const std::size_t size, const std::align_val_t align) { void UntrackMemory([[maybe_unused]] void *ptr, [[maybe_unused]] std::size_t size = 0) noexcept { try { #if USE_JEMALLOC - if (LIKELY(ptr != nullptr)) { + if (ptr != nullptr) [[likely]] { utils::total_memory_tracker.Free(sallocx(ptr, 0)); } #else @@ -138,7 +145,7 @@ void UntrackMemory([[maybe_unused]] void *ptr, [[maybe_unused]] std::size_t size void UntrackMemory(void *ptr, const std::align_val_t align, [[maybe_unused]] std::size_t size = 0) noexcept { try { #if USE_JEMALLOC - if (LIKELY(ptr != nullptr)) { + if (ptr != nullptr) [[likely]] { utils::total_memory_tracker.Free(sallocx(ptr, MALLOCX_ALIGN(align))); // NOLINT(hicpp-signed-bitwise) } #else @@ -176,28 +183,28 @@ void *operator new[](const std::size_t size, const std::align_val_t align) { } void *operator new(const std::size_t size, const std::nothrow_t & /*unused*/) noexcept { - if (LIKELY(TrackMemoryNoExcept(size))) { + if (TrackMemoryNoExcept(size)) [[likely]] { return newNoExcept(size); } return nullptr; } void *operator new[](const std::size_t size, const std::nothrow_t & /*unused*/) noexcept { - if (LIKELY(TrackMemoryNoExcept(size))) { + if (TrackMemoryNoExcept(size)) [[likely]] { return newNoExcept(size); } return nullptr; } void *operator new(const std::size_t size, const std::align_val_t align, const std::nothrow_t & /*unused*/) noexcept { - if (LIKELY(TrackMemoryNoExcept(size, align))) { + if (TrackMemoryNoExcept(size, align)) [[likely]] { return newNoExcept(size, align); } return nullptr; } void *operator new[](const std::size_t size, const std::align_val_t align, const std::nothrow_t & /*unused*/) noexcept { - if (LIKELY(TrackMemoryNoExcept(size, align))) { + if (TrackMemoryNoExcept(size, align)) [[likely]] { return newNoExcept(size, align); } return nullptr; diff --git a/src/query/CMakeLists.txt b/src/query/CMakeLists.txt index 3e2e2eade..7e3ee8cd9 100644 --- a/src/query/CMakeLists.txt +++ b/src/query/CMakeLists.txt @@ -44,7 +44,7 @@ add_library(mg-query STATIC ${mg_query_sources}) add_dependencies(mg-query generate_lcp_query) target_include_directories(mg-query PUBLIC ${CMAKE_SOURCE_DIR}/include) target_link_libraries(mg-query dl cppitertools) -target_link_libraries(mg-query mg-integrations-kafka mg-storage-v2 mg-utils mg-kvstore mg-memory) +target_link_libraries(mg-query mg-integrations-pulsar mg-integrations-kafka mg-storage-v2 mg-utils mg-kvstore mg-memory) if("${MG_PYTHON_VERSION}" STREQUAL "") find_package(Python3 3.5 REQUIRED COMPONENTS Development) else() diff --git a/src/query/config.hpp b/src/query/config.hpp index 148b04b28..16a7b5b1c 100644 --- a/src/query/config.hpp +++ b/src/query/config.hpp @@ -10,6 +10,7 @@ // licenses/APL.txt. #pragma once +#include namespace query { struct InterpreterConfig { @@ -19,5 +20,7 @@ struct InterpreterConfig { // The default execution timeout is 10 minutes. double execution_timeout_sec{600.0}; + + std::string default_kafka_bootstrap_servers; }; } // namespace query diff --git a/src/query/interpreter.cpp b/src/query/interpreter.cpp index 1d493ce4b..571c96f08 100644 --- a/src/query/interpreter.cpp +++ b/src/query/interpreter.cpp @@ -560,7 +560,9 @@ Callback HandleStreamQuery(StreamQuery *stream_query, const Parameters ¶mete batch_size = GetOptionalValue(stream_query->batch_size_, evaluator), transformation_name = stream_query->transform_name_, bootstrap_servers = std::move(bootstrap), owner = StringPointerToOptional(username)]() mutable { - std::string bootstrap = bootstrap_servers ? std::move(*bootstrap_servers) : ""; + std::string bootstrap = bootstrap_servers + ? std::move(*bootstrap_servers) + : std::string{interpreter_context->config.default_kafka_bootstrap_servers}; interpreter_context->streams.Create( stream_name, {.common_info = {.batch_interval = batch_interval, @@ -570,6 +572,15 @@ Callback HandleStreamQuery(StreamQuery *stream_query, const Parameters ¶mete .consumer_group = std::move(consumer_group), .bootstrap_servers = std::move(bootstrap)}, std::move(owner)); + // interpreter_context->streams.Create( + // stream_name, + // {.common_info = {.batch_interval = batch_interval, + // .batch_size = batch_size, + // .transformation_name = std::move(transformation_name)}, + // .topics = std::move(topic_names), + // .service_url = std::move(bootstrap)}, + // std::move(owner)); + return std::vector>{}; }; notifications->emplace_back(SeverityLevel::INFO, NotificationCode::CREATE_STREAM, @@ -918,11 +929,8 @@ using RWType = plan::ReadWriteTypeChecker::RWType; } // namespace InterpreterContext::InterpreterContext(storage::Storage *db, const InterpreterConfig config, - const std::filesystem::path &data_directory, std::string kafka_bootstrap_servers) - : db(db), - trigger_store(data_directory / "triggers"), - config(config), - streams{this, std::move(kafka_bootstrap_servers), data_directory / "streams"} {} + const std::filesystem::path &data_directory) + : db(db), trigger_store(data_directory / "triggers"), config(config), streams{this, data_directory / "streams"} {} Interpreter::Interpreter(InterpreterContext *interpreter_context) : interpreter_context_(interpreter_context) { MG_ASSERT(interpreter_context_, "Interpreter context must not be NULL"); diff --git a/src/query/interpreter.hpp b/src/query/interpreter.hpp index 8c3310ccd..95b8249be 100644 --- a/src/query/interpreter.hpp +++ b/src/query/interpreter.hpp @@ -165,7 +165,7 @@ struct PreparedQuery { */ struct InterpreterContext { explicit InterpreterContext(storage::Storage *db, InterpreterConfig config, - const std::filesystem::path &data_directory, std::string kafka_bootstrap_servers); + const std::filesystem::path &data_directory); storage::Storage *db; diff --git a/src/query/procedure/mg_procedure_impl.cpp b/src/query/procedure/mg_procedure_impl.cpp index c603a38c0..06ec8aa59 100644 --- a/src/query/procedure/mg_procedure_impl.cpp +++ b/src/query/procedure/mg_procedure_impl.cpp @@ -2500,6 +2500,8 @@ mgp_error mgp_message_payload(mgp_message *message, const char **result) { using MessageType = std::decay_t; if constexpr (std::same_as) { return msg->Payload().data(); + } else if constexpr (std::same_as) { + return msg.Payload().data(); } else { throw std::invalid_argument("Invalid source type"); } @@ -2517,6 +2519,8 @@ mgp_error mgp_message_payload_size(mgp_message *message, size_t *result) { using MessageType = std::decay_t; if constexpr (std::same_as) { return msg->Payload().size(); + } else if constexpr (std::same_as) { + return msg.Payload().size(); } else { throw std::invalid_argument("Invalid source type"); } diff --git a/src/query/procedure/mg_procedure_impl.hpp b/src/query/procedure/mg_procedure_impl.hpp index c32578c7f..fc7168259 100644 --- a/src/query/procedure/mg_procedure_impl.hpp +++ b/src/query/procedure/mg_procedure_impl.hpp @@ -20,6 +20,7 @@ #include #include "integrations/kafka/consumer.hpp" +#include "integrations/pulsar/consumer.hpp" #include "query/context.hpp" #include "query/db_accessor.hpp" #include "query/procedure/cypher_type_ptr.hpp" @@ -802,9 +803,11 @@ bool IsValidIdentifierName(const char *name); struct mgp_message { explicit mgp_message(const integrations::kafka::Message &message) : msg{&message} {} + explicit mgp_message(const integrations::pulsar::Message &message) : msg{message} {} using KafkaMessage = const integrations::kafka::Message *; - std::variant msg; + using PulsarMessage = integrations::pulsar::Message; + std::variant msg; }; struct mgp_messages { diff --git a/src/query/stream/common.hpp b/src/query/stream/common.hpp index 339f4f1f3..7adc9cdd0 100644 --- a/src/query/stream/common.hpp +++ b/src/query/stream/common.hpp @@ -61,7 +61,7 @@ concept Stream = requires(TStream stream) { requires ConvertableToJson; }; -enum class StreamSourceType : uint8_t { KAFKA }; +enum class StreamSourceType : uint8_t { KAFKA, PULSAR }; template StreamSourceType StreamType(const T & /*stream*/); diff --git a/src/query/stream/sources.cpp b/src/query/stream/sources.cpp index 50ca02996..9beb2e639 100644 --- a/src/query/stream/sources.cpp +++ b/src/query/stream/sources.cpp @@ -20,10 +20,11 @@ KafkaStream::KafkaStream(std::string stream_name, StreamInfo stream_info, .consumer_name = std::move(stream_name), .topics = std::move(stream_info.topics), .consumer_group = std::move(stream_info.consumer_group), + .bootstrap_servers = std::move(stream_info.bootstrap_servers), .batch_interval = stream_info.common_info.batch_interval, .batch_size = stream_info.common_info.batch_size, }; - consumer_.emplace(std::move(stream_info.bootstrap_servers), std::move(consumer_info), std::move(consumer_function)); + consumer_.emplace(std::move(consumer_info), std::move(consumer_function)); }; KafkaStream::StreamInfo KafkaStream::Info(std::string transformation_name) const { @@ -32,7 +33,8 @@ KafkaStream::StreamInfo KafkaStream::Info(std::string transformation_name) const .batch_size = info.batch_size, .transformation_name = std::move(transformation_name)}, .topics = info.topics, - .consumer_group = info.consumer_group}; + .consumer_group = info.consumer_group, + .bootstrap_servers = info.bootstrap_servers}; } void KafkaStream::Start() { consumer_->Start(); } @@ -67,4 +69,49 @@ void from_json(const nlohmann::json &data, KafkaStream::StreamInfo &info) { data.at(kConsumerGroupKey).get_to(info.consumer_group); data.at(kBoostrapServers).get_to(info.bootstrap_servers); } + +PulsarStream::PulsarStream(std::string stream_name, StreamInfo stream_info, + ConsumerFunction consumer_function) { + integrations::pulsar::ConsumerInfo consumer_info{.batch_size = stream_info.common_info.batch_size, + .batch_interval = stream_info.common_info.batch_interval, + .topics = std::move(stream_info.topics), + .consumer_name = std::move(stream_name), + .service_url = std::move(stream_info.service_url)}; + + consumer_.emplace(std::move(consumer_info), std::move(consumer_function)); +}; + +PulsarStream::StreamInfo PulsarStream::Info(std::string transformation_name) const { + const auto &info = consumer_->Info(); + return {{.batch_interval = info.batch_interval, + .batch_size = info.batch_size, + .transformation_name = std::move(transformation_name)}, + .topics = info.topics, + .service_url = info.service_url}; +} + +void PulsarStream::Start() { consumer_->Start(); } +void PulsarStream::Stop() { consumer_->Stop(); } +bool PulsarStream::IsRunning() const { return consumer_->IsRunning(); } + +void PulsarStream::Check(std::optional timeout, std::optional batch_limit, + const ConsumerFunction &consumer_function) const { + consumer_->Check(timeout, batch_limit, consumer_function); +} + +namespace { +const std::string kServiceUrl{"service_url"}; +} // namespace + +void to_json(nlohmann::json &data, PulsarStream::StreamInfo &&info) { + data[kCommonInfoKey] = std::move(info.common_info); + data[kTopicsKey] = std::move(info.topics); + data[kServiceUrl] = std::move(info.service_url); +} + +void from_json(const nlohmann::json &data, PulsarStream::StreamInfo &info) { + data.at(kCommonInfoKey).get_to(info.common_info); + data.at(kTopicsKey).get_to(info.topics); + data.at(kServiceUrl).get_to(info.service_url); +} } // namespace query diff --git a/src/query/stream/sources.hpp b/src/query/stream/sources.hpp index 1f7954d49..afe1bf767 100644 --- a/src/query/stream/sources.hpp +++ b/src/query/stream/sources.hpp @@ -14,6 +14,7 @@ #include "query/stream/common.hpp" #include "integrations/kafka/consumer.hpp" +#include "integrations/pulsar/consumer.hpp" namespace query { @@ -54,4 +55,37 @@ inline StreamSourceType StreamType(const KafkaStream & /*stream*/) { return StreamSourceType::KAFKA; } +struct PulsarStream { + struct StreamInfo { + CommonStreamInfo common_info; + std::vector topics; + std::string service_url; + }; + + using Message = integrations::pulsar::Message; + + PulsarStream(std::string stream_name, StreamInfo stream_info, ConsumerFunction consumer_function); + + StreamInfo Info(std::string transformation_name) const; + + void Start(); + void Stop(); + bool IsRunning() const; + + void Check(std::optional timeout, std::optional batch_limit, + const ConsumerFunction &consumer_function) const; + + private: + using Consumer = integrations::pulsar::Consumer; + std::optional consumer_; +}; + +void to_json(nlohmann::json &data, PulsarStream::StreamInfo &&info); +void from_json(const nlohmann::json &data, PulsarStream::StreamInfo &info); + +template <> +inline StreamSourceType StreamType(const PulsarStream & /*stream*/) { + return StreamSourceType::PULSAR; +} + } // namespace query diff --git a/src/query/stream/streams.cpp b/src/query/stream/streams.cpp index dc72773c1..220a50152 100644 --- a/src/query/stream/streams.cpp +++ b/src/query/stream/streams.cpp @@ -164,11 +164,9 @@ struct Overloaded : Ts... { template Overloaded(Ts...) -> Overloaded; } // namespace -Streams::Streams(InterpreterContext *interpreter_context, std::string bootstrap_servers, - std::filesystem::path directory) - : interpreter_context_(interpreter_context), - bootstrap_servers_(std::move(bootstrap_servers)), - storage_(std::move(directory)) { + +Streams::Streams(InterpreterContext *interpreter_context, std::filesystem::path directory) + : interpreter_context_(interpreter_context), storage_(std::move(directory)) { constexpr std::string_view proc_name = "kafka_set_stream_offset"; auto set_stream_offset = [this, proc_name](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory * /*memory*/) { @@ -221,6 +219,8 @@ void Streams::Create(const std::string &stream_name, typename TStream::StreamInf template void Streams::Create(const std::string &stream_name, KafkaStream::StreamInfo info, std::optional owner); +template void Streams::Create(const std::string &stream_name, PulsarStream::StreamInfo info, + std::optional owner); template Streams::StreamsMap::iterator Streams::CreateConsumer(StreamsMap &map, const std::string &stream_name, @@ -276,10 +276,6 @@ Streams::StreamsMap::iterator Streams::CreateConsumer(StreamsMap &map, const std result.rows.clear(); }; - if (stream_info.bootstrap_servers.empty()) { - stream_info.bootstrap_servers = bootstrap_servers_; - } - auto insert_result = map.try_emplace( stream_name, StreamData{std::move(stream_info.common_info.transformation_name), std::move(owner), std::make_unique>( @@ -313,7 +309,7 @@ void Streams::RestoreStreams() { MG_ASSERT(status.name == stream_name, "Expected stream name is '{}', but got '{}'", status.name, stream_name); try { - auto it = CreateConsumer(*locked_streams_map, stream_name, std::move(status.info), {}); + auto it = CreateConsumer(*locked_streams_map, stream_name, std::move(status.info), std::move(status.owner)); if (status.is_running) { std::visit( [&](auto &&stream_data) { @@ -335,6 +331,9 @@ void Streams::RestoreStreams() { case StreamSourceType::KAFKA: create_consumer(StreamStatus{}, std::move(stream_json_data)); break; + case StreamSourceType::PULSAR: + create_consumer(StreamStatus{}, std::move(stream_json_data)); + break; } } } @@ -474,6 +473,4 @@ TransformationResult Streams::Check(const std::string &stream_name, std::optiona it->second); } -std::string_view Streams::BootstrapServers() const { return bootstrap_servers_; } - } // namespace query diff --git a/src/query/stream/streams.hpp b/src/query/stream/streams.hpp index f46bfd927..c797d016b 100644 --- a/src/query/stream/streams.hpp +++ b/src/query/stream/streams.hpp @@ -75,9 +75,8 @@ class Streams final { /// Initializes the streams. /// /// @param interpreter_context context to use to run the result of transformations - /// @param bootstrap_servers initial list of brokers as a comma separated list of broker host or host:port /// @param directory a directory path to store the persisted streams metadata - Streams(InterpreterContext *interpreter_context, std::string bootstrap_servers, std::filesystem::path directory); + Streams(InterpreterContext *interpreter_context, std::filesystem::path directory); /// Restores the streams from the persisted metadata. /// The restoration is done in a best effort manner, therefore no exception is thrown on failure, but the error is @@ -163,7 +162,7 @@ class Streams final { std::unique_ptr> stream_source; }; - using StreamDataVariant = std::variant>; + using StreamDataVariant = std::variant, StreamData>; using StreamsMap = std::unordered_map; using SynchronizedStreamsMap = utils::Synchronized; @@ -180,7 +179,6 @@ class Streams final { } InterpreterContext *interpreter_context_; - std::string bootstrap_servers_; kvstore::KVStore storage_; SynchronizedStreamsMap streams_; diff --git a/tests/benchmark/expansion.cpp b/tests/benchmark/expansion.cpp index 1a800d678..f35b99640 100644 --- a/tests/benchmark/expansion.cpp +++ b/tests/benchmark/expansion.cpp @@ -48,7 +48,7 @@ class ExpansionBenchFixture : public benchmark::Fixture { MG_ASSERT(db->CreateIndex(label)); - interpreter_context.emplace(&*db, query::InterpreterConfig{}, data_directory, "non existing bootstrap servers"); + interpreter_context.emplace(&*db, query::InterpreterConfig{}, data_directory); interpreter.emplace(&*interpreter_context); } diff --git a/tests/manual/single_query.cpp b/tests/manual/single_query.cpp index 3c96b3712..eab474d07 100644 --- a/tests/manual/single_query.cpp +++ b/tests/manual/single_query.cpp @@ -31,8 +31,7 @@ int main(int argc, char *argv[]) { utils::OnScopeExit([&data_directory] { std::filesystem::remove_all(data_directory); }); utils::license::global_license_checker.EnableTesting(); - query::InterpreterContext interpreter_context{&db, query::InterpreterConfig{}, data_directory, - "non existing bootstrap servers"}; + query::InterpreterContext interpreter_context{&db, query::InterpreterConfig{}, data_directory}; query::Interpreter interpreter{&interpreter_context}; ResultStreamFaker stream(&db); diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index 5c5c13634..940b6cceb 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -53,7 +53,7 @@ target_link_libraries(storage_test_utils mg-storage-v2) # Test integrations-kafka add_library(kafka-mock STATIC kafka_mock.cpp) -target_link_libraries(kafka-mock mg-utils librdkafka++ librdkafka Threads::Threads zlib gtest) +target_link_libraries(kafka-mock mg-utils librdkafka++ librdkafka Threads::Threads gtest) # Include directories are intentionally not set, because kafka-mock isn't meant to be used apart from unit tests add_unit_test(integrations_kafka_consumer.cpp kafka_mock.cpp) diff --git a/tests/unit/integrations_kafka_consumer.cpp b/tests/unit/integrations_kafka_consumer.cpp index 89f4593ff..639f7b93c 100644 --- a/tests/unit/integrations_kafka_consumer.cpp +++ b/tests/unit/integrations_kafka_consumer.cpp @@ -51,6 +51,7 @@ struct ConsumerTest : public ::testing::Test { .consumer_name = "Consumer" + test_name, .topics = {kTopicName}, .consumer_group = "ConsumerGroup " + test_name, + .bootstrap_servers = cluster.Bootstraps(), .batch_interval = std::nullopt, .batch_size = std::nullopt, }; @@ -76,8 +77,7 @@ struct ConsumerTest : public ::testing::Test { } }; - auto consumer = - std::make_unique(cluster.Bootstraps(), std::move(info), std::move(consumer_function_wrapper)); + auto consumer = std::make_unique(std::move(info), std::move(consumer_function_wrapper)); int sent_messages{1}; SeedTopicWithInt(kTopicName, sent_messages); @@ -169,7 +169,7 @@ TEST_F(ConsumerTest, BatchInterval) { } TEST_F(ConsumerTest, StartStop) { - Consumer consumer{cluster.Bootstraps(), CreateDefaultConsumerInfo(), kDummyConsumerFunction}; + Consumer consumer{CreateDefaultConsumerInfo(), kDummyConsumerFunction}; auto start = [&consumer](const bool use_conditional) { if (use_conditional) { @@ -279,41 +279,41 @@ TEST_F(ConsumerTest, BatchSize) { TEST_F(ConsumerTest, InvalidBootstrapServers) { auto info = CreateDefaultConsumerInfo(); + info.bootstrap_servers = "non.existing.host:9092"; - EXPECT_THROW(Consumer("non.existing.host:9092", std::move(info), kDummyConsumerFunction), - ConsumerFailedToInitializeException); + EXPECT_THROW(Consumer(std::move(info), kDummyConsumerFunction), ConsumerFailedToInitializeException); } TEST_F(ConsumerTest, InvalidTopic) { auto info = CreateDefaultConsumerInfo(); info.topics = {"Non existing topic"}; - EXPECT_THROW(Consumer(cluster.Bootstraps(), std::move(info), kDummyConsumerFunction), TopicNotFoundException); + EXPECT_THROW(Consumer(std::move(info), kDummyConsumerFunction), TopicNotFoundException); } TEST_F(ConsumerTest, InvalidBatchInterval) { auto info = CreateDefaultConsumerInfo(); info.batch_interval = std::chrono::milliseconds{0}; - EXPECT_THROW(Consumer(cluster.Bootstraps(), info, kDummyConsumerFunction), ConsumerFailedToInitializeException); + EXPECT_THROW(Consumer(info, kDummyConsumerFunction), ConsumerFailedToInitializeException); info.batch_interval = std::chrono::milliseconds{-1}; - EXPECT_THROW(Consumer(cluster.Bootstraps(), info, kDummyConsumerFunction), ConsumerFailedToInitializeException); + EXPECT_THROW(Consumer(info, kDummyConsumerFunction), ConsumerFailedToInitializeException); info.batch_interval = std::chrono::milliseconds{1}; - EXPECT_NO_THROW(Consumer(cluster.Bootstraps(), info, kDummyConsumerFunction)); + EXPECT_NO_THROW(Consumer(info, kDummyConsumerFunction)); } TEST_F(ConsumerTest, InvalidBatchSize) { auto info = CreateDefaultConsumerInfo(); info.batch_size = 0; - EXPECT_THROW(Consumer(cluster.Bootstraps(), info, kDummyConsumerFunction), ConsumerFailedToInitializeException); + EXPECT_THROW(Consumer(info, kDummyConsumerFunction), ConsumerFailedToInitializeException); info.batch_size = -1; - EXPECT_THROW(Consumer(cluster.Bootstraps(), info, kDummyConsumerFunction), ConsumerFailedToInitializeException); + EXPECT_THROW(Consumer(info, kDummyConsumerFunction), ConsumerFailedToInitializeException); info.batch_size = 1; - EXPECT_NO_THROW(Consumer(cluster.Bootstraps(), info, kDummyConsumerFunction)); + EXPECT_NO_THROW(Consumer(info, kDummyConsumerFunction)); } TEST_F(ConsumerTest, DISABLED_StartsFromPreviousOffset) { @@ -348,7 +348,7 @@ TEST_F(ConsumerTest, DISABLED_StartsFromPreviousOffset) { std::string_view{kMessagePrefix + std::to_string(received_message_count + sent_messages)}); } auto expected_total_messages = received_message_count + batch_count; - auto consumer = std::make_unique(cluster.Bootstraps(), ConsumerInfo{info}, consumer_function); + auto consumer = std::make_unique(ConsumerInfo{info}, consumer_function); ASSERT_FALSE(consumer->IsRunning()); consumer->Start(); const auto start = std::chrono::steady_clock::now(); @@ -419,7 +419,7 @@ TEST_F(ConsumerTest, CheckMethodWorks) { } TEST_F(ConsumerTest, CheckMethodTimeout) { - Consumer consumer{cluster.Bootstraps(), CreateDefaultConsumerInfo(), kDummyConsumerFunction}; + Consumer consumer{CreateDefaultConsumerInfo(), kDummyConsumerFunction}; std::chrono::milliseconds timeout{3000}; @@ -433,7 +433,7 @@ TEST_F(ConsumerTest, CheckMethodTimeout) { } TEST_F(ConsumerTest, CheckWithInvalidTimeout) { - Consumer consumer{cluster.Bootstraps(), CreateDefaultConsumerInfo(), kDummyConsumerFunction}; + Consumer consumer{CreateDefaultConsumerInfo(), kDummyConsumerFunction}; const auto start = std::chrono::steady_clock::now(); EXPECT_THROW(consumer.Check(std::chrono::milliseconds{0}, std::nullopt, kDummyConsumerFunction), @@ -448,7 +448,7 @@ TEST_F(ConsumerTest, CheckWithInvalidTimeout) { } TEST_F(ConsumerTest, CheckWithInvalidBatchSize) { - Consumer consumer{cluster.Bootstraps(), CreateDefaultConsumerInfo(), kDummyConsumerFunction}; + Consumer consumer{CreateDefaultConsumerInfo(), kDummyConsumerFunction}; const auto start = std::chrono::steady_clock::now(); EXPECT_THROW(consumer.Check(std::nullopt, 0, kDummyConsumerFunction), ConsumerCheckFailedException); @@ -482,9 +482,9 @@ TEST_F(ConsumerTest, ConsumerStatus) { EXPECT_EQ(topics[1], info.topics[1]); }; - Consumer consumer{cluster.Bootstraps(), - ConsumerInfo{kConsumerName, topics, kConsumerGroupName, kBatchInterval, kBatchSize}, - kDummyConsumerFunction}; + Consumer consumer{ + ConsumerInfo{kConsumerName, topics, kConsumerGroupName, cluster.Bootstraps(), kBatchInterval, kBatchSize}, + kDummyConsumerFunction}; check_info(consumer.Info()); consumer.Start(); diff --git a/tests/unit/interpreter.cpp b/tests/unit/interpreter.cpp index c32596334..a1889b81a 100644 --- a/tests/unit/interpreter.cpp +++ b/tests/unit/interpreter.cpp @@ -54,8 +54,7 @@ auto ToEdgeList(const communication::bolt::Value &v) { struct InterpreterFaker { InterpreterFaker(storage::Storage *db, const query::InterpreterConfig config, const std::filesystem::path &data_directory) - : interpreter_context(db, config, data_directory, "not used bootstrap servers"), - interpreter(&interpreter_context) { + : interpreter_context(db, config, data_directory), interpreter(&interpreter_context) { interpreter_context.auth_checker = &auth_checker; } diff --git a/tests/unit/query_dump.cpp b/tests/unit/query_dump.cpp index 45aad28c0..5040e51ee 100644 --- a/tests/unit/query_dump.cpp +++ b/tests/unit/query_dump.cpp @@ -203,7 +203,7 @@ DatabaseState GetState(storage::Storage *db) { auto Execute(storage::Storage *db, const std::string &query) { auto data_directory = std::filesystem::temp_directory_path() / "MG_tests_unit_query_dump"; - query::InterpreterContext context(db, query::InterpreterConfig{}, data_directory, "non existing bootstrap servers"); + query::InterpreterContext context(db, query::InterpreterConfig{}, data_directory); query::Interpreter interpreter(&context); ResultStreamFaker stream(db); @@ -746,9 +746,7 @@ TEST(DumpTest, ExecuteDumpDatabase) { class StatefulInterpreter { public: explicit StatefulInterpreter(storage::Storage *db) - : db_(db), - context_(db_, query::InterpreterConfig{}, data_directory_, "non existing bootstrap servers"), - interpreter_(&context_) {} + : db_(db), context_(db_, query::InterpreterConfig{}, data_directory_), interpreter_(&context_) {} auto Execute(const std::string &query) { ResultStreamFaker stream(db_); diff --git a/tests/unit/query_plan_edge_cases.cpp b/tests/unit/query_plan_edge_cases.cpp index 8858c8721..0e3c8e348 100644 --- a/tests/unit/query_plan_edge_cases.cpp +++ b/tests/unit/query_plan_edge_cases.cpp @@ -35,7 +35,7 @@ class QueryExecution : public testing::Test { void SetUp() { db_.emplace(); - interpreter_context_.emplace(&*db_, query::InterpreterConfig{}, data_directory, "non existing bootstrap servers"); + interpreter_context_.emplace(&*db_, query::InterpreterConfig{}, data_directory); interpreter_.emplace(&*interpreter_context_); } diff --git a/tests/unit/query_streams.cpp b/tests/unit/query_streams.cpp index 0fe2fbbb9..817b7948c 100644 --- a/tests/unit/query_streams.cpp +++ b/tests/unit/query_streams.cpp @@ -39,21 +39,6 @@ std::string GetDefaultStreamName() { return std::string{::testing::UnitTest::GetInstance()->current_test_info()->name()}; } -StreamInfo CreateDefaultStreamInfo() { - return StreamInfo{.common_info{ - .batch_interval = std::nullopt, - .batch_size = std::nullopt, - .transformation_name = "not used in the tests", - }, - .topics = {kTopicName}, - .consumer_group = "ConsumerGroup " + GetDefaultStreamName(), - .bootstrap_servers = ""}; -} - -StreamCheckData CreateDefaultStreamCheckData() { - return {GetDefaultStreamName(), CreateDefaultStreamInfo(), false, std::nullopt}; -} - std::filesystem::path GetCleanDataDirectory() { const auto path = std::filesystem::temp_directory_path() / "query-streams"; std::filesystem::remove_all(path); @@ -74,14 +59,11 @@ class StreamsTest : public ::testing::Test { // Streams constructor. // InterpreterContext::auth_checker_ is used in the Streams object, but only in the message processing part. Because // these tests don't send any messages, the auth_checker_ pointer can be left as nullptr. - query::InterpreterContext interpreter_context_{&db_, query::InterpreterConfig{}, data_directory_, - "dont care bootstrap servers"}; + query::InterpreterContext interpreter_context_{&db_, query::InterpreterConfig{}, data_directory_}; std::filesystem::path streams_data_directory_{data_directory_ / "separate-dir-for-test"}; std::optional streams_; - void ResetStreamsObject() { - streams_.emplace(&interpreter_context_, mock_cluster_.Bootstraps(), streams_data_directory_); - } + void ResetStreamsObject() { streams_.emplace(&interpreter_context_, streams_data_directory_); } void CheckStreamStatus(const StreamCheckData &check_data) { SCOPED_TRACE(fmt::format("Checking status of '{}'", check_data.name)); @@ -106,6 +88,21 @@ class StreamsTest : public ::testing::Test { check_data.is_running = false; } + StreamInfo CreateDefaultStreamInfo() { + return StreamInfo{.common_info{ + .batch_interval = std::nullopt, + .batch_size = std::nullopt, + .transformation_name = "not used in the tests", + }, + .topics = {kTopicName}, + .consumer_group = "ConsumerGroup " + GetDefaultStreamName(), + .bootstrap_servers = mock_cluster_.Bootstraps()}; + } + + StreamCheckData CreateDefaultStreamCheckData() { + return {GetDefaultStreamName(), CreateDefaultStreamInfo(), false, std::nullopt}; + } + void Clear() { if (!std::filesystem::exists(data_directory_)) return; std::filesystem::remove_all(data_directory_);