You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2025/05/28 11:45:22 UTC
(pulsar-client-cpp) branch main updated: [CI] Add clang-tidy check with clang-analyzer and performance checks (#490)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new b0c4412 [CI] Add clang-tidy check with clang-analyzer and performance checks (#490)
b0c4412 is described below
commit b0c44122b389ddf8e2ff9f3bd3d8a4e5d00dcc37
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed May 28 19:45:17 2025 +0800
[CI] Add clang-tidy check with clang-analyzer and performance checks (#490)
---
.clang-tidy | 32 +++++
.github/workflows/ci-pr-validation.yaml | 53 ++++++---
.gitignore | 3 +
CMakeLists.txt | 2 -
build-support/run_clang_tidy.sh | 41 +++++++
include/pulsar/Authentication.h | 5 +-
include/pulsar/Client.h | 28 ++---
include/pulsar/ClientConfiguration.h | 4 +-
include/pulsar/CompressionType.h | 4 +-
include/pulsar/ConsoleLoggerFactory.h | 2 +
include/pulsar/Consumer.h | 34 +++---
include/pulsar/ConsumerConfiguration.h | 2 +-
include/pulsar/ConsumerType.h | 4 +-
include/pulsar/CryptoKeyReader.h | 2 +-
include/pulsar/EncryptionKeyInfo.h | 4 +-
include/pulsar/FileLoggerFactory.h | 2 +
include/pulsar/InitialPosition.h | 4 +-
include/pulsar/KeySharedPolicy.h | 3 +-
include/pulsar/Logger.h | 4 +-
include/pulsar/Producer.h | 8 +-
include/pulsar/ProducerConfiguration.h | 11 +-
include/pulsar/ProducerCryptoFailureAction.h | 4 +-
include/pulsar/Reader.h | 14 +--
include/pulsar/Result.h | 3 +-
include/pulsar/Schema.h | 9 +-
include/pulsar/TableView.h | 6 +-
lib/AckGroupingTracker.cc | 5 +-
lib/AckGroupingTracker.h | 27 ++---
lib/AckGroupingTrackerDisabled.cc | 9 +-
lib/AckGroupingTrackerDisabled.h | 8 +-
lib/AckGroupingTrackerEnabled.cc | 23 ++--
lib/AckGroupingTrackerEnabled.h | 15 ++-
lib/BinaryProtoLookupService.cc | 17 +--
lib/BinaryProtoLookupService.h | 15 +--
lib/BrokerConsumerStats.cc | 3 +-
lib/BrokerConsumerStatsImpl.cc | 6 +-
lib/Client.cc | 32 ++---
lib/ClientConnection.cc | 47 ++++----
lib/ClientConnection.h | 29 ++---
lib/ClientImpl.cc | 63 +++++-----
lib/ClientImpl.h | 58 ++++-----
lib/Commands.cc | 14 +--
lib/Commands.h | 2 +-
lib/ConnectionPool.cc | 3 +-
lib/ConnectionPool.h | 2 +-
lib/Consumer.cc | 28 ++---
lib/ConsumerConfiguration.cc | 8 +-
lib/ConsumerConfigurationImpl.h | 43 ++++---
lib/ConsumerImpl.cc | 87 +++++++-------
lib/ConsumerImpl.h | 56 +++++----
lib/ConsumerImplBase.cc | 12 +-
lib/ConsumerImplBase.h | 30 ++---
lib/CryptoKeyReader.cc | 2 +-
lib/EncryptionKeyInfo.cc | 4 +-
lib/EncryptionKeyInfoImpl.cc | 4 +-
lib/EncryptionKeyInfoImpl.h | 2 +-
lib/Future.h | 3 +-
lib/HTTPLookupService.cc | 13 ++-
lib/HTTPLookupService.h | 15 ++-
lib/KeyValue.cc | 2 +-
lib/LogUtils.cc | 5 +-
lib/MessageCrypto.cc | 8 +-
lib/MessageCrypto.h | 8 +-
lib/MessageImpl.cc | 2 +-
lib/MessageImpl.h | 2 +-
lib/MultiTopicsBrokerConsumerStatsImpl.cc | 2 +-
lib/MultiTopicsBrokerConsumerStatsImpl.h | 2 +-
lib/MultiTopicsConsumerImpl.cc | 168 ++++++++++++++-------------
lib/MultiTopicsConsumerImpl.h | 90 +++++++-------
lib/NegativeAcksTracker.cc | 2 +-
lib/NegativeAcksTracker.h | 3 +-
lib/PartitionedProducerImpl.cc | 26 ++---
lib/PartitionedProducerImpl.h | 16 ++-
lib/PatternMultiTopicsConsumerImpl.cc | 31 ++---
lib/PatternMultiTopicsConsumerImpl.h | 23 ++--
lib/Producer.cc | 8 +-
lib/ProducerConfiguration.cc | 4 +-
lib/ProducerImpl.cc | 16 +--
lib/ProducerImpl.h | 5 +-
lib/ProducerImplBase.h | 2 +-
lib/ProtobufNativeSchema.cc | 4 +-
lib/Reader.cc | 14 +--
lib/ReaderConfiguration.cc | 6 +-
lib/ReaderImpl.cc | 22 ++--
lib/ReaderImpl.h | 23 ++--
lib/Schema.cc | 8 +-
lib/SharedBuffer.h | 4 +-
lib/TableView.cc | 8 +-
lib/TableViewImpl.cc | 12 +-
lib/TableViewImpl.h | 12 +-
lib/UnAckedMessageTrackerEnabled.cc | 10 +-
lib/UnAckedMessageTrackerEnabled.h | 7 +-
lib/Utils.h | 2 +-
lib/auth/AuthOauth2.cc | 4 +-
lib/auth/AuthOauth2.h | 2 +-
lib/auth/AuthToken.cc | 3 +-
lib/stats/ConsumerStatsImpl.cc | 9 +-
lib/stats/ConsumerStatsImpl.h | 6 +-
lib/stats/ProducerStatsImpl.cc | 4 +-
lib/stats/ProducerStatsImpl.h | 2 +-
tests/AcknowledgeTest.cc | 2 +-
tests/AuthPluginTest.cc | 105 +++++++++++++----
tests/BasicEndToEndTest.cc | 44 ++++---
tests/BatchMessageTest.cc | 10 +-
tests/ClientTest.cc | 12 +-
tests/ConsumerConfigurationTest.cc | 2 +-
tests/ConsumerStatsTest.cc | 10 +-
tests/ConsumerTest.cc | 2 +-
tests/CustomLoggerTest.cc | 2 +-
tests/InterceptorsTest.cc | 7 +-
tests/LookupServiceTest.cc | 4 +-
tests/MultiTopicsConsumerTest.cc | 2 +-
tests/PartitionsUpdateTest.cc | 6 +-
tests/ProducerTest.cc | 6 +-
tests/ProtobufNativeSchemaTest.cc | 2 +-
tests/ReaderConfigurationTest.cc | 2 +-
tests/ReaderTest.cc | 4 +-
tests/ServiceURITest.cc | 2 +-
tests/SynchronizedHashMapTest.cc | 8 +-
tests/ZeroQueueSizeTest.cc | 8 +-
tests/c/c_BasicEndToEndTest.cc | 4 +-
tests/oauth2/Oauth2Test.cc | 2 +-
tests/unix/ConnectionFailTest.cc | 4 +-
123 files changed, 1024 insertions(+), 786 deletions(-)
diff --git a/.clang-tidy b/.clang-tidy
new file mode 100644
index 0000000..e091e0a
--- /dev/null
+++ b/.clang-tidy
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+# TODO: add more checks
+# - bugprone
+# - cppcoreguidelines
+# - portability
+# - readability
+Checks: >
+ -*,
+ clang-analyzer-*,
+ -clang-analyzer-security.insecureAPI.rand,
+ performance-*,
+ -performance-inefficient-string-concatenation
+WarningsAsErrors: '*'
+HeaderFilterRegex: '^(?!.*PulsarApi\.pb\.h$).*$'
+FormatStyle: none
diff --git a/.github/workflows/ci-pr-validation.yaml b/.github/workflows/ci-pr-validation.yaml
index 861b0d9..c514009 100644
--- a/.github/workflows/ci-pr-validation.yaml
+++ b/.github/workflows/ci-pr-validation.yaml
@@ -77,10 +77,37 @@ jobs:
cmake -S wireshark -B build-wireshark
cmake --build build-wireshark
+ lint:
+ name: Lint
+ needs: formatting-check
+ runs-on: ubuntu-24.04
+ timeout-minutes: 120
+
+ steps:
+ - name: checkout
+ uses: actions/checkout@v3
+ with:
+ fetch-depth: 0
+ submodules: recursive
+
+ - name: Build the project
+ run: |
+ cmake -B build -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=ON -DCMAKE_EXPORT_COMPILE_COMMANDS=ON
+ cmake --build build -j8
+
+ - name: Tidy check
+ run: |
+ sudo apt-get install -y clang-tidy
+ ./build-support/run_clang_tidy.sh
+ if [[ $? -ne 0 ]]; then
+ echo "clang-tidy failed"
+ exit 1
+ fi
+
unit-tests:
name: Run unit tests
needs: formatting-check
- runs-on: ubuntu-22.04
+ runs-on: ubuntu-24.04
timeout-minutes: 120
steps:
@@ -92,8 +119,8 @@ jobs:
- name: Build core libraries
run: |
- cmake . -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=OFF
- cmake --build . -j8
+ cmake -B build -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=OFF
+ cmake --build build -j8
- name: Check formatting
run: |
@@ -105,29 +132,29 @@ jobs:
- name: Build tests
run: |
- cmake . -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=ON
- cmake --build . -j8
+ cmake -B build -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=ON -DCMAKE_EXPORT_COMPILE_COMMANDS=ON
+ cmake --build build -j8
- name: Install gtest-parallel
run: |
sudo curl -o /gtest-parallel https://raw.githubusercontent.com/google/gtest-parallel/master/gtest_parallel.py
- name: Run unit tests
- run: RETRY_FAILED=3 ./run-unit-tests.sh
+ run: RETRY_FAILED=3 CMAKE_BUILD_DIRECTORY=./build ./run-unit-tests.sh
- name: Build perf tools
run: |
- cmake . -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=ON -DBUILD_PERF_TOOLS=ON
- cmake --build . -j8
+ cmake -B build -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=ON -DBUILD_PERF_TOOLS=ON
+ cmake --build build -j8
- name: Verify custom vcpkg installation
run: |
mv vcpkg /tmp/
- cmake -B build -DINTEGRATE_VCPKG=ON -DCMAKE_TOOLCHAIN_FILE="/tmp/vcpkg/scripts/buildsystems/vcpkg.cmake"
+ cmake -B build-2 -DINTEGRATE_VCPKG=ON -DCMAKE_TOOLCHAIN_FILE="/tmp/vcpkg/scripts/buildsystems/vcpkg.cmake"
cpp20-build:
name: Build with the C++20 standard
- needs: formatting-check
+ needs: lint
runs-on: ubuntu-22.04
timeout-minutes: 60
@@ -270,7 +297,7 @@ jobs:
package:
name: Build ${{matrix.pkg.name}} ${{matrix.cpu.platform}}
runs-on: ubuntu-22.04
- needs: unit-tests
+ needs: [lint, unit-tests]
timeout-minutes: 500
strategy:
@@ -314,7 +341,7 @@ jobs:
timeout-minutes: 120
name: Build CPP Client on macOS with static dependencies
runs-on: macos-14
- needs: formatting-check
+ needs: lint
steps:
- name: checkout
uses: actions/checkout@v3
@@ -329,7 +356,7 @@ jobs:
check-completion:
name: Check Completion
runs-on: ubuntu-latest
- needs: [formatting-check, wireshark-dissector-build, unit-tests, cpp20-build, cpp-build-windows, package, cpp-build-macos]
+ needs: [formatting-check, wireshark-dissector-build, lint, unit-tests, cpp20-build, cpp-build-windows, package, cpp-build-macos]
steps:
- run: true
diff --git a/.gitignore b/.gitignore
index a1e5224..f961f37 100644
--- a/.gitignore
+++ b/.gitignore
@@ -106,3 +106,6 @@ Testing
.test-token.txt
pkg/mac/.build
pkg/mac/.install
+
+# Used in ./build-support/run_clang_tidy.sh
+files.txt
diff --git a/CMakeLists.txt b/CMakeLists.txt
index b004653..d9af806 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -97,8 +97,6 @@ else() # GCC or Clang are mostly compatible:
# Options unique to Clang or GCC:
if (CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Qunused-arguments)
- elseif (CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND NOT (CMAKE_CXX_COMPILER_VERSION VERSION_LESS 8.1))
- add_compile_options(-Wno-stringop-truncation)
endif()
endif()
diff --git a/build-support/run_clang_tidy.sh b/build-support/run_clang_tidy.sh
new file mode 100755
index 0000000..d6e068b
--- /dev/null
+++ b/build-support/run_clang_tidy.sh
@@ -0,0 +1,41 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+set -e
+cd `dirname $0`/..
+
+FILES=$(find $PWD/include $PWD/lib $PWD/tests $PWD/examples -name "*.h" -o -name "*.cc" \
+ | grep -v "lib\/c\/" | grep -v "lib\/checksum\/" | grep -v "lib\/lz4\/" \
+ | grep -v "include\/pulsar\/c\/" | grep -v "tests\/c\/")
+
+rm -f files.txt
+for FILE in $FILES; do
+ echo $FILE >> files.txt
+done
+# run-clang-tidy from older version of LLVM requires python but not python3 as the env, so we cannot run it directly
+SCRIPT=$(which run-clang-tidy)
+set +e
+nproc
+if [[ $? == 0 ]]; then
+ python3 $SCRIPT -p build -j$(nproc) $(cat files.txt)
+else
+ python3 $SCRIPT -p build -j8 $(cat files.txt)
+fi
+rm -f files.txt
diff --git a/include/pulsar/Authentication.h b/include/pulsar/Authentication.h
index fb74df6..34b70cd 100644
--- a/include/pulsar/Authentication.h
+++ b/include/pulsar/Authentication.h
@@ -394,10 +394,7 @@ class PULSAR_PUBLIC AuthAthenz : public Authentication {
// currently mainly works for access token
class Oauth2TokenResult {
public:
- enum
- {
- undefined_expiration = -1
- };
+ static constexpr int undefined_expiration = -1;
Oauth2TokenResult();
~Oauth2TokenResult();
diff --git a/include/pulsar/Client.h b/include/pulsar/Client.h
index 47e82f4..5613066 100644
--- a/include/pulsar/Client.h
+++ b/include/pulsar/Client.h
@@ -101,7 +101,7 @@ class PULSAR_PUBLIC Client {
* @param callback the callback that is triggered when the producer is created successfully or not
* @param callback Callback function that is invoked when the operation is completed
*/
- void createProducerAsync(const std::string& topic, CreateProducerCallback callback);
+ void createProducerAsync(const std::string& topic, const CreateProducerCallback& callback);
/**
* Asynchronously create a producer with the customized ProducerConfiguration for publishing on a specific
@@ -110,8 +110,8 @@ class PULSAR_PUBLIC Client {
* @param topic the name of the topic where to produce
* @param conf the customized ProducerConfiguration
*/
- void createProducerAsync(const std::string& topic, ProducerConfiguration conf,
- CreateProducerCallback callback);
+ void createProducerAsync(const std::string& topic, const ProducerConfiguration& conf,
+ const CreateProducerCallback& callback);
/**
* Subscribe to a given topic and subscription combination with the default ConsumerConfiguration
@@ -144,7 +144,7 @@ class PULSAR_PUBLIC Client {
* default ConsumerConfiguration are asynchronously subscribed successfully or not
*/
void subscribeAsync(const std::string& topic, const std::string& subscriptionName,
- SubscribeCallback callback);
+ const SubscribeCallback& callback);
/**
* Asynchronously subscribe to a given topic and subscription combination with the customized
@@ -157,7 +157,7 @@ class PULSAR_PUBLIC Client {
* customized ConsumerConfiguration are asynchronously subscribed successfully or not
*/
void subscribeAsync(const std::string& topic, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback);
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback);
/**
* Subscribe to multiple topics under the same namespace.
@@ -191,7 +191,7 @@ class PULSAR_PUBLIC Client {
*/
void subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
- SubscribeCallback callback);
+ const SubscribeCallback& callback);
/**
* Asynchronously subscribe to a list of topics and subscription combination using the customized
@@ -204,7 +204,7 @@ class PULSAR_PUBLIC Client {
* the customized ConsumerConfiguration are asynchronously subscribed successfully or not
*/
void subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback);
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback);
/**
* Subscribe to multiple topics, which match given regexPattern, under the same namespace.
@@ -227,7 +227,7 @@ class PULSAR_PUBLIC Client {
* SubscribeCallback)
*/
void subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
- SubscribeCallback callback);
+ const SubscribeCallback& callback);
/**
* Asynchronously subscribe to multiple topics (which match given regexPatterns) with the customized
@@ -240,7 +240,7 @@ class PULSAR_PUBLIC Client {
* ConsumerConfiguration under the same namespace are asynchronously subscribed successfully or not
*/
void subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback);
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback);
/**
* Create a topic reader with given {@code ReaderConfiguration} for reading messages from the specified
@@ -301,7 +301,7 @@ class PULSAR_PUBLIC Client {
* @return the Reader object
*/
void createReaderAsync(const std::string& topic, const MessageId& startMessageId,
- const ReaderConfiguration& conf, ReaderCallback callback);
+ const ReaderConfiguration& conf, const ReaderCallback& callback);
/**
* Create a table view with given {@code TableViewConfiguration} for specified topic.
@@ -331,7 +331,7 @@ class PULSAR_PUBLIC Client {
* built from a message that already exists
*/
void createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
- TableViewCallback callBack);
+ const TableViewCallback& callBack);
/**
* Get the list of partitions for a given topic.
@@ -363,7 +363,7 @@ class PULSAR_PUBLIC Client {
* the callback that will be invoked when the list of partitions is available
* @since 2.3.0
*/
- void getPartitionsForTopicAsync(const std::string& topic, GetPartitionsCallback callback);
+ void getPartitionsForTopicAsync(const std::string& topic, const GetPartitionsCallback& callback);
/**
*
@@ -380,7 +380,7 @@ class PULSAR_PUBLIC Client {
* @param callback the callback that is triggered when the Pulsar client is asynchronously closed
* successfully or not
*/
- void closeAsync(CloseCallback callback);
+ void closeAsync(const CloseCallback& callback);
/**
* Perform immediate shutdown of Pulsar client.
@@ -415,7 +415,7 @@ class PULSAR_PUBLIC Client {
std::function<void(Result, const SchemaInfo&)> callback);
private:
- Client(const std::shared_ptr<ClientImpl>);
+ Client(const std::shared_ptr<ClientImpl>&);
friend class PulsarFriend;
friend class PulsarWrapper;
diff --git a/include/pulsar/ClientConfiguration.h b/include/pulsar/ClientConfiguration.h
index e8c4657..03e56e7 100644
--- a/include/pulsar/ClientConfiguration.h
+++ b/include/pulsar/ClientConfiguration.h
@@ -23,6 +23,8 @@
#include <pulsar/Logger.h>
#include <pulsar/defines.h>
+#include <cstdint>
+
namespace pulsar {
class PulsarWrapper;
struct ClientConfigurationImpl;
@@ -32,7 +34,7 @@ class PULSAR_PUBLIC ClientConfiguration {
~ClientConfiguration();
ClientConfiguration(const ClientConfiguration&);
ClientConfiguration& operator=(const ClientConfiguration&);
- enum ProxyProtocol
+ enum ProxyProtocol : uint8_t
{
SNI = 0
};
diff --git a/include/pulsar/CompressionType.h b/include/pulsar/CompressionType.h
index 6fd663a..0a889d1 100644
--- a/include/pulsar/CompressionType.h
+++ b/include/pulsar/CompressionType.h
@@ -19,8 +19,10 @@
#ifndef PULSAR_COMPRESSIONTYPE_H_
#define PULSAR_COMPRESSIONTYPE_H_
+#include <cstdint>
+
namespace pulsar {
-enum CompressionType
+enum CompressionType : std::uint8_t
{
CompressionNone = 0,
CompressionLZ4 = 1,
diff --git a/include/pulsar/ConsoleLoggerFactory.h b/include/pulsar/ConsoleLoggerFactory.h
index bfb5e9e..ec1de26 100644
--- a/include/pulsar/ConsoleLoggerFactory.h
+++ b/include/pulsar/ConsoleLoggerFactory.h
@@ -21,6 +21,8 @@
#include <pulsar/Logger.h>
+#include <memory>
+
namespace pulsar {
class ConsoleLoggerFactoryImpl;
diff --git a/include/pulsar/Consumer.h b/include/pulsar/Consumer.h
index 63defdb..6687b1d 100644
--- a/include/pulsar/Consumer.h
+++ b/include/pulsar/Consumer.h
@@ -83,7 +83,7 @@ class PULSAR_PUBLIC Consumer {
*
* @param callback the callback to get notified when the operation is complete
*/
- void unsubscribeAsync(ResultCallback callback);
+ void unsubscribeAsync(const ResultCallback& callback);
/**
* Receive a single message.
@@ -134,10 +134,10 @@ class PULSAR_PUBLIC Consumer {
* </p>
* @param ReceiveCallback will be completed when message is available
*/
- void receiveAsync(ReceiveCallback callback);
+ void receiveAsync(const ReceiveCallback& callback);
template <typename T>
- void receiveAsync(std::function<void(Result result, const TypedMessage<T>&)> callback,
+ void receiveAsync(const std::function<void(Result result, const TypedMessage<T>&)>& callback,
typename TypedMessage<T>::Decoder decoder) {
receiveAsync([callback, decoder](Result result, const Message& msg) {
callback(result, TypedMessage<T>{msg, decoder});
@@ -167,7 +167,7 @@ class PULSAR_PUBLIC Consumer {
* </p>
* @param BatchReceiveCallback will be completed when messages are available.
*/
- void batchReceiveAsync(BatchReceiveCallback callback);
+ void batchReceiveAsync(const BatchReceiveCallback& callback);
/**
* Acknowledge the reception of a single message.
@@ -209,7 +209,7 @@ class PULSAR_PUBLIC Consumer {
* @param message the message to acknowledge
* @param callback callback that will be triggered when the message has been acknowledged
*/
- void acknowledgeAsync(const Message& message, ResultCallback callback);
+ void acknowledgeAsync(const Message& message, const ResultCallback& callback);
/**
* Asynchronously acknowledge the reception of a single message.
@@ -220,7 +220,7 @@ class PULSAR_PUBLIC Consumer {
* @param messageId the messageId to acknowledge
* @param callback the callback that is triggered when the message has been acknowledged or not
*/
- void acknowledgeAsync(const MessageId& messageId, ResultCallback callback);
+ void acknowledgeAsync(const MessageId& messageId, const ResultCallback& callback);
/**
* Asynchronously acknowledge the consumption of a list of message.
@@ -228,7 +228,7 @@ class PULSAR_PUBLIC Consumer {
* @param callback the callback that is triggered when the message has been acknowledged or not
* @return
*/
- void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback);
+ void acknowledgeAsync(const MessageIdList& messageIdList, const ResultCallback& callback);
/**
* Acknowledge the reception of all the messages in the stream up to (and including)
@@ -239,7 +239,7 @@ class PULSAR_PUBLIC Consumer {
*
* Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.
*
- * It's equivalent to calling asyncAcknowledgeCumulative(const Message&, ResultCallback) and
+ * It's equivalent to calling asyncAcknowledgeCumulative(const Message&, const ResultCallback&) and
* waiting for the callback to be triggered.
*
* @param message the last message in the stream to acknowledge
@@ -258,8 +258,8 @@ class PULSAR_PUBLIC Consumer {
*
* Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.
*
- * It is equivalent to calling the asyncAcknowledgeCumulative(const Message&, ResultCallback) method and
- * waiting for the callback to be triggered.
+ * It is equivalent to calling the asyncAcknowledgeCumulative(const Message&, const ResultCallback&)
+ * method and waiting for the callback to be triggered.
*
* @param messageId the last messageId in the stream to acknowledge
* @return ResultOk if the message is successfully acknowledged. All previously delivered messages for
@@ -277,7 +277,7 @@ class PULSAR_PUBLIC Consumer {
* @param message the message to acknowledge
* @param callback callback that will be triggered when the message has been acknowledged
*/
- void acknowledgeCumulativeAsync(const Message& message, ResultCallback callback);
+ void acknowledgeCumulativeAsync(const Message& message, const ResultCallback& callback);
/**
* Asynchronously acknowledge the reception of all the messages in the stream up to (and
@@ -289,7 +289,7 @@ class PULSAR_PUBLIC Consumer {
* @param messageId the messageId to acknowledge
* @param callback the callback that is triggered when the message has been acknowledged or not
*/
- void acknowledgeCumulativeAsync(const MessageId& messageId, ResultCallback callback);
+ void acknowledgeCumulativeAsync(const MessageId& messageId, const ResultCallback& callback);
/**
* Acknowledge the failure to process a single message.
@@ -364,7 +364,7 @@ class PULSAR_PUBLIC Consumer {
* Asynchronously close the consumer and stop the broker to push more messages
*
*/
- void closeAsync(ResultCallback callback);
+ void closeAsync(const ResultCallback& callback);
/**
* Pause receiving messages via the messageListener, till resumeMessageListener() is called.
@@ -413,7 +413,7 @@ class PULSAR_PUBLIC Consumer {
* @param callback - callback function to get the brokerConsumerStats,
* if result is ResultOk then the brokerConsumerStats will be populated
*/
- void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback);
+ void getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback);
/**
* Reset the subscription associated with this consumer to a specific message id.
@@ -445,7 +445,7 @@ class PULSAR_PUBLIC Consumer {
* @param messageId
* the message id where to reposition the subscription
*/
- virtual void seekAsync(const MessageId& messageId, ResultCallback callback);
+ virtual void seekAsync(const MessageId& messageId, const ResultCallback& callback);
/**
* Asynchronously reset the subscription associated with this consumer to a specific message publish time.
@@ -453,7 +453,7 @@ class PULSAR_PUBLIC Consumer {
* @param timestamp
* the message publish time where to reposition the subscription
*/
- virtual void seekAsync(uint64_t timestamp, ResultCallback callback);
+ virtual void seekAsync(uint64_t timestamp, const ResultCallback& callback);
/**
* @return Whether the consumer is currently connected to the broker
@@ -464,7 +464,7 @@ class PULSAR_PUBLIC Consumer {
* Asynchronously get an ID of the last available message or a message ID with -1 as an entryId if the
* topic is empty.
*/
- void getLastMessageIdAsync(GetLastMessageIdCallback callback);
+ void getLastMessageIdAsync(const GetLastMessageIdCallback& callback);
/**
* Get an ID of the last available message or a message ID with -1 as an entryId if the topic is empty.
diff --git a/include/pulsar/ConsumerConfiguration.h b/include/pulsar/ConsumerConfiguration.h
index eef2bdd..3254a95 100644
--- a/include/pulsar/ConsumerConfiguration.h
+++ b/include/pulsar/ConsumerConfiguration.h
@@ -117,7 +117,7 @@ class PULSAR_PUBLIC ConsumerConfiguration {
*
* @param keySharedPolicy The {@link KeySharedPolicy} want to specify
*/
- ConsumerConfiguration& setKeySharedPolicy(KeySharedPolicy keySharedPolicy);
+ ConsumerConfiguration& setKeySharedPolicy(const KeySharedPolicy& keySharedPolicy);
/**
* @return the KeyShared subscription policy
diff --git a/include/pulsar/ConsumerType.h b/include/pulsar/ConsumerType.h
index 8068a5c..8ff0abb 100644
--- a/include/pulsar/ConsumerType.h
+++ b/include/pulsar/ConsumerType.h
@@ -19,8 +19,10 @@
#ifndef PULSAR_CPP_CONSUMERTYPE_H
#define PULSAR_CPP_CONSUMERTYPE_H
+#include <cstdint>
+
namespace pulsar {
-enum ConsumerType
+enum ConsumerType : uint8_t
{
/**
* There can be only 1 consumer on the same topic with the same consumerName
diff --git a/include/pulsar/CryptoKeyReader.h b/include/pulsar/CryptoKeyReader.h
index d11e1f8..262ff16 100644
--- a/include/pulsar/CryptoKeyReader.h
+++ b/include/pulsar/CryptoKeyReader.h
@@ -71,7 +71,7 @@ class PULSAR_PUBLIC DefaultCryptoKeyReader : public CryptoKeyReader {
private:
std::string publicKeyPath_;
std::string privateKeyPath_;
- void readFile(std::string fileName, std::string& fileContents) const;
+ void readFile(const std::string& fileName, std::string& fileContents) const;
public:
/**
diff --git a/include/pulsar/EncryptionKeyInfo.h b/include/pulsar/EncryptionKeyInfo.h
index 401d7be..68d7b4c 100644
--- a/include/pulsar/EncryptionKeyInfo.h
+++ b/include/pulsar/EncryptionKeyInfo.h
@@ -59,7 +59,7 @@ class PULSAR_PUBLIC EncryptionKeyInfo {
*
* @param Key the key of the message for routing policy
*/
- void setKey(std::string key);
+ void setKey(const std::string& key);
/**
* @return the metadata information
@@ -74,7 +74,7 @@ class PULSAR_PUBLIC EncryptionKeyInfo {
void setMetadata(StringMap& metadata);
private:
- explicit EncryptionKeyInfo(EncryptionKeyInfoImplPtr);
+ explicit EncryptionKeyInfo(const EncryptionKeyInfoImplPtr&);
EncryptionKeyInfoImplPtr impl_;
diff --git a/include/pulsar/FileLoggerFactory.h b/include/pulsar/FileLoggerFactory.h
index 92cd180..138df93 100644
--- a/include/pulsar/FileLoggerFactory.h
+++ b/include/pulsar/FileLoggerFactory.h
@@ -21,6 +21,8 @@
#include <pulsar/Logger.h>
+#include <memory>
+
namespace pulsar {
class FileLoggerFactoryImpl;
diff --git a/include/pulsar/InitialPosition.h b/include/pulsar/InitialPosition.h
index f5ac19e..71a6929 100644
--- a/include/pulsar/InitialPosition.h
+++ b/include/pulsar/InitialPosition.h
@@ -19,8 +19,10 @@
#ifndef PULSAR_CPP_INITIAL_POSITION_H
#define PULSAR_CPP_INITIAL_POSITION_H
+#include <cstdint>
+
namespace pulsar {
-enum InitialPosition
+enum InitialPosition : uint8_t
{
InitialPositionLatest,
InitialPositionEarliest
diff --git a/include/pulsar/KeySharedPolicy.h b/include/pulsar/KeySharedPolicy.h
index ee0a979..70705a6 100644
--- a/include/pulsar/KeySharedPolicy.h
+++ b/include/pulsar/KeySharedPolicy.h
@@ -20,6 +20,7 @@
#include <pulsar/defines.h>
+#include <cstdint>
#include <memory>
#include <utility>
#include <vector>
@@ -29,7 +30,7 @@ namespace pulsar {
/**
* KeyShared mode of KeyShared subscription.
*/
-enum KeySharedMode
+enum KeySharedMode : uint8_t
{
/**
diff --git a/include/pulsar/Logger.h b/include/pulsar/Logger.h
index 710cdc2..8dea8ee 100644
--- a/include/pulsar/Logger.h
+++ b/include/pulsar/Logger.h
@@ -20,14 +20,14 @@
#include <pulsar/defines.h>
-#include <memory>
+#include <cstdint>
#include <string>
namespace pulsar {
class PULSAR_PUBLIC Logger {
public:
- enum Level
+ enum Level : std::uint8_t
{
LEVEL_DEBUG = 0,
LEVEL_INFO = 1,
diff --git a/include/pulsar/Producer.h b/include/pulsar/Producer.h
index 51d2980..4d5dbfa 100644
--- a/include/pulsar/Producer.h
+++ b/include/pulsar/Producer.h
@@ -97,7 +97,7 @@ class PULSAR_PUBLIC Producer {
* @param msg message to publish
* @param callback the callback to get notification of the completion
*/
- void sendAsync(const Message& msg, SendCallback callback);
+ void sendAsync(const Message& msg, const SendCallback& callback);
/**
* Flush all the messages buffered in the client and wait until all messages have been successfully
@@ -109,7 +109,7 @@ class PULSAR_PUBLIC Producer {
* Flush all the messages buffered in the client and wait until all messages have been successfully
* persisted.
*/
- void flushAsync(FlushCallback callback);
+ void flushAsync(const FlushCallback& callback);
/**
* Get the last sequence id that was published by this producer.
@@ -153,7 +153,7 @@ class PULSAR_PUBLIC Producer {
* triggered when all pending write requests are persisted. In case of errors,
* pending writes will not be retried.
*/
- void closeAsync(CloseCallback callback);
+ void closeAsync(const CloseCallback& callback);
/**
* @return Whether the producer is currently connected to the broker
@@ -161,7 +161,7 @@ class PULSAR_PUBLIC Producer {
bool isConnected() const;
private:
- explicit Producer(ProducerImplBasePtr);
+ explicit Producer(const ProducerImplBasePtr&);
friend class ClientImpl;
friend class PulsarFriend;
diff --git a/include/pulsar/ProducerConfiguration.h b/include/pulsar/ProducerConfiguration.h
index 62a6380..1724fff 100644
--- a/include/pulsar/ProducerConfiguration.h
+++ b/include/pulsar/ProducerConfiguration.h
@@ -28,6 +28,7 @@
#include <pulsar/Schema.h>
#include <pulsar/defines.h>
+#include <cstdint>
#include <functional>
#include <set>
@@ -44,19 +45,19 @@ class PulsarWrapper;
*/
class PULSAR_PUBLIC ProducerConfiguration {
public:
- enum PartitionsRoutingMode
+ enum PartitionsRoutingMode : uint8_t
{
UseSinglePartition,
RoundRobinDistribution,
CustomPartition
};
- enum HashingScheme
+ enum HashingScheme : uint8_t
{
Murmur3_32Hash,
BoostHash,
JavaStringHash
};
- enum BatchingType
+ enum BatchingType : uint8_t
{
/**
* Default batching.
@@ -80,7 +81,7 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
KeyBasedBatching
};
- enum ProducerAccessMode
+ enum ProducerAccessMode : uint8_t
{
/**
* By default multiple producers can publish on a topic.
@@ -460,7 +461,7 @@ class PULSAR_PUBLIC ProducerConfiguration {
* @key the encryption key to add
* @return the ProducerConfiguration self
*/
- ProducerConfiguration& addEncryptionKey(std::string key);
+ ProducerConfiguration& addEncryptionKey(const std::string& key);
/**
* Check whether the producer has a specific property attached.
diff --git a/include/pulsar/ProducerCryptoFailureAction.h b/include/pulsar/ProducerCryptoFailureAction.h
index 7693956..4124f83 100644
--- a/include/pulsar/ProducerCryptoFailureAction.h
+++ b/include/pulsar/ProducerCryptoFailureAction.h
@@ -19,9 +19,11 @@
#ifndef PRODUCERCRYPTOFAILUREACTION_H_
#define PRODUCERCRYPTOFAILUREACTION_H_
+#include <cstdint>
+
namespace pulsar {
-enum class ProducerCryptoFailureAction
+enum class ProducerCryptoFailureAction : uint8_t
{
FAIL, // This is the default option to fail send if crypto operation fails
SEND // Ignore crypto failure and proceed with sending unencrypted messages
diff --git a/include/pulsar/Reader.h b/include/pulsar/Reader.h
index 7769051..0aa47cc 100644
--- a/include/pulsar/Reader.h
+++ b/include/pulsar/Reader.h
@@ -74,7 +74,7 @@ class PULSAR_PUBLIC Reader {
*
* @param callback
*/
- void readNextAsync(ReadNextCallback callback);
+ void readNextAsync(const ReadNextCallback& callback);
/**
* Close the reader and stop the broker to push more messages
@@ -88,12 +88,12 @@ class PULSAR_PUBLIC Reader {
*
* @param callback the callback that is triggered when the reader is closed
*/
- void closeAsync(ResultCallback callback);
+ void closeAsync(const ResultCallback& callback);
/**
* Asynchronously check if there is any message available to read from the current position.
*/
- void hasMessageAvailableAsync(HasMessageAvailableCallback callback);
+ void hasMessageAvailableAsync(const HasMessageAvailableCallback& callback);
/**
* Check if there is any message available to read from the current position.
@@ -130,7 +130,7 @@ class PULSAR_PUBLIC Reader {
* @param messageId
* the message id where to reposition the subscription
*/
- void seekAsync(const MessageId& msgId, ResultCallback callback);
+ void seekAsync(const MessageId& msgId, const ResultCallback& callback);
/**
* Asynchronously reset this reader to a specific message publish time.
@@ -138,7 +138,7 @@ class PULSAR_PUBLIC Reader {
* @param timestamp
* the message publish time where to reposition the subscription
*/
- void seekAsync(uint64_t timestamp, ResultCallback callback);
+ void seekAsync(uint64_t timestamp, const ResultCallback& callback);
/**
* @return Whether the reader is currently connected to the broker
@@ -149,7 +149,7 @@ class PULSAR_PUBLIC Reader {
* Asynchronously get an ID of the last available message or a message ID with -1 as an entryId if the
* topic is empty.
*/
- void getLastMessageIdAsync(GetLastMessageIdCallback callback);
+ void getLastMessageIdAsync(const GetLastMessageIdCallback& callback);
/**
* Get an ID of the last available message or a message ID with -1 as an entryId if the topic is empty.
@@ -159,7 +159,7 @@ class PULSAR_PUBLIC Reader {
private:
typedef std::shared_ptr<ReaderImpl> ReaderImplPtr;
ReaderImplPtr impl_;
- explicit Reader(ReaderImplPtr);
+ explicit Reader(const ReaderImplPtr&);
friend class PulsarFriend;
friend class PulsarWrapper;
diff --git a/include/pulsar/Result.h b/include/pulsar/Result.h
index 033f4cb..a6c30d4 100644
--- a/include/pulsar/Result.h
+++ b/include/pulsar/Result.h
@@ -21,6 +21,7 @@
#include <pulsar/defines.h>
+#include <cstdint>
#include <iosfwd>
namespace pulsar {
@@ -28,7 +29,7 @@ namespace pulsar {
/**
* Collection of return codes
*/
-enum Result
+enum Result : int8_t
{
ResultRetryable = -1, /// An internal error code used for retry
ResultOk = 0, /// Operation successful
diff --git a/include/pulsar/Schema.h b/include/pulsar/Schema.h
index 93dac4f..bec1735 100644
--- a/include/pulsar/Schema.h
+++ b/include/pulsar/Schema.h
@@ -20,6 +20,7 @@
#include <pulsar/defines.h>
+#include <cstdint>
#include <iosfwd>
#include <map>
#include <memory>
@@ -30,7 +31,7 @@ namespace pulsar {
/**
* Encoding types of supported KeyValueSchema for Pulsar messages.
*/
-enum class KeyValueEncodingType
+enum class KeyValueEncodingType : uint8_t
{
/**
* Key is stored as message key, while value is stored as message payload.
@@ -46,9 +47,9 @@ enum class KeyValueEncodingType
// Return string representation of result code
PULSAR_PUBLIC const char *strEncodingType(pulsar::KeyValueEncodingType encodingType);
-PULSAR_PUBLIC KeyValueEncodingType enumEncodingType(std::string encodingTypeStr);
+PULSAR_PUBLIC KeyValueEncodingType enumEncodingType(const std::string &encodingTypeStr);
-enum SchemaType
+enum SchemaType : int8_t
{
/**
* No schema defined
@@ -134,7 +135,7 @@ enum SchemaType
// Return string representation of result code
PULSAR_PUBLIC const char *strSchemaType(SchemaType schemaType);
-PULSAR_PUBLIC SchemaType enumSchemaType(std::string schemaTypeStr);
+PULSAR_PUBLIC SchemaType enumSchemaType(const std::string &schemaTypeStr);
class SchemaInfoImpl;
diff --git a/include/pulsar/TableView.h b/include/pulsar/TableView.h
index d0c278a..550c79d 100644
--- a/include/pulsar/TableView.h
+++ b/include/pulsar/TableView.h
@@ -98,18 +98,18 @@ class PULSAR_PUBLIC TableView {
* Performs the given action for each entry in this map until all entries have been processed or the
* action throws an exception.
*/
- void forEach(TableViewAction action);
+ void forEach(const TableViewAction& action);
/**
* Performs the given action for each entry in this map until all entries have been processed and
* register the callback, which will be called each time a key-value pair is updated.
*/
- void forEachAndListen(TableViewAction action);
+ void forEachAndListen(const TableViewAction& action);
/**
* Asynchronously close the tableview and stop the broker to push more messages
*/
- void closeAsync(ResultCallback callback);
+ void closeAsync(const ResultCallback& callback);
/**
* Close the table view and stop the broker to push more messages
diff --git a/lib/AckGroupingTracker.cc b/lib/AckGroupingTracker.cc
index ab7381f..8a9ea0d 100644
--- a/lib/AckGroupingTracker.cc
+++ b/lib/AckGroupingTracker.cc
@@ -34,7 +34,7 @@ namespace pulsar {
DECLARE_LOG_OBJECT();
-void AckGroupingTracker::doImmediateAck(const MessageId& msgId, ResultCallback callback,
+void AckGroupingTracker::doImmediateAck(const MessageId& msgId, const ResultCallback& callback,
CommandAck_AckType ackType) const {
const auto cnx = connectionSupplier_();
if (!cnx) {
@@ -87,7 +87,8 @@ static std::ostream& operator<<(std::ostream& os, const std::set<MessageId>& msg
return os;
}
-void AckGroupingTracker::doImmediateAck(const std::set<MessageId>& msgIds, ResultCallback callback) const {
+void AckGroupingTracker::doImmediateAck(const std::set<MessageId>& msgIds,
+ const ResultCallback& callback) const {
const auto cnx = connectionSupplier_();
if (!cnx) {
LOG_DEBUG("Connection is not ready, ACK failed for " << msgIds);
diff --git a/lib/AckGroupingTracker.h b/lib/AckGroupingTracker.h
index 97d0d85..f62492f 100644
--- a/lib/AckGroupingTracker.h
+++ b/lib/AckGroupingTracker.h
@@ -44,8 +44,8 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
public:
AckGroupingTracker(std::function<ClientConnectionPtr()> connectionSupplier,
std::function<uint64_t()> requestIdSupplier, uint64_t consumerId, bool waitResponse)
- : connectionSupplier_(connectionSupplier),
- requestIdSupplier_(requestIdSupplier),
+ : connectionSupplier_(std::move(connectionSupplier)),
+ requestIdSupplier_(std::move(requestIdSupplier)),
consumerId_(consumerId),
waitResponse_(waitResponse) {}
@@ -71,14 +71,16 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
* @param[in] msgId ID of the message to be ACKed.
* @param[in] callback the callback that is triggered when the message is acknowledged
*/
- virtual void addAcknowledge(const MessageId& msgId, ResultCallback callback) { callback(ResultOk); }
+ virtual void addAcknowledge(const MessageId& msgId, const ResultCallback& callback) {
+ callback(ResultOk);
+ }
/**
* Adding message ID list into ACK group for individual ACK.
* @param[in] msgIds of the message to be ACKed.
* @param[in] callback the callback that is triggered when the messages are acknowledged
*/
- virtual void addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) {
+ virtual void addAcknowledgeList(const MessageIdList& msgIds, const ResultCallback& callback) {
callback(ResultOk);
}
@@ -87,20 +89,10 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
* @param[in] msgId ID of the message to be ACKed.
* @param[in] callback the callback that is triggered when the message is acknowledged
*/
- virtual void addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) {
+ virtual void addAcknowledgeCumulative(const MessageId& msgId, const ResultCallback& callback) {
callback(ResultOk);
}
- /**
- * Flush all the pending grouped ACKs (as flush() does), and stop period ACKs sending.
- */
- virtual void close() {}
-
- /**
- * Flush all the pending grouped ACKs and send them to the broker.
- */
- virtual void flush() {}
-
/**
* Flush all the pending grouped ACKs (as flush() does), and clean all records about ACKed
* messages, such as last cumulative ACKed message ID.
@@ -108,8 +100,9 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
virtual void flushAndClean() {}
protected:
- void doImmediateAck(const MessageId& msgId, ResultCallback callback, CommandAck_AckType ackType) const;
- void doImmediateAck(const std::set<MessageId>& msgIds, ResultCallback callback) const;
+ void doImmediateAck(const MessageId& msgId, const ResultCallback& callback,
+ CommandAck_AckType ackType) const;
+ void doImmediateAck(const std::set<MessageId>& msgIds, const ResultCallback& callback) const;
private:
const std::function<ClientConnectionPtr()> connectionSupplier_;
diff --git a/lib/AckGroupingTrackerDisabled.cc b/lib/AckGroupingTrackerDisabled.cc
index b29e1f7..d20de44 100644
--- a/lib/AckGroupingTrackerDisabled.cc
+++ b/lib/AckGroupingTrackerDisabled.cc
@@ -19,16 +19,16 @@
#include "AckGroupingTrackerDisabled.h"
-#include "HandlerBase.h"
#include "ProtoApiEnums.h"
namespace pulsar {
-void AckGroupingTrackerDisabled::addAcknowledge(const MessageId& msgId, ResultCallback callback) {
+void AckGroupingTrackerDisabled::addAcknowledge(const MessageId& msgId, const ResultCallback& callback) {
doImmediateAck(msgId, callback, CommandAck_AckType_Individual);
}
-void AckGroupingTrackerDisabled::addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) {
+void AckGroupingTrackerDisabled::addAcknowledgeList(const MessageIdList& msgIds,
+ const ResultCallback& callback) {
std::set<MessageId> msgIdSet;
for (auto&& msgId : msgIds) {
msgIdSet.emplace(msgId);
@@ -36,7 +36,8 @@ void AckGroupingTrackerDisabled::addAcknowledgeList(const MessageIdList& msgIds,
doImmediateAck(msgIdSet, callback);
}
-void AckGroupingTrackerDisabled::addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) {
+void AckGroupingTrackerDisabled::addAcknowledgeCumulative(const MessageId& msgId,
+ const ResultCallback& callback) {
doImmediateAck(msgId, callback, CommandAck_AckType_Cumulative);
}
diff --git a/lib/AckGroupingTrackerDisabled.h b/lib/AckGroupingTrackerDisabled.h
index 9619360..6cf1a0c 100644
--- a/lib/AckGroupingTrackerDisabled.h
+++ b/lib/AckGroupingTrackerDisabled.h
@@ -19,8 +19,6 @@
#ifndef LIB_ACKGROUPINGTRACKERDISABLED_H_
#define LIB_ACKGROUPINGTRACKERDISABLED_H_
-#include <cstdint>
-
#include "AckGroupingTracker.h"
namespace pulsar {
@@ -37,9 +35,9 @@ class AckGroupingTrackerDisabled : public AckGroupingTracker {
using AckGroupingTracker::AckGroupingTracker;
virtual ~AckGroupingTrackerDisabled() = default;
- void addAcknowledge(const MessageId& msgId, ResultCallback callback) override;
- void addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) override;
- void addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) override;
+ void addAcknowledge(const MessageId& msgId, const ResultCallback& callback) override;
+ void addAcknowledgeList(const MessageIdList& msgIds, const ResultCallback& callback) override;
+ void addAcknowledgeCumulative(const MessageId& msgId, const ResultCallback& callback) override;
}; // class AckGroupingTrackerDisabled
} // namespace pulsar
diff --git a/lib/AckGroupingTrackerEnabled.cc b/lib/AckGroupingTrackerEnabled.cc
index 7233b2c..72f46f2 100644
--- a/lib/AckGroupingTrackerEnabled.cc
+++ b/lib/AckGroupingTrackerEnabled.cc
@@ -20,6 +20,7 @@
#include "AckGroupingTrackerEnabled.h"
#include <climits>
+#include <memory>
#include <mutex>
#include "ClientConnection.h"
@@ -60,7 +61,7 @@ bool AckGroupingTrackerEnabled::isDuplicate(const MessageId& msgId) {
return this->pendingIndividualAcks_.count(msgId) > 0;
}
-void AckGroupingTrackerEnabled::addAcknowledge(const MessageId& msgId, ResultCallback callback) {
+void AckGroupingTrackerEnabled::addAcknowledge(const MessageId& msgId, const ResultCallback& callback) {
std::lock_guard<std::recursive_mutex> lock(this->rmutexPendingIndAcks_);
this->pendingIndividualAcks_.insert(msgId);
if (waitResponse_) {
@@ -73,7 +74,8 @@ void AckGroupingTrackerEnabled::addAcknowledge(const MessageId& msgId, ResultCal
}
}
-void AckGroupingTrackerEnabled::addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) {
+void AckGroupingTrackerEnabled::addAcknowledgeList(const MessageIdList& msgIds,
+ const ResultCallback& callback) {
std::lock_guard<std::recursive_mutex> lock(this->rmutexPendingIndAcks_);
for (const auto& msgId : msgIds) {
this->pendingIndividualAcks_.emplace(msgId);
@@ -88,8 +90,10 @@ void AckGroupingTrackerEnabled::addAcknowledgeList(const MessageIdList& msgIds,
}
}
-void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) {
+void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId,
+ const ResultCallback& callback) {
std::unique_lock<std::mutex> lock(this->mutexCumulativeAckMsgId_);
+ bool completeCallback = true;
if (compare(msgId, this->nextCumulativeAckMsgId_) > 0) {
this->nextCumulativeAckMsgId_ = msgId;
this->requireCumulativeAck_ = true;
@@ -101,18 +105,18 @@ void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId,
// Move the callback to latestCumulativeCallback_ so that it will be triggered when receiving the
// AckResponse or being replaced by a newer MessageId
latestCumulativeCallback_ = callback;
- callback = nullptr;
+ completeCallback = false;
} else {
latestCumulativeCallback_ = nullptr;
}
}
lock.unlock();
- if (callback) {
+ if (callback && completeCallback) {
callback(ResultOk);
}
}
-void AckGroupingTrackerEnabled::close() {
+AckGroupingTrackerEnabled::~AckGroupingTrackerEnabled() {
isClosed_ = true;
this->flush();
std::lock_guard<std::mutex> lock(this->mutexTimer_);
@@ -169,9 +173,10 @@ void AckGroupingTrackerEnabled::scheduleTimer() {
std::lock_guard<std::mutex> lock(this->mutexTimer_);
this->timer_ = this->executor_->createDeadlineTimer();
this->timer_->expires_from_now(std::chrono::milliseconds(std::max(1L, this->ackGroupingTimeMs_)));
- auto self = shared_from_this();
- this->timer_->async_wait([this, self](const ASIO_ERROR& ec) -> void {
- if (!ec) {
+ std::weak_ptr<AckGroupingTracker> weakSelf = shared_from_this();
+ this->timer_->async_wait([this, weakSelf](const ASIO_ERROR& ec) -> void {
+ auto self = weakSelf.lock();
+ if (self && !ec) {
this->flush();
this->scheduleTimer();
}
diff --git a/lib/AckGroupingTrackerEnabled.h b/lib/AckGroupingTrackerEnabled.h
index b04f405..5eb04b9 100644
--- a/lib/AckGroupingTrackerEnabled.h
+++ b/lib/AckGroupingTrackerEnabled.h
@@ -45,8 +45,8 @@ using HandlerBaseWeakPtr = std::weak_ptr<HandlerBase>;
*/
class AckGroupingTrackerEnabled : public AckGroupingTracker {
public:
- AckGroupingTrackerEnabled(std::function<ClientConnectionPtr()> connectionSupplier,
- std::function<uint64_t()> requestIdSupplier, uint64_t consumerId,
+ AckGroupingTrackerEnabled(const std::function<ClientConnectionPtr()>& connectionSupplier,
+ const std::function<uint64_t()>& requestIdSupplier, uint64_t consumerId,
bool waitResponse, long ackGroupingTimeMs, long ackGroupingMaxSize,
const ExecutorServicePtr& executor)
: AckGroupingTracker(connectionSupplier, requestIdSupplier, consumerId, waitResponse),
@@ -56,15 +56,14 @@ class AckGroupingTrackerEnabled : public AckGroupingTracker {
pendingIndividualCallbacks_.reserve(ackGroupingMaxSize);
}
- virtual ~AckGroupingTrackerEnabled() { this->close(); }
+ ~AckGroupingTrackerEnabled();
void start() override;
bool isDuplicate(const MessageId& msgId) override;
- void addAcknowledge(const MessageId& msgId, ResultCallback callback) override;
- void addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) override;
- void addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) override;
- void close() override;
- void flush() override;
+ void addAcknowledge(const MessageId& msgId, const ResultCallback& callback) override;
+ void addAcknowledgeList(const MessageIdList& msgIds, const ResultCallback& callback) override;
+ void addAcknowledgeCumulative(const MessageId& msgId, const ResultCallback& callback) override;
+ void flush();
void flushAndClean() override;
protected:
diff --git a/lib/BinaryProtoLookupService.cc b/lib/BinaryProtoLookupService.cc
index 489d8a2..a110e96 100644
--- a/lib/BinaryProtoLookupService.cc
+++ b/lib/BinaryProtoLookupService.cc
@@ -117,7 +117,7 @@ Future<Result, LookupDataResultPtr> BinaryProtoLookupService::getPartitionMetada
void BinaryProtoLookupService::sendPartitionMetadataLookupRequest(const std::string& topicName, Result result,
const ClientConnectionWeakPtr& clientCnx,
- LookupDataResultPromisePtr promise) {
+ const LookupDataResultPromisePtr& promise) {
if (result != ResultOk) {
promise->setFailed(result);
return;
@@ -136,9 +136,9 @@ void BinaryProtoLookupService::sendPartitionMetadataLookupRequest(const std::str
}
void BinaryProtoLookupService::handlePartitionMetadataLookup(const std::string& topicName, Result result,
- LookupDataResultPtr data,
+ const LookupDataResultPtr& data,
const ClientConnectionWeakPtr& clientCnx,
- LookupDataResultPromisePtr promise) {
+ const LookupDataResultPromisePtr& promise) {
if (data) {
LOG_DEBUG("PartitionMetadataLookup response for " << topicName << ", lookup-broker-url "
<< data->getBrokerUrl());
@@ -185,7 +185,7 @@ Future<Result, SchemaInfo> BinaryProtoLookupService::getSchema(const TopicNamePt
void BinaryProtoLookupService::sendGetSchemaRequest(const std::string& topicName, const std::string& version,
Result result, const ClientConnectionWeakPtr& clientCnx,
- GetSchemaPromisePtr promise) {
+ const GetSchemaPromisePtr& promise) {
if (result != ResultOk) {
promise->setFailed(result);
return;
@@ -197,7 +197,7 @@ void BinaryProtoLookupService::sendGetSchemaRequest(const std::string& topicName
<< " version: " << version);
conn->newGetSchema(topicName, version, requestId)
- .addListener([promise](Result result, SchemaInfo schemaInfo) {
+ .addListener([promise](Result result, const SchemaInfo& schemaInfo) {
if (result != ResultOk) {
promise->setFailed(result);
return;
@@ -210,7 +210,7 @@ void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string
CommandGetTopicsOfNamespace_Mode mode,
Result result,
const ClientConnectionWeakPtr& clientCnx,
- NamespaceTopicsPromisePtr promise) {
+ const NamespaceTopicsPromisePtr& promise) {
if (result != ResultOk) {
promise->setFailed(result);
return;
@@ -228,8 +228,9 @@ void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string
std::placeholders::_1, std::placeholders::_2, promise));
}
-void BinaryProtoLookupService::getTopicsOfNamespaceListener(Result result, NamespaceTopicsPtr topicsPtr,
- NamespaceTopicsPromisePtr promise) {
+void BinaryProtoLookupService::getTopicsOfNamespaceListener(Result result,
+ const NamespaceTopicsPtr& topicsPtr,
+ const NamespaceTopicsPromisePtr& promise) {
if (result != ResultOk) {
promise->setFailed(ResultLookupError);
return;
diff --git a/lib/BinaryProtoLookupService.h b/lib/BinaryProtoLookupService.h
index 6132825..948c7f1 100644
--- a/lib/BinaryProtoLookupService.h
+++ b/lib/BinaryProtoLookupService.h
@@ -72,21 +72,22 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
void sendPartitionMetadataLookupRequest(const std::string& topicName, Result result,
const ClientConnectionWeakPtr& clientCnx,
- LookupDataResultPromisePtr promise);
+ const LookupDataResultPromisePtr& promise);
- void handlePartitionMetadataLookup(const std::string& topicName, Result result, LookupDataResultPtr data,
+ void handlePartitionMetadataLookup(const std::string& topicName, Result result,
+ const LookupDataResultPtr& data,
const ClientConnectionWeakPtr& clientCnx,
- LookupDataResultPromisePtr promise);
+ const LookupDataResultPromisePtr& promise);
void sendGetTopicsOfNamespaceRequest(const std::string& nsName, CommandGetTopicsOfNamespace_Mode mode,
Result result, const ClientConnectionWeakPtr& clientCnx,
- NamespaceTopicsPromisePtr promise);
+ const NamespaceTopicsPromisePtr& promise);
void sendGetSchemaRequest(const std::string& topicName, const std::string& version, Result result,
- const ClientConnectionWeakPtr& clientCnx, GetSchemaPromisePtr promise);
+ const ClientConnectionWeakPtr& clientCnx, const GetSchemaPromisePtr& promise);
- void getTopicsOfNamespaceListener(Result result, NamespaceTopicsPtr topicsPtr,
- NamespaceTopicsPromisePtr promise);
+ void getTopicsOfNamespaceListener(Result result, const NamespaceTopicsPtr& topicsPtr,
+ const NamespaceTopicsPromisePtr& promise);
uint64_t newRequestId();
};
diff --git a/lib/BrokerConsumerStats.cc b/lib/BrokerConsumerStats.cc
index b2eaa8f..c3be31e 100644
--- a/lib/BrokerConsumerStats.cc
+++ b/lib/BrokerConsumerStats.cc
@@ -22,7 +22,8 @@
#include "BrokerConsumerStatsImplBase.h"
namespace pulsar {
-BrokerConsumerStats::BrokerConsumerStats(std::shared_ptr<BrokerConsumerStatsImplBase> impl) : impl_(impl) {}
+BrokerConsumerStats::BrokerConsumerStats(std::shared_ptr<BrokerConsumerStatsImplBase> impl)
+ : impl_(std::move(impl)) {}
std::shared_ptr<BrokerConsumerStatsImplBase> BrokerConsumerStats::getImpl() const { return impl_; }
diff --git a/lib/BrokerConsumerStatsImpl.cc b/lib/BrokerConsumerStatsImpl.cc
index 2f7b6ad..2ea6ad1 100644
--- a/lib/BrokerConsumerStatsImpl.cc
+++ b/lib/BrokerConsumerStatsImpl.cc
@@ -33,12 +33,12 @@ BrokerConsumerStatsImpl::BrokerConsumerStatsImpl(double msgRateOut, double msgTh
: msgRateOut_(msgRateOut),
msgThroughputOut_(msgThroughputOut),
msgRateRedeliver_(msgRateRedeliver),
- consumerName_(consumerName),
+ consumerName_(std::move(consumerName)),
availablePermits_(availablePermits),
unackedMessages_(unackedMessages),
blockedConsumerOnUnackedMsgs_(blockedConsumerOnUnackedMsgs),
- address_(address),
- connectedSince_(connectedSince),
+ address_(std::move(address)),
+ connectedSince_(std::move(connectedSince)),
type_(convertStringToConsumerType(type)),
msgRateExpired_(msgRateExpired),
msgBacklog_(msgBacklog) {}
diff --git a/lib/Client.cc b/lib/Client.cc
index 8e916c5..39a5948 100644
--- a/lib/Client.cc
+++ b/lib/Client.cc
@@ -33,7 +33,7 @@ DECLARE_LOG_OBJECT()
namespace pulsar {
-Client::Client(const std::shared_ptr<ClientImpl> impl) : impl_(impl) {}
+Client::Client(const std::shared_ptr<ClientImpl>& impl) : impl_(impl) {}
Client::Client(const std::string& serviceUrl)
: impl_(std::make_shared<ClientImpl>(serviceUrl, ClientConfiguration())) {}
@@ -54,12 +54,12 @@ Result Client::createProducer(const std::string& topic, const ProducerConfigurat
return future.get(producer);
}
-void Client::createProducerAsync(const std::string& topic, CreateProducerCallback callback) {
+void Client::createProducerAsync(const std::string& topic, const CreateProducerCallback& callback) {
createProducerAsync(topic, ProducerConfiguration(), callback);
}
-void Client::createProducerAsync(const std::string& topic, ProducerConfiguration conf,
- CreateProducerCallback callback) {
+void Client::createProducerAsync(const std::string& topic, const ProducerConfiguration& conf,
+ const CreateProducerCallback& callback) {
impl_->createProducerAsync(topic, conf, callback);
}
@@ -77,12 +77,12 @@ Result Client::subscribe(const std::string& topic, const std::string& subscripti
}
void Client::subscribeAsync(const std::string& topic, const std::string& subscriptionName,
- SubscribeCallback callback) {
+ const SubscribeCallback& callback) {
subscribeAsync(topic, subscriptionName, ConsumerConfiguration(), callback);
}
void Client::subscribeAsync(const std::string& topic, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback) {
LOG_INFO("Subscribing on Topic :" << topic);
impl_->subscribeAsync(topic, subscriptionName, conf, callback);
}
@@ -102,12 +102,12 @@ Result Client::subscribe(const std::vector<std::string>& topics, const std::stri
}
void Client::subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
- SubscribeCallback callback) {
+ const SubscribeCallback& callback) {
subscribeAsync(topics, subscriptionName, ConsumerConfiguration(), callback);
}
void Client::subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback) {
impl_->subscribeAsync(topics, subscriptionName, conf, callback);
}
@@ -126,12 +126,12 @@ Result Client::subscribeWithRegex(const std::string& regexPattern, const std::st
}
void Client::subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
- SubscribeCallback callback) {
+ const SubscribeCallback& callback) {
subscribeWithRegexAsync(regexPattern, subscriptionName, ConsumerConfiguration(), callback);
}
void Client::subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback) {
impl_->subscribeWithRegexAsync(regexPattern, subscriptionName, conf, callback);
}
@@ -145,7 +145,7 @@ Result Client::createReader(const std::string& topic, const MessageId& startMess
}
void Client::createReaderAsync(const std::string& topic, const MessageId& startMessageId,
- const ReaderConfiguration& conf, ReaderCallback callback) {
+ const ReaderConfiguration& conf, const ReaderCallback& callback) {
impl_->createReaderAsync(topic, startMessageId, conf, callback);
}
@@ -159,8 +159,8 @@ Result Client::createTableView(const std::string& topic, const TableViewConfigur
}
void Client::createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
- TableViewCallback callBack) {
- impl_->createTableViewAsync(topic, conf, callBack);
+ const TableViewCallback& callback) {
+ impl_->createTableViewAsync(topic, conf, callback);
}
Result Client::getPartitionsForTopic(const std::string& topic, std::vector<std::string>& partitions) {
@@ -171,7 +171,7 @@ Result Client::getPartitionsForTopic(const std::string& topic, std::vector<std::
return future.get(partitions);
}
-void Client::getPartitionsForTopicAsync(const std::string& topic, GetPartitionsCallback callback) {
+void Client::getPartitionsForTopicAsync(const std::string& topic, const GetPartitionsCallback& callback) {
impl_->getPartitionsForTopicAsync(topic, callback);
}
@@ -184,7 +184,7 @@ Result Client::close() {
return result;
}
-void Client::closeAsync(CloseCallback callback) { impl_->closeAsync(callback); }
+void Client::closeAsync(const CloseCallback& callback) { impl_->closeAsync(callback); }
void Client::shutdown() { impl_->shutdown(); }
@@ -195,6 +195,6 @@ void Client::getSchemaInfoAsync(const std::string& topic, int64_t version,
std::function<void(Result, const SchemaInfo&)> callback) {
impl_->getLookup()
->getSchema(TopicName::get(topic), (version >= 0) ? toBigEndianBytes(version) : "")
- .addListener(callback);
+ .addListener(std::move(callback));
}
} // namespace pulsar
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index d036491..9c7537a 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -160,7 +160,7 @@ static bool file_exists(const std::string& path) {
std::atomic<int32_t> ClientConnection::maxMessageSize_{Commands::DefaultMaxMessageSize};
ClientConnection::ClientConnection(const std::string& logicalAddress, const std::string& physicalAddress,
- ExecutorServicePtr executor,
+ const ExecutorServicePtr& executor,
const ClientConfiguration& clientConfiguration,
const AuthenticationPtr& authentication, const std::string& clientVersion,
ConnectionPool& pool, size_t poolIndex)
@@ -223,7 +223,7 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
} else {
ctx.set_verify_mode(ASIO::ssl::context::verify_peer);
- std::string trustCertFilePath = clientConfiguration.getTlsTrustCertsFilePath();
+ const auto& trustCertFilePath = clientConfiguration.getTlsTrustCertsFilePath();
if (!trustCertFilePath.empty()) {
if (file_exists(trustCertFilePath)) {
ctx.load_verify_file(trustCertFilePath);
@@ -605,15 +605,16 @@ void ClientConnection::tcpConnectAsync() {
LOG_DEBUG(cnxString_ << "Resolving " << service_url.host() << ":" << service_url.port());
tcp::resolver::query query(service_url.host(), std::to_string(service_url.port()));
auto weakSelf = weak_from_this();
- resolver_->async_resolve(query, [weakSelf](const ASIO_ERROR& err, tcp::resolver::iterator iterator) {
- auto self = weakSelf.lock();
- if (self) {
- self->handleResolve(err, iterator);
- }
- });
+ resolver_->async_resolve(query,
+ [weakSelf](const ASIO_ERROR& err, const tcp::resolver::iterator& iterator) {
+ auto self = weakSelf.lock();
+ if (self) {
+ self->handleResolve(err, iterator);
+ }
+ });
}
-void ClientConnection::handleResolve(const ASIO_ERROR& err, tcp::resolver::iterator endpointIterator) {
+void ClientConnection::handleResolve(const ASIO_ERROR& err, const tcp::resolver::iterator& endpointIterator) {
if (err) {
std::string hostUrl = isSniProxy_ ? cnxString_ : proxyServiceUrl_;
LOG_ERROR(hostUrl << "Resolve error: " << err << " : " << err.message());
@@ -1033,18 +1034,18 @@ Future<Result, BrokerConsumerStatsImpl> ClientConnection::newConsumerStats(uint6
}
void ClientConnection::newTopicLookup(const std::string& topicName, bool authoritative,
- const std::string& listenerName, const uint64_t requestId,
- LookupDataResultPromisePtr promise) {
+ const std::string& listenerName, uint64_t requestId,
+ const LookupDataResultPromisePtr& promise) {
newLookup(Commands::newLookup(topicName, authoritative, requestId, listenerName), requestId, promise);
}
-void ClientConnection::newPartitionedMetadataLookup(const std::string& topicName, const uint64_t requestId,
- LookupDataResultPromisePtr promise) {
+void ClientConnection::newPartitionedMetadataLookup(const std::string& topicName, uint64_t requestId,
+ const LookupDataResultPromisePtr& promise) {
newLookup(Commands::newPartitionMetadataRequest(topicName, requestId), requestId, promise);
}
-void ClientConnection::newLookup(const SharedBuffer& cmd, const uint64_t requestId,
- LookupDataResultPromisePtr promise) {
+void ClientConnection::newLookup(const SharedBuffer& cmd, uint64_t requestId,
+ const LookupDataResultPromisePtr& promise) {
Lock lock(mutex_);
std::shared_ptr<LookupDataResultPtr> lookupDataResult;
lookupDataResult = std::make_shared<LookupDataResultPtr>();
@@ -1188,7 +1189,7 @@ void ClientConnection::sendPendingCommands() {
}
}
-Future<Result, ResponseData> ClientConnection::sendRequestWithId(SharedBuffer cmd, int requestId) {
+Future<Result, ResponseData> ClientConnection::sendRequestWithId(const SharedBuffer& cmd, int requestId) {
Lock lock(mutex_);
if (isClosed()) {
@@ -1216,20 +1217,22 @@ Future<Result, ResponseData> ClientConnection::sendRequestWithId(SharedBuffer cm
return requestData.promise.getFuture();
}
-void ClientConnection::handleRequestTimeout(const ASIO_ERROR& ec, PendingRequestData pendingRequestData) {
+void ClientConnection::handleRequestTimeout(const ASIO_ERROR& ec,
+ const PendingRequestData& pendingRequestData) {
if (!ec && !pendingRequestData.hasGotResponse->load()) {
pendingRequestData.promise.setFailed(ResultTimeout);
}
}
-void ClientConnection::handleLookupTimeout(const ASIO_ERROR& ec, LookupRequestData pendingRequestData) {
+void ClientConnection::handleLookupTimeout(const ASIO_ERROR& ec,
+ const LookupRequestData& pendingRequestData) {
if (!ec) {
pendingRequestData.promise->setFailed(ResultTimeout);
}
}
void ClientConnection::handleGetLastMessageIdTimeout(const ASIO_ERROR& ec,
- ClientConnection::LastMessageIdRequestData data) {
+ const ClientConnection::LastMessageIdRequestData& data) {
if (!ec) {
data.promise->setFailed(ResultTimeout);
}
@@ -1267,7 +1270,7 @@ void ClientConnection::handleKeepAliveTimeout() {
}
void ClientConnection::handleConsumerStatsTimeout(const ASIO_ERROR& ec,
- std::vector<uint64_t> consumerStatsRequests) {
+ const std::vector<uint64_t>& consumerStatsRequests) {
if (ec) {
LOG_DEBUG(cnxString_ << " Ignoring timer cancelled event, code[" << ec << "]");
return;
@@ -1386,12 +1389,12 @@ Future<Result, ClientConnectionWeakPtr> ClientConnection::getConnectFuture() {
return connectPromise_.getFuture();
}
-void ClientConnection::registerProducer(int producerId, ProducerImplPtr producer) {
+void ClientConnection::registerProducer(int producerId, const ProducerImplPtr& producer) {
Lock lock(mutex_);
producers_.insert(std::make_pair(producerId, producer));
}
-void ClientConnection::registerConsumer(int consumerId, ConsumerImplPtr consumer) {
+void ClientConnection::registerConsumer(int consumerId, const ConsumerImplPtr& consumer) {
Lock lock(mutex_);
consumers_.insert(std::make_pair(consumerId, consumer));
}
diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h
index 7646f85..ff40281 100644
--- a/lib/ClientConnection.h
+++ b/lib/ClientConnection.h
@@ -23,6 +23,7 @@
#include <pulsar/defines.h>
#include <atomic>
+#include <cstdint>
#ifdef USE_ASIO
#include <asio/bind_executor.hpp>
#include <asio/io_service.hpp>
@@ -113,7 +114,7 @@ struct ResponseData {
typedef std::shared_ptr<std::vector<std::string>> NamespaceTopicsPtr;
class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<ClientConnection> {
- enum State
+ enum State : uint8_t
{
Pending,
TcpConnected,
@@ -135,7 +136,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
*
*/
ClientConnection(const std::string& logicalAddress, const std::string& physicalAddress,
- ExecutorServicePtr executor, const ClientConfiguration& clientConfiguration,
+ const ExecutorServicePtr& executor, const ClientConfiguration& clientConfiguration,
const AuthenticationPtr& authentication, const std::string& clientVersion,
ConnectionPool& pool, size_t poolIndex);
~ClientConnection();
@@ -167,17 +168,17 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
Future<Result, ClientConnectionWeakPtr> getCloseFuture();
void newTopicLookup(const std::string& topicName, bool authoritative, const std::string& listenerName,
- const uint64_t requestId, LookupDataResultPromisePtr promise);
+ uint64_t requestId, const LookupDataResultPromisePtr& promise);
- void newPartitionedMetadataLookup(const std::string& topicName, const uint64_t requestId,
- LookupDataResultPromisePtr promise);
+ void newPartitionedMetadataLookup(const std::string& topicName, uint64_t requestId,
+ const LookupDataResultPromisePtr& promise);
void sendCommand(const SharedBuffer& cmd);
void sendCommandInternal(const SharedBuffer& cmd);
void sendMessage(const std::shared_ptr<SendArguments>& args);
- void registerProducer(int producerId, ProducerImplPtr producer);
- void registerConsumer(int consumerId, ConsumerImplPtr consumer);
+ void registerProducer(int producerId, const ProducerImplPtr& producer);
+ void registerConsumer(int consumerId, const ConsumerImplPtr& consumer);
void removeProducer(int producerId);
void removeConsumer(int consumerId);
@@ -186,7 +187,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
* Send a request with a specific Id over the connection. The future will be
* triggered when the response for this request is received
*/
- Future<Result, ResponseData> sendRequestWithId(SharedBuffer cmd, int requestId);
+ Future<Result, ResponseData> sendRequestWithId(const SharedBuffer& cmd, int requestId);
const std::string& brokerAddress() const;
@@ -260,18 +261,18 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
void handlePulsarConnected(const proto::CommandConnected& cmdConnected);
- void handleResolve(const ASIO_ERROR& err, ASIO::ip::tcp::resolver::iterator endpointIterator);
+ void handleResolve(const ASIO_ERROR& err, const ASIO::ip::tcp::resolver::iterator& endpointIterator);
void handleSend(const ASIO_ERROR& err, const SharedBuffer& cmd);
void handleSendPair(const ASIO_ERROR& err);
void sendPendingCommands();
- void newLookup(const SharedBuffer& cmd, const uint64_t requestId, LookupDataResultPromisePtr promise);
+ void newLookup(const SharedBuffer& cmd, uint64_t requestId, const LookupDataResultPromisePtr& promise);
- void handleRequestTimeout(const ASIO_ERROR& ec, PendingRequestData pendingRequestData);
+ void handleRequestTimeout(const ASIO_ERROR& ec, const PendingRequestData& pendingRequestData);
- void handleLookupTimeout(const ASIO_ERROR&, LookupRequestData);
+ void handleLookupTimeout(const ASIO_ERROR&, const LookupRequestData&);
- void handleGetLastMessageIdTimeout(const ASIO_ERROR&, LastMessageIdRequestData data);
+ void handleGetLastMessageIdTimeout(const ASIO_ERROR&, const LastMessageIdRequestData& data);
void handleKeepAliveTimeout();
@@ -392,7 +393,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
DeadlineTimerPtr keepAliveTimer_;
DeadlineTimerPtr consumerStatsRequestTimer_;
- void handleConsumerStatsTimeout(const ASIO_ERROR& ec, std::vector<uint64_t> consumerStatsRequests);
+ void handleConsumerStatsTimeout(const ASIO_ERROR& ec, const std::vector<uint64_t>& consumerStatsRequests);
void startConsumerStatsTimer(std::vector<uint64_t> consumerStatsRequests);
uint32_t maxPendingLookupRequest_;
diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc
index a6f8b5f..8298ebc 100644
--- a/lib/ClientImpl.cc
+++ b/lib/ClientImpl.cc
@@ -148,8 +148,8 @@ LookupServicePtr ClientImpl::getLookup(const std::string& redirectedClusterURI)
return it->second;
}
-void ClientImpl::createProducerAsync(const std::string& topic, ProducerConfiguration conf,
- CreateProducerCallback callback, bool autoDownloadSchema) {
+void ClientImpl::createProducerAsync(const std::string& topic, const ProducerConfiguration& conf,
+ const CreateProducerCallback& callback, bool autoDownloadSchema) {
if (conf.isChunkingEnabled() && conf.getBatchingEnabled()) {
throw std::invalid_argument("Batching and chunking of messages can't be enabled together");
}
@@ -170,7 +170,7 @@ void ClientImpl::createProducerAsync(const std::string& topic, ProducerConfigura
if (autoDownloadSchema) {
auto self = shared_from_this();
lookupServicePtr_->getSchema(topicName).addListener(
- [self, topicName, callback](Result res, SchemaInfo topicSchema) {
+ [self, topicName, callback](Result res, const SchemaInfo& topicSchema) {
if (res != ResultOk) {
callback(res, Producer());
return;
@@ -188,9 +188,9 @@ void ClientImpl::createProducerAsync(const std::string& topic, ProducerConfigura
}
}
-void ClientImpl::handleCreateProducer(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, ProducerConfiguration conf,
- CreateProducerCallback callback) {
+void ClientImpl::handleCreateProducer(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const ProducerConfiguration& conf,
+ const CreateProducerCallback& callback) {
if (!result) {
ProducerImplBasePtr producer;
@@ -219,8 +219,9 @@ void ClientImpl::handleCreateProducer(const Result result, const LookupDataResul
}
}
-void ClientImpl::handleProducerCreated(Result result, ProducerImplBaseWeakPtr producerBaseWeakPtr,
- CreateProducerCallback callback, ProducerImplBasePtr producer) {
+void ClientImpl::handleProducerCreated(Result result, const ProducerImplBaseWeakPtr& producerBaseWeakPtr,
+ const CreateProducerCallback& callback,
+ const ProducerImplBasePtr& producer) {
if (result == ResultOk) {
auto address = producer.get();
auto existingProducer = producers_.putIfAbsent(address, producer);
@@ -238,7 +239,7 @@ void ClientImpl::handleProducerCreated(Result result, ProducerImplBaseWeakPtr pr
}
void ClientImpl::createReaderAsync(const std::string& topic, const MessageId& startMessageId,
- const ReaderConfiguration& conf, ReaderCallback callback) {
+ const ReaderConfiguration& conf, const ReaderCallback& callback) {
TopicNamePtr topicName;
{
Lock lock(mutex_);
@@ -260,7 +261,7 @@ void ClientImpl::createReaderAsync(const std::string& topic, const MessageId& st
}
void ClientImpl::createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
- TableViewCallback callback) {
+ const TableViewCallback& callback) {
TopicNamePtr topicName;
{
Lock lock(mutex_);
@@ -277,7 +278,7 @@ void ClientImpl::createTableViewAsync(const std::string& topic, const TableViewC
TableViewImplPtr tableViewPtr =
std::make_shared<TableViewImpl>(shared_from_this(), topicName->toString(), conf);
- tableViewPtr->start().addListener([callback](Result result, TableViewImplPtr tableViewImplPtr) {
+ tableViewPtr->start().addListener([callback](Result result, const TableViewImplPtr& tableViewImplPtr) {
if (result == ResultOk) {
callback(result, TableView{tableViewImplPtr});
} else {
@@ -286,9 +287,9 @@ void ClientImpl::createTableViewAsync(const std::string& topic, const TableViewC
});
}
-void ClientImpl::handleReaderMetadataLookup(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, MessageId startMessageId,
- ReaderConfiguration conf, ReaderCallback callback) {
+void ClientImpl::handleReaderMetadataLookup(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const MessageId& startMessageId,
+ const ReaderConfiguration& conf, const ReaderCallback& callback) {
if (result != ResultOk) {
LOG_ERROR("Error Checking/Getting Partition Metadata while creating readeron "
<< topicName->toString() << " -- " << result);
@@ -325,7 +326,8 @@ void ClientImpl::handleReaderMetadataLookup(const Result result, const LookupDat
}
void ClientImpl::subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ const ConsumerConfiguration& conf,
+ const SubscribeCallback& callback) {
TopicNamePtr topicNamePtr = TopicName::get(regexPattern);
Lock lock(mutex_);
@@ -372,12 +374,12 @@ void ClientImpl::subscribeWithRegexAsync(const std::string& regexPattern, const
subscriptionName, conf, callback));
}
-void ClientImpl::createPatternMultiTopicsConsumer(const Result result, const NamespaceTopicsPtr topics,
+void ClientImpl::createPatternMultiTopicsConsumer(Result result, const NamespaceTopicsPtr& topics,
const std::string& regexPattern,
CommandGetTopicsOfNamespace_Mode mode,
const std::string& subscriptionName,
const ConsumerConfiguration& conf,
- SubscribeCallback callback) {
+ const SubscribeCallback& callback) {
if (result == ResultOk) {
ConsumerImplBasePtr consumer;
@@ -403,7 +405,7 @@ void ClientImpl::createPatternMultiTopicsConsumer(const Result result, const Nam
}
void ClientImpl::subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback) {
TopicNamePtr topicNamePtr;
Lock lock(mutex_);
@@ -439,7 +441,7 @@ void ClientImpl::subscribeAsync(const std::vector<std::string>& topics, const st
}
void ClientImpl::subscribeAsync(const std::string& topic, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback) {
TopicNamePtr topicName;
{
Lock lock(mutex_);
@@ -465,9 +467,9 @@ void ClientImpl::subscribeAsync(const std::string& topic, const std::string& sub
std::placeholders::_2, topicName, subscriptionName, conf, callback));
}
-void ClientImpl::handleSubscribe(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, const std::string& subscriptionName,
- ConsumerConfiguration conf, SubscribeCallback callback) {
+void ClientImpl::handleSubscribe(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const std::string& subscriptionName,
+ ConsumerConfiguration conf, const SubscribeCallback& callback) {
if (result == ResultOk) {
// generate random name if not supplied by the customer.
if (conf.getConsumerName().empty()) {
@@ -509,8 +511,9 @@ void ClientImpl::handleSubscribe(const Result result, const LookupDataResultPtr
}
}
-void ClientImpl::handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr consumerImplBaseWeakPtr,
- SubscribeCallback callback, ConsumerImplBasePtr consumer) {
+void ClientImpl::handleConsumerCreated(Result result, const ConsumerImplBaseWeakPtr& consumerImplBaseWeakPtr,
+ const SubscribeCallback& callback,
+ const ConsumerImplBasePtr& consumer) {
if (result == ResultOk) {
auto address = consumer.get();
auto existingConsumer = consumers_.putIfAbsent(address, consumer);
@@ -602,8 +605,8 @@ GetConnectionFuture ClientImpl::connect(const std::string& redirectedClusterURI,
return promise.getFuture();
}
-void ClientImpl::handleGetPartitions(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, GetPartitionsCallback callback) {
+void ClientImpl::handleGetPartitions(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const GetPartitionsCallback& callback) {
if (result != ResultOk) {
LOG_ERROR("Error getting topic partitions metadata: " << result);
callback(result, StringList());
@@ -623,7 +626,7 @@ void ClientImpl::handleGetPartitions(const Result result, const LookupDataResult
callback(ResultOk, partitions);
}
-void ClientImpl::getPartitionsForTopicAsync(const std::string& topic, GetPartitionsCallback callback) {
+void ClientImpl::getPartitionsForTopicAsync(const std::string& topic, const GetPartitionsCallback& callback) {
TopicNamePtr topicName;
{
Lock lock(mutex_);
@@ -642,7 +645,7 @@ void ClientImpl::getPartitionsForTopicAsync(const std::string& topic, GetPartiti
std::placeholders::_2, topicName, callback));
}
-void ClientImpl::closeAsync(CloseCallback callback) {
+void ClientImpl::closeAsync(const CloseCallback& callback) {
if (state_ != Open) {
if (callback) {
callback(ResultAlreadyClosed);
@@ -693,7 +696,8 @@ void ClientImpl::closeAsync(CloseCallback callback) {
lookupCount_ = 0;
}
-void ClientImpl::handleClose(Result result, SharedInt numberOfOpenHandlers, ResultCallback callback) {
+void ClientImpl::handleClose(Result result, const SharedInt& numberOfOpenHandlers,
+ const ResultCallback& callback) {
Result expected = ResultOk;
if (!closingError.compare_exchange_strong(expected, result)) {
LOG_DEBUG("Tried to updated closingError, but already set to "
@@ -753,6 +757,7 @@ void ClientImpl::shutdown() {
LOG_DEBUG(producers.size() << " producers and " << consumers.size()
<< " consumers have been shutdown.");
}
+
if (!pool_.close()) {
// pool_ has already been closed. It means shutdown() has been called before.
return;
diff --git a/lib/ClientImpl.h b/lib/ClientImpl.h
index 27cde3a..000e443 100644
--- a/lib/ClientImpl.h
+++ b/lib/ClientImpl.h
@@ -22,6 +22,7 @@
#include <pulsar/Client.h>
#include <atomic>
+#include <cstdint>
#include <memory>
#include "ConnectionPool.h"
@@ -29,7 +30,6 @@
#include "LookupDataResult.h"
#include "MemoryLimitController.h"
#include "ProtoApiEnums.h"
-#include "ServiceNameResolver.h"
#include "SynchronizedHashMap.h"
namespace pulsar {
@@ -76,25 +76,25 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
* @param autoDownloadSchema When it is true, Before creating a producer, it will try to get the schema
* that exists for the topic.
*/
- void createProducerAsync(const std::string& topic, ProducerConfiguration conf,
- CreateProducerCallback callback, bool autoDownloadSchema = false);
+ void createProducerAsync(const std::string& topic, const ProducerConfiguration& conf,
+ const CreateProducerCallback& callback, bool autoDownloadSchema = false);
void subscribeAsync(const std::string& topic, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback);
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback);
void subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback);
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback);
void subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, SubscribeCallback callback);
+ const ConsumerConfiguration& conf, const SubscribeCallback& callback);
void createReaderAsync(const std::string& topic, const MessageId& startMessageId,
- const ReaderConfiguration& conf, ReaderCallback callback);
+ const ReaderConfiguration& conf, const ReaderCallback& callback);
void createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
- TableViewCallback callback);
+ const TableViewCallback& callback);
- void getPartitionsForTopicAsync(const std::string& topic, GetPartitionsCallback callback);
+ void getPartitionsForTopicAsync(const std::string& topic, const GetPartitionsCallback& callback);
// Use virtual method to test
virtual GetConnectionFuture getConnection(const std::string& redirectedClusterURI,
@@ -103,7 +103,7 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
GetConnectionFuture connect(const std::string& redirectedClusterURI, const std::string& logicalAddress,
size_t key);
- void closeAsync(CloseCallback callback);
+ void closeAsync(const CloseCallback& callback);
void shutdown();
MemoryLimitController& getMemoryLimitController();
@@ -137,35 +137,35 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
friend class PulsarFriend;
private:
- void handleCreateProducer(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, ProducerConfiguration conf,
- CreateProducerCallback callback);
+ void handleCreateProducer(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const ProducerConfiguration& conf,
+ const CreateProducerCallback& callback);
- void handleSubscribe(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, const std::string& consumerName, ConsumerConfiguration conf,
- SubscribeCallback callback);
+ void handleSubscribe(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const std::string& consumerName,
+ ConsumerConfiguration conf, const SubscribeCallback& callback);
- void handleReaderMetadataLookup(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, MessageId startMessageId,
- ReaderConfiguration conf, ReaderCallback callback);
+ void handleReaderMetadataLookup(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const MessageId& startMessageId,
+ const ReaderConfiguration& conf, const ReaderCallback& callback);
- void handleGetPartitions(const Result result, const LookupDataResultPtr partitionMetadata,
- TopicNamePtr topicName, GetPartitionsCallback callback);
+ void handleGetPartitions(Result result, const LookupDataResultPtr& partitionMetadata,
+ const TopicNamePtr& topicName, const GetPartitionsCallback& callback);
- void handleProducerCreated(Result result, ProducerImplBaseWeakPtr producerWeakPtr,
- CreateProducerCallback callback, ProducerImplBasePtr producer);
- void handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr consumerWeakPtr,
- SubscribeCallback callback, ConsumerImplBasePtr consumer);
+ void handleProducerCreated(Result result, const ProducerImplBaseWeakPtr& producerWeakPtr,
+ const CreateProducerCallback& callback, const ProducerImplBasePtr& producer);
+ void handleConsumerCreated(Result result, const ConsumerImplBaseWeakPtr& consumerWeakPtr,
+ const SubscribeCallback& callback, const ConsumerImplBasePtr& consumer);
typedef std::shared_ptr<int> SharedInt;
- void handleClose(Result result, SharedInt remaining, ResultCallback callback);
+ void handleClose(Result result, const SharedInt& remaining, const ResultCallback& callback);
- void createPatternMultiTopicsConsumer(const Result result, const NamespaceTopicsPtr topics,
+ void createPatternMultiTopicsConsumer(Result result, const NamespaceTopicsPtr& topics,
const std::string& regexPattern,
CommandGetTopicsOfNamespace_Mode mode,
const std::string& consumerName, const ConsumerConfiguration& conf,
- SubscribeCallback callback);
+ const SubscribeCallback& callback);
const std::string& getPhysicalAddress(const std::string& redirectedClusterURI,
const std::string& logicalAddress);
@@ -174,7 +174,7 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
static std::string getClientVersion(const ClientConfiguration& clientConfiguration);
- enum State
+ enum State : uint8_t
{
Open,
Closing,
diff --git a/lib/Commands.cc b/lib/Commands.cc
index 84f272b..dd62b21 100644
--- a/lib/Commands.cc
+++ b/lib/Commands.cc
@@ -138,7 +138,7 @@ SharedBuffer Commands::newPartitionMetadataRequest(const std::string& topic, uin
CommandPartitionedTopicMetadata* partitionMetadata = cmd.mutable_partitionmetadata();
partitionMetadata->set_topic(topic);
partitionMetadata->set_request_id(requestId);
- const SharedBuffer buffer = writeMessageWithSize(cmd);
+ SharedBuffer buffer = writeMessageWithSize(cmd);
cmd.clear_partitionmetadata();
return buffer;
}
@@ -154,7 +154,7 @@ SharedBuffer Commands::newLookup(const std::string& topic, const bool authoritat
lookup->set_authoritative(authoritative);
lookup->set_request_id(requestId);
lookup->set_advertised_listener_name(listenerName);
- const SharedBuffer buffer = writeMessageWithSize(cmd);
+ SharedBuffer buffer = writeMessageWithSize(cmd);
cmd.clear_lookuptopic();
return buffer;
}
@@ -173,7 +173,7 @@ SharedBuffer Commands::newGetSchema(const std::string& topic, const std::string&
getSchema->set_schema_version(version);
}
- const SharedBuffer buffer = writeMessageWithSize(cmd);
+ SharedBuffer buffer = writeMessageWithSize(cmd);
cmd.clear_getschema();
return buffer;
}
@@ -186,7 +186,7 @@ SharedBuffer Commands::newConsumerStats(uint64_t consumerId, uint64_t requestId)
CommandConsumerStats* consumerStats = cmd.mutable_consumerstats();
consumerStats->set_consumer_id(consumerId);
consumerStats->set_request_id(requestId);
- const SharedBuffer buffer = writeMessageWithSize(cmd);
+ SharedBuffer buffer = writeMessageWithSize(cmd);
cmd.clear_consumerstats();
return buffer;
}
@@ -334,7 +334,7 @@ SharedBuffer Commands::newSubscribe(const std::string& topic, const std::string&
const std::map<std::string, std::string>& subscriptionProperties,
const SchemaInfo& schemaInfo,
CommandSubscribe_InitialPosition subscriptionInitialPosition,
- bool replicateSubscriptionState, KeySharedPolicy keySharedPolicy,
+ bool replicateSubscriptionState, const KeySharedPolicy& keySharedPolicy,
int priorityLevel) {
BaseCommand cmd;
cmd.set_type(BaseCommand::SUBSCRIBE);
@@ -618,7 +618,7 @@ SharedBuffer Commands::newGetLastMessageId(uint64_t consumerId, uint64_t request
CommandGetLastMessageId* getLastMessageId = cmd.mutable_getlastmessageid();
getLastMessageId->set_consumer_id(consumerId);
getLastMessageId->set_request_id(requestId);
- const SharedBuffer buffer = writeMessageWithSize(cmd);
+ SharedBuffer buffer = writeMessageWithSize(cmd);
cmd.clear_getlastmessageid();
return buffer;
}
@@ -632,7 +632,7 @@ SharedBuffer Commands::newGetTopicsOfNamespace(const std::string& nsName,
getTopics->set_namespace_(nsName);
getTopics->set_mode(static_cast<proto::CommandGetTopicsOfNamespace_Mode>(mode));
- const SharedBuffer buffer = writeMessageWithSize(cmd);
+ SharedBuffer buffer = writeMessageWithSize(cmd);
cmd.clear_gettopicsofnamespace();
return buffer;
}
diff --git a/lib/Commands.h b/lib/Commands.h
index de5445d..15c3166 100644
--- a/lib/Commands.h
+++ b/lib/Commands.h
@@ -109,7 +109,7 @@ class Commands {
const std::map<std::string, std::string>& metadata,
const std::map<std::string, std::string>& subscriptionProperties, const SchemaInfo& schemaInfo,
CommandSubscribe_InitialPosition subscriptionInitialPosition, bool replicateSubscriptionState,
- KeySharedPolicy keySharedPolicy, int priorityLevel = 0);
+ const KeySharedPolicy& keySharedPolicy, int priorityLevel = 0);
static SharedBuffer newUnsubscribe(uint64_t consumerId, uint64_t requestId);
diff --git a/lib/ConnectionPool.cc b/lib/ConnectionPool.cc
index 9a614ed..df050b0 100644
--- a/lib/ConnectionPool.cc
+++ b/lib/ConnectionPool.cc
@@ -38,7 +38,8 @@ DECLARE_LOG_OBJECT()
namespace pulsar {
-ConnectionPool::ConnectionPool(const ClientConfiguration& conf, ExecutorServiceProviderPtr executorProvider,
+ConnectionPool::ConnectionPool(const ClientConfiguration& conf,
+ const ExecutorServiceProviderPtr& executorProvider,
const AuthenticationPtr& authentication, const std::string& clientVersion)
: clientConfiguration_(conf),
executorProvider_(executorProvider),
diff --git a/lib/ConnectionPool.h b/lib/ConnectionPool.h
index e044d62..0e3a6d0 100644
--- a/lib/ConnectionPool.h
+++ b/lib/ConnectionPool.h
@@ -41,7 +41,7 @@ using ExecutorServiceProviderPtr = std::shared_ptr<ExecutorServiceProvider>;
class PULSAR_PUBLIC ConnectionPool {
public:
- ConnectionPool(const ClientConfiguration& conf, ExecutorServiceProviderPtr executorProvider,
+ ConnectionPool(const ClientConfiguration& conf, const ExecutorServiceProviderPtr& executorProvider,
const AuthenticationPtr& authentication, const std::string& clientVersion);
/**
diff --git a/lib/Consumer.cc b/lib/Consumer.cc
index 031cdff..2c4e8be 100644
--- a/lib/Consumer.cc
+++ b/lib/Consumer.cc
@@ -31,7 +31,7 @@ static const std::string EMPTY_STRING;
Consumer::Consumer() : impl_() {}
-Consumer::Consumer(ConsumerImplBasePtr impl) : impl_(impl) {}
+Consumer::Consumer(ConsumerImplBasePtr impl) : impl_(std::move(impl)) {}
const std::string& Consumer::getTopic() const { return impl_ != NULL ? impl_->getTopic() : EMPTY_STRING; }
@@ -54,7 +54,7 @@ Result Consumer::unsubscribe() {
return result;
}
-void Consumer::unsubscribeAsync(ResultCallback callback) {
+void Consumer::unsubscribeAsync(const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -79,7 +79,7 @@ Result Consumer::receive(Message& msg, int timeoutMs) {
return impl_->receive(msg, timeoutMs);
}
-void Consumer::receiveAsync(ReceiveCallback callback) {
+void Consumer::receiveAsync(const ReceiveCallback& callback) {
if (!impl_) {
Message msg;
callback(ResultConsumerNotInitialized, msg);
@@ -97,7 +97,7 @@ Result Consumer::batchReceive(Messages& msgs) {
return promise.getFuture().get(msgs);
}
-void Consumer::batchReceiveAsync(BatchReceiveCallback callback) {
+void Consumer::batchReceiveAsync(const BatchReceiveCallback& callback) {
if (!impl_) {
Messages msgs;
callback(ResultConsumerNotInitialized, msgs);
@@ -130,7 +130,7 @@ Result Consumer::acknowledge(const MessageIdList& messageIdList) {
return result;
}
-void Consumer::acknowledgeAsync(const Message& message, ResultCallback callback) {
+void Consumer::acknowledgeAsync(const Message& message, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -139,7 +139,7 @@ void Consumer::acknowledgeAsync(const Message& message, ResultCallback callback)
impl_->acknowledgeAsync(message.getMessageId(), callback);
}
-void Consumer::acknowledgeAsync(const MessageId& messageId, ResultCallback callback) {
+void Consumer::acknowledgeAsync(const MessageId& messageId, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -148,7 +148,7 @@ void Consumer::acknowledgeAsync(const MessageId& messageId, ResultCallback callb
impl_->acknowledgeAsync(messageId, callback);
}
-void Consumer::acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) {
+void Consumer::acknowledgeAsync(const MessageIdList& messageIdList, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -173,11 +173,11 @@ Result Consumer::acknowledgeCumulative(const MessageId& messageId) {
return result;
}
-void Consumer::acknowledgeCumulativeAsync(const Message& message, ResultCallback callback) {
+void Consumer::acknowledgeCumulativeAsync(const Message& message, const ResultCallback& callback) {
acknowledgeCumulativeAsync(message.getMessageId(), callback);
}
-void Consumer::acknowledgeCumulativeAsync(const MessageId& messageId, ResultCallback callback) {
+void Consumer::acknowledgeCumulativeAsync(const MessageId& messageId, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -204,7 +204,7 @@ Result Consumer::close() {
return result;
}
-void Consumer::closeAsync(ResultCallback callback) {
+void Consumer::closeAsync(const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -244,7 +244,7 @@ Result Consumer::getBrokerConsumerStats(BrokerConsumerStats& brokerConsumerStats
return promise.getFuture().get(brokerConsumerStats);
}
-void Consumer::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) {
+void Consumer::getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized, BrokerConsumerStats());
return;
@@ -252,7 +252,7 @@ void Consumer::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback)
impl_->getBrokerConsumerStatsAsync(callback);
}
-void Consumer::seekAsync(const MessageId& msgId, ResultCallback callback) {
+void Consumer::seekAsync(const MessageId& msgId, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -260,7 +260,7 @@ void Consumer::seekAsync(const MessageId& msgId, ResultCallback callback) {
impl_->seekAsync(msgId, callback);
}
-void Consumer::seekAsync(uint64_t timestamp, ResultCallback callback) {
+void Consumer::seekAsync(uint64_t timestamp, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -294,7 +294,7 @@ Result Consumer::seek(uint64_t timestamp) {
bool Consumer::isConnected() const { return impl_ && impl_->isConnected(); }
-void Consumer::getLastMessageIdAsync(GetLastMessageIdCallback callback) {
+void Consumer::getLastMessageIdAsync(const GetLastMessageIdCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized, MessageId());
return;
diff --git a/lib/ConsumerConfiguration.cc b/lib/ConsumerConfiguration.cc
index 9648278..60f8fea 100644
--- a/lib/ConsumerConfiguration.cc
+++ b/lib/ConsumerConfiguration.cc
@@ -66,7 +66,7 @@ ConsumerConfiguration& ConsumerConfiguration::setConsumerType(ConsumerType consu
ConsumerType ConsumerConfiguration::getConsumerType() const { return impl_->consumerType; }
ConsumerConfiguration& ConsumerConfiguration::setMessageListener(MessageListener messageListener) {
- impl_->messageListener = messageListener;
+ impl_->messageListener = std::move(messageListener);
impl_->hasMessageListener = true;
return *this;
}
@@ -77,7 +77,7 @@ bool ConsumerConfiguration::hasMessageListener() const { return impl_->hasMessag
ConsumerConfiguration& ConsumerConfiguration::setConsumerEventListener(
ConsumerEventListenerPtr eventListener) {
- impl_->eventListener = eventListener;
+ impl_->eventListener = std::move(eventListener);
impl_->hasConsumerEventListener = true;
return *this;
}
@@ -147,7 +147,7 @@ bool ConsumerConfiguration::isEncryptionEnabled() const { return (impl_->cryptoK
const CryptoKeyReaderPtr ConsumerConfiguration::getCryptoKeyReader() const { return impl_->cryptoKeyReader; }
ConsumerConfiguration& ConsumerConfiguration::setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader) {
- impl_->cryptoKeyReader = cryptoKeyReader;
+ impl_->cryptoKeyReader = std::move(cryptoKeyReader);
return *this;
}
@@ -238,7 +238,7 @@ ConsumerConfiguration& ConsumerConfiguration::setPriorityLevel(int priorityLevel
int ConsumerConfiguration::getPriorityLevel() const { return impl_->priorityLevel; }
-ConsumerConfiguration& ConsumerConfiguration::setKeySharedPolicy(KeySharedPolicy keySharedPolicy) {
+ConsumerConfiguration& ConsumerConfiguration::setKeySharedPolicy(const KeySharedPolicy& keySharedPolicy) {
impl_->keySharedPolicy = keySharedPolicy.clone();
return *this;
}
diff --git a/lib/ConsumerConfigurationImpl.h b/lib/ConsumerConfigurationImpl.h
index 54e4599..711b6f9 100644
--- a/lib/ConsumerConfigurationImpl.h
+++ b/lib/ConsumerConfigurationImpl.h
@@ -21,48 +21,45 @@
#include <pulsar/ConsumerConfiguration.h>
-#include <chrono>
-
namespace pulsar {
struct ConsumerConfigurationImpl {
- SchemaInfo schemaInfo;
long unAckedMessagesTimeoutMs{0};
long tickDurationInMs{1000};
-
long negativeAckRedeliveryDelayMs{60000};
long ackGroupingTimeMs{100};
long ackGroupingMaxSize{1000};
- ConsumerType consumerType{ConsumerExclusive};
- MessageListener messageListener;
- bool hasMessageListener{false};
+ long brokerConsumerStatsCacheTimeInMs{30 * 1000L}; // 30 seconds
+ long expireTimeOfIncompleteChunkedMessageMs{60000};
+ SchemaInfo schemaInfo;
ConsumerEventListenerPtr eventListener;
+ CryptoKeyReaderPtr cryptoKeyReader;
+ InitialPosition subscriptionInitialPosition{InitialPosition::InitialPositionLatest};
+ int patternAutoDiscoveryPeriod{60};
+ RegexSubscriptionMode regexSubscriptionMode{RegexSubscriptionMode::PersistentOnly};
+ int priorityLevel{0};
+ bool hasMessageListener{false};
bool hasConsumerEventListener{false};
+ bool readCompacted{false};
+ bool replicateSubscriptionStateEnabled{false};
+ bool autoAckOldestChunkedMessageOnQueueFull{false};
+ bool startMessageIdInclusive{false};
+ bool batchIndexAckEnabled{false};
+ bool ackReceiptEnabled{false};
+ bool startPaused{false};
+
+ size_t maxPendingChunkedMessage{10};
+ ConsumerType consumerType{ConsumerExclusive};
+ MessageListener messageListener;
int receiverQueueSize{1000};
int maxTotalReceiverQueueSizeAcrossPartitions{50000};
std::string consumerName;
- long brokerConsumerStatsCacheTimeInMs{30 * 1000L}; // 30 seconds
- CryptoKeyReaderPtr cryptoKeyReader;
ConsumerCryptoFailureAction cryptoFailureAction{ConsumerCryptoFailureAction::FAIL};
- bool readCompacted{false};
- InitialPosition subscriptionInitialPosition{InitialPosition::InitialPositionLatest};
BatchReceivePolicy batchReceivePolicy{};
DeadLetterPolicy deadLetterPolicy;
- int patternAutoDiscoveryPeriod{60};
- RegexSubscriptionMode regexSubscriptionMode{RegexSubscriptionMode::PersistentOnly};
-
- bool replicateSubscriptionStateEnabled{false};
std::map<std::string, std::string> properties;
std::map<std::string, std::string> subscriptionProperties;
- int priorityLevel{0};
KeySharedPolicy keySharedPolicy;
- size_t maxPendingChunkedMessage{10};
- bool autoAckOldestChunkedMessageOnQueueFull{false};
- bool startMessageIdInclusive{false};
- long expireTimeOfIncompleteChunkedMessageMs{60000};
- bool batchIndexAckEnabled{false};
std::vector<ConsumerInterceptorPtr> interceptors;
- bool ackReceiptEnabled{false};
- bool startPaused{false};
};
} // namespace pulsar
#endif /* LIB_CONSUMERCONFIGURATIONIMPL_H_ */
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index f429bfb..64d36c6 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -74,14 +74,14 @@ static boost::optional<MessageId> getStartMessageId(const boost::optional<Messag
return startMessageId;
}
-ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
+ConsumerImpl::ConsumerImpl(const ClientImplPtr& client, const std::string& topic,
const std::string& subscriptionName, const ConsumerConfiguration& conf,
bool isPersistent, const ConsumerInterceptorsPtr& interceptors,
- const ExecutorServicePtr listenerExecutor /* = NULL by default */,
+ const ExecutorServicePtr& listenerExecutor /* = NULL by default */,
bool hasParent /* = false by default */,
const ConsumerTopicType consumerTopicType /* = NonPartitioned by default */,
Commands::SubscriptionMode subscriptionMode,
- boost::optional<MessageId> startMessageId)
+ const boost::optional<MessageId>& startMessageId)
: ConsumerImplBase(
client, topic,
Backoff(milliseconds(client->getClientConfig().getInitialBackoffIntervalMs()),
@@ -129,7 +129,8 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
unsigned int statsIntervalInSeconds = client->getClientConfig().getStatsIntervalInSeconds();
if (statsIntervalInSeconds) {
consumerStatsBasePtr_ = std::make_shared<ConsumerStatsImpl>(
- consumerStr_, client->getIOExecutorProvider()->get(), statsIntervalInSeconds);
+ consumerStr_, client->getIOExecutorProvider()->get()->createDeadlineTimer(),
+ statsIntervalInSeconds);
} else {
consumerStatsBasePtr_ = std::make_shared<ConsumerStatsDisabled>();
}
@@ -141,7 +142,7 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
}
// Config dlq
- auto deadLetterPolicy = conf.getDeadLetterPolicy();
+ const auto& deadLetterPolicy = conf.getDeadLetterPolicy();
if (deadLetterPolicy.getMaxRedeliverCount() > 0) {
auto deadLetterPolicyBuilder =
DeadLetterPolicyBuilder()
@@ -159,13 +160,13 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
}
ConsumerImpl::~ConsumerImpl() {
- LOG_DEBUG(getName() << "~ConsumerImpl");
+ LOG_DEBUG(consumerStr_ << "~ConsumerImpl");
if (state_ == Ready) {
// this could happen at least in this condition:
// consumer seek, caused reconnection, if consumer close happened before connection ready,
// then consumer will not send closeConsumer to Broker side, and caused a leak of consumer in
// broker.
- LOG_WARN(getName() << "Destroyed consumer which was not properly closed");
+ LOG_WARN(consumerStr_ << "Destroyed consumer which was not properly closed");
ClientConnectionPtr cnx = getCnx().lock();
ClientImplPtr client = client_.lock();
@@ -173,12 +174,12 @@ ConsumerImpl::~ConsumerImpl() {
int requestId = client->newRequestId();
cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId);
cnx->removeConsumer(consumerId_);
- LOG_INFO(getName() << "Closed consumer for race condition: " << consumerId_);
+ LOG_INFO(consumerStr_ << "Closed consumer for race condition: " << consumerId_);
} else {
- LOG_WARN(getName() << "Client is destroyed and cannot send the CloseConsumer command");
+ LOG_WARN(consumerStr_ << "Client is destroyed and cannot send the CloseConsumer command");
}
}
- shutdown();
+ internalShutdown();
}
void ConsumerImpl::setPartitionIndex(int partitionIndex) { partitionIndex_ = partitionIndex; }
@@ -367,12 +368,12 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
return handleResult;
}
-void ConsumerImpl::unsubscribeAsync(ResultCallback originalCallback) {
+void ConsumerImpl::unsubscribeAsync(const ResultCallback& originalCallback) {
LOG_INFO(getName() << "Unsubscribing");
auto callback = [this, originalCallback](Result result) {
if (result == ResultOk) {
- shutdown();
+ internalShutdown();
LOG_INFO(getName() << "Unsubscribed successfully");
} else {
state_ = Ready;
@@ -408,7 +409,7 @@ void ConsumerImpl::unsubscribeAsync(ResultCallback originalCallback) {
}
}
-void ConsumerImpl::discardChunkMessages(std::string uuid, MessageId messageId, bool autoAck) {
+void ConsumerImpl::discardChunkMessages(const std::string& uuid, const MessageId& messageId, bool autoAck) {
if (autoAck) {
acknowledgeAsync(messageId, [uuid, messageId](Result result) {
if (result != ResultOk) {
@@ -460,7 +461,7 @@ boost::optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuff
const ClientConnectionPtr& cnx,
MessageId& messageId) {
const auto chunkId = metadata.chunk_id();
- const auto uuid = metadata.uuid();
+ const auto& uuid = metadata.uuid();
LOG_DEBUG("Process message chunk (chunkId: " << chunkId << ", uuid: " << uuid
<< ", messageId: " << messageId << ") of "
<< payload.readableBytes() << " bytes");
@@ -591,11 +592,16 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
<< metadata.has_num_messages_in_batch());
uint32_t numOfMessageReceived = m.impl_->metadata.num_messages_in_batch();
- if (this->ackGroupingTrackerPtr_->isDuplicate(m.getMessageId())) {
+ auto ackGroupingTrackerPtr = ackGroupingTrackerPtr_;
+ if (ackGroupingTrackerPtr == nullptr) { // The consumer is closing
+ return;
+ }
+ if (ackGroupingTrackerPtr->isDuplicate(m.getMessageId())) {
LOG_DEBUG(getName() << " Ignoring message as it was ACKed earlier by same consumer.");
increaseAvailablePermits(cnx, numOfMessageReceived);
return;
}
+ ackGroupingTrackerPtr.reset();
if (metadata.has_num_messages_in_batch()) {
BitSet::Data words(msg.ack_set_size());
@@ -974,7 +980,7 @@ Result ConsumerImpl::receive(Message& msg) {
return res;
}
-void ConsumerImpl::receiveAsync(ReceiveCallback callback) {
+void ConsumerImpl::receiveAsync(const ReceiveCallback& callback) {
Message msg;
// fail the callback if consumer is closing or closed
@@ -1195,7 +1201,7 @@ inline CommandSubscribe_InitialPosition ConsumerImpl::getInitialPosition() {
BOOST_THROW_EXCEPTION(std::logic_error("Invalid InitialPosition enumeration value"));
}
-void ConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callback) {
+void ConsumerImpl::acknowledgeAsync(const MessageId& msgId, const ResultCallback& callback) {
auto pair = prepareIndividualAck(msgId);
const auto& msgIdToAck = pair.first;
const bool readyToAck = pair.second;
@@ -1209,7 +1215,7 @@ void ConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callb
interceptors_->onAcknowledge(Consumer(shared_from_this()), ResultOk, msgId);
}
-void ConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) {
+void ConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdList, const ResultCallback& callback) {
MessageIdList messageIdListToAck;
// TODO: Need to check if the consumer is ready. Same to all other public methods
for (auto&& msgId : messageIdList) {
@@ -1226,7 +1232,7 @@ void ConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdList, ResultCa
this->ackGroupingTrackerPtr_->addAcknowledgeList(messageIdListToAck, callback);
}
-void ConsumerImpl::acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) {
+void ConsumerImpl::acknowledgeCumulativeAsync(const MessageId& msgId, const ResultCallback& callback) {
if (!isCumulativeAcknowledgementAllowed(config_.getConsumerType())) {
interceptors_->onAcknowledgeCumulative(Consumer(shared_from_this()),
ResultCumulativeAcknowledgementNotAllowedError, msgId);
@@ -1304,9 +1310,9 @@ void ConsumerImpl::disconnectConsumer(const boost::optional<std::string>& assign
scheduleReconnection(assignedBrokerUrl);
}
-void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
+void ConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
auto callback = [this, originalCallback](Result result, bool alreadyClosed = false) {
- shutdown();
+ internalShutdown();
if (result == ResultOk) {
if (!alreadyClosed) {
LOG_INFO(getName() << "Closed consumer " << consumerId_);
@@ -1330,9 +1336,12 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
incomingMessages_.close();
// Flush pending grouped ACK requests.
- if (ackGroupingTrackerPtr_) {
- ackGroupingTrackerPtr_->close();
+ if (ackGroupingTrackerPtr_.use_count() != 1) {
+ LOG_ERROR("AckGroupingTracker is shared by other "
+ << (ackGroupingTrackerPtr_.use_count() - 1)
+ << " threads, which will prevent flushing the ACKs");
}
+ ackGroupingTrackerPtr_.reset();
negativeAcksTracker_->close();
ClientConnectionPtr cnx = getCnx().lock();
@@ -1359,10 +1368,10 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
const std::string& ConsumerImpl::getName() const { return consumerStr_; }
-void ConsumerImpl::shutdown() {
- if (ackGroupingTrackerPtr_) {
- ackGroupingTrackerPtr_->close();
- }
+void ConsumerImpl::shutdown() { internalShutdown(); }
+
+void ConsumerImpl::internalShutdown() {
+ ackGroupingTrackerPtr_.reset();
incomingMessages_.clear();
possibleSendToDeadLetterTopicMessages_.clear();
resetCnx();
@@ -1465,7 +1474,7 @@ void ConsumerImpl::redeliverMessages(const std::set<MessageId>& messageIds) {
int ConsumerImpl::getNumOfPrefetchedMessages() const { return incomingMessages_.size(); }
-void ConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) {
+void ConsumerImpl::getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback) {
if (state_ != Ready) {
LOG_ERROR(getName() << "Client connection is not open, please try again later.")
callback(ResultConsumerNotInitialized, BrokerConsumerStats());
@@ -1507,7 +1516,7 @@ void ConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callb
}
void ConsumerImpl::brokerConsumerStatsListener(Result res, BrokerConsumerStatsImpl brokerConsumerStats,
- BrokerConsumerStatsCallback callback) {
+ const BrokerConsumerStatsCallback& callback) {
if (res == ResultOk) {
Lock lock(mutex_);
brokerConsumerStats.setCacheTime(config_.getBrokerConsumerStatsCacheTimeInMs());
@@ -1519,7 +1528,7 @@ void ConsumerImpl::brokerConsumerStatsListener(Result res, BrokerConsumerStatsIm
}
}
-void ConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
+void ConsumerImpl::seekAsync(const MessageId& msgId, const ResultCallback& callback) {
const auto state = state_.load();
if (state == Closed || state == Closing) {
LOG_ERROR(getName() << "Client connection already closed.");
@@ -1538,7 +1547,7 @@ void ConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, msgId), SeekArg{msgId}, callback);
}
-void ConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
+void ConsumerImpl::seekAsync(uint64_t timestamp, const ResultCallback& callback) {
const auto state = state_.load();
if (state == Closed || state == Closing) {
LOG_ERROR(getName() << "Client connection already closed.");
@@ -1560,7 +1569,7 @@ void ConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
bool ConsumerImpl::isReadCompacted() { return readCompacted_; }
-void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback) {
+void ConsumerImpl::hasMessageAvailableAsync(const HasMessageAvailableCallback& callback) {
bool compareMarkDeletePosition;
{
std::lock_guard<std::mutex> lock{mutexForMessageId_};
@@ -1613,7 +1622,7 @@ void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback
}
}
-void ConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) {
+void ConsumerImpl::getLastMessageIdAsync(const BrokerGetLastMessageIdCallback& callback) {
const auto state = state_.load();
if (state == Closed || state == Closing) {
LOG_ERROR(getName() << "Client connection already closed.");
@@ -1632,7 +1641,7 @@ void ConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback
void ConsumerImpl::internalGetLastMessageIdAsync(const BackoffPtr& backoff, TimeDuration remainTime,
const DeadlineTimerPtr& timer,
- BrokerGetLastMessageIdCallback callback) {
+ const BrokerGetLastMessageIdCallback& callback) {
ClientConnectionPtr cnx = getCnx().lock();
if (cnx) {
if (cnx->getServerProtocolVersion() >= proto::v12) {
@@ -1704,8 +1713,8 @@ bool ConsumerImpl::isConnected() const { return !getCnx().expired() && state_ ==
uint64_t ConsumerImpl::getNumberOfConnectedConsumer() { return isConnected() ? 1 : 0; }
-void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const SeekArg& seekArg,
- ResultCallback callback) {
+void ConsumerImpl::seekAsyncInternal(long requestId, const SharedBuffer& seek, const SeekArg& seekArg,
+ const ResultCallback& callback) {
ClientConnectionPtr cnx = getCnx().lock();
if (!cnx) {
LOG_ERROR(getName() << " Client Connection not ready for Consumer");
@@ -1728,7 +1737,7 @@ void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const Se
seekMessageId_ = *boost::get<MessageId>(&seekArg);
}
seekStatus_ = SeekStatus::IN_PROGRESS;
- seekCallback_ = std::move(callback);
+ seekCallback_ = callback;
LOG_INFO(getName() << " Seeking subscription to " << seekArg);
std::weak_ptr<ConsumerImpl> weakSelf{get_shared_this_ptr()};
@@ -1798,7 +1807,7 @@ void ConsumerImpl::cancelTimers() noexcept {
consumerStatsBasePtr_->stop();
}
-void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, ProcessDLQCallBack cb) {
+void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, const ProcessDLQCallBack& cb) {
auto messages = possibleSendToDeadLetterTopicMessages_.find(messageId);
if (!messages) {
cb(false);
@@ -1821,7 +1830,7 @@ void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, ProcessDLQCa
auto self = get_shared_this_ptr();
client->createProducerAsync(
deadLetterPolicy_.getDeadLetterTopic(), producerConfiguration,
- [self](Result res, Producer producer) {
+ [self](Result res, const Producer& producer) {
if (res == ResultOk) {
self->deadLetterProducer_->setValue(producer);
} else {
diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h
index 35636ad..ce0c415 100644
--- a/lib/ConsumerImpl.h
+++ b/lib/ConsumerImpl.h
@@ -23,6 +23,7 @@
#include <boost/optional.hpp>
#include <boost/variant.hpp>
+#include <cstdint>
#include <functional>
#include <list>
#include <memory>
@@ -66,7 +67,7 @@ class BrokerEntryMetadata;
class MessageMetadata;
} // namespace proto
-enum ConsumerTopicType
+enum ConsumerTopicType : uint8_t
{
NonPartitioned,
Partitioned
@@ -85,12 +86,12 @@ enum class SeekStatus : std::uint8_t
class ConsumerImpl : public ConsumerImplBase {
public:
- ConsumerImpl(const ClientImplPtr client, const std::string& topic, const std::string& subscriptionName,
+ ConsumerImpl(const ClientImplPtr& client, const std::string& topic, const std::string& subscriptionName,
const ConsumerConfiguration&, bool isPersistent, const ConsumerInterceptorsPtr& interceptors,
- const ExecutorServicePtr listenerExecutor = ExecutorServicePtr(), bool hasParent = false,
+ const ExecutorServicePtr& listenerExecutor = ExecutorServicePtr(), bool hasParent = false,
const ConsumerTopicType consumerTopicType = NonPartitioned,
Commands::SubscriptionMode = Commands::SubscriptionModeDurable,
- boost::optional<MessageId> startMessageId = boost::none);
+ const boost::optional<MessageId>& startMessageId = boost::none);
~ConsumerImpl();
void setPartitionIndex(int partitionIndex);
int getPartitionIndex();
@@ -113,14 +114,15 @@ class ConsumerImpl : public ConsumerImplBase {
const std::string& getTopic() const override;
Result receive(Message& msg) override;
Result receive(Message& msg, int timeout) override;
- void receiveAsync(ReceiveCallback callback) override;
- void unsubscribeAsync(ResultCallback callback) override;
- void acknowledgeAsync(const MessageId& msgId, ResultCallback callback) override;
- void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) override;
- void acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) override;
- void closeAsync(ResultCallback callback) override;
+ void receiveAsync(const ReceiveCallback& callback) override;
+ void unsubscribeAsync(const ResultCallback& callback) override;
+ void acknowledgeAsync(const MessageId& msgId, const ResultCallback& callback) override;
+ void acknowledgeAsync(const MessageIdList& messageIdList, const ResultCallback& callback) override;
+ void acknowledgeCumulativeAsync(const MessageId& msgId, const ResultCallback& callback) override;
+ void closeAsync(const ResultCallback& callback) override;
void start() override;
void shutdown() override;
+ void internalShutdown();
bool isClosed() override;
bool isOpen() override;
Result pauseMessageListener() override;
@@ -129,14 +131,14 @@ class ConsumerImpl : public ConsumerImplBase {
void redeliverUnacknowledgedMessages(const std::set<MessageId>& messageIds) override;
const std::string& getName() const override;
int getNumOfPrefetchedMessages() const override;
- void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) override;
- void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) override;
- void seekAsync(const MessageId& msgId, ResultCallback callback) override;
- void seekAsync(uint64_t timestamp, ResultCallback callback) override;
+ void getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback) override;
+ void getLastMessageIdAsync(const BrokerGetLastMessageIdCallback& callback) override;
+ void seekAsync(const MessageId& msgId, const ResultCallback& callback) override;
+ void seekAsync(uint64_t timestamp, const ResultCallback& callback) override;
void negativeAcknowledge(const MessageId& msgId) override;
bool isConnected() const override;
uint64_t getNumberOfConnectedConsumer() override;
- void hasMessageAvailableAsync(HasMessageAvailableCallback callback) override;
+ void hasMessageAvailableAsync(const HasMessageAvailableCallback& callback) override;
virtual void disconnectConsumer();
virtual void disconnectConsumer(const boost::optional<std::string>& assignedBrokerUrl);
@@ -184,7 +186,7 @@ class ConsumerImpl : public ConsumerImplBase {
const BitSet& ackSet, int redeliveryCount);
bool isPriorBatchIndex(int32_t idx);
bool isPriorEntryIndex(int64_t idx);
- void brokerConsumerStatsListener(Result, BrokerConsumerStatsImpl, BrokerConsumerStatsCallback);
+ void brokerConsumerStatsListener(Result, BrokerConsumerStatsImpl, const BrokerConsumerStatsCallback&);
bool decryptMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
const proto::MessageMetadata& metadata, SharedBuffer& payload);
@@ -199,7 +201,7 @@ class ConsumerImpl : public ConsumerImplBase {
void trackMessage(const MessageId& messageId);
void internalGetLastMessageIdAsync(const BackoffPtr& backoff, TimeDuration remainTime,
const DeadlineTimerPtr& timer,
- BrokerGetLastMessageIdCallback callback);
+ const BrokerGetLastMessageIdCallback& callback);
void clearReceiveQueue();
using SeekArg = boost::variant<uint64_t, MessageId>;
@@ -213,9 +215,9 @@ class ConsumerImpl : public ConsumerImplBase {
return os;
}
- void seekAsyncInternal(long requestId, SharedBuffer seek, const SeekArg& seekArg,
- ResultCallback callback);
- void processPossibleToDLQ(const MessageId& messageId, ProcessDLQCallBack cb);
+ void seekAsyncInternal(long requestId, const SharedBuffer& seek, const SeekArg& seekArg,
+ const ResultCallback& callback);
+ void processPossibleToDLQ(const MessageId& messageId, const ProcessDLQCallBack& cb);
std::mutex mutexForReceiveWithZeroQueueSize;
const ConsumerConfiguration config_;
@@ -277,9 +279,9 @@ class ConsumerImpl : public ConsumerImplBase {
ChunkedMessageCtx(const ChunkedMessageCtx&) = delete;
// Here we don't use =default to be compatible with GCC 4.8
ChunkedMessageCtx(ChunkedMessageCtx&& rhs) noexcept
- : totalChunks_(rhs.totalChunks_),
- chunkedMsgBuffer_(std::move(rhs.chunkedMsgBuffer_)),
- chunkedMessageIds_(std::move(rhs.chunkedMessageIds_)) {}
+ : totalChunks_(rhs.totalChunks_), chunkedMsgBuffer_(std::move(rhs.chunkedMsgBuffer_)) {
+ std::swap(chunkedMessageIds_, rhs.chunkedMessageIds_);
+ }
bool validateChunkId(int chunkId) const noexcept { return chunkId == numChunks(); }
@@ -295,7 +297,11 @@ class ConsumerImpl : public ConsumerImplBase {
const std::vector<MessageId>& getChunkedMessageIds() const noexcept { return chunkedMessageIds_; }
- std::vector<MessageId> moveChunkedMessageIds() noexcept { return std::move(chunkedMessageIds_); }
+ std::vector<MessageId> moveChunkedMessageIds() noexcept {
+ std::vector<MessageId> result;
+ result.swap(chunkedMessageIds_);
+ return result;
+ }
long getReceivedTimeMs() const noexcept { return receivedTimeMs_; }
@@ -335,7 +341,7 @@ class ConsumerImpl : public ConsumerImplBase {
ConsumerInterceptorsPtr interceptors_;
void triggerCheckExpiredChunkedTimer();
- void discardChunkMessages(std::string uuid, MessageId messageId, bool autoAck);
+ void discardChunkMessages(const std::string& uuid, const MessageId& messageId, bool autoAck);
/**
* Process a chunk. If the chunk is the last chunk of a message, concatenate all buffered chunks into the
diff --git a/lib/ConsumerImplBase.cc b/lib/ConsumerImplBase.cc
index 098f2d5..2963a67 100644
--- a/lib/ConsumerImplBase.cc
+++ b/lib/ConsumerImplBase.cc
@@ -18,9 +18,6 @@
*/
#include "ConsumerImplBase.h"
-#include <algorithm>
-
-#include "ConsumerImpl.h"
#include "ExecutorService.h"
#include "LogUtils.h"
#include "TimeUtils.h"
@@ -29,13 +26,14 @@ DECLARE_LOG_OBJECT()
namespace pulsar {
-ConsumerImplBase::ConsumerImplBase(ClientImplPtr client, const std::string& topic, Backoff backoff,
- const ConsumerConfiguration& conf, ExecutorServicePtr listenerExecutor)
+ConsumerImplBase::ConsumerImplBase(const ClientImplPtr& client, const std::string& topic, Backoff backoff,
+ const ConsumerConfiguration& conf,
+ const ExecutorServicePtr& listenerExecutor)
: HandlerBase(client, topic, backoff),
listenerExecutor_(listenerExecutor),
batchReceivePolicy_(conf.getBatchReceivePolicy()),
consumerName_(conf.getConsumerName()) {
- auto userBatchReceivePolicy = conf.getBatchReceivePolicy();
+ const auto& userBatchReceivePolicy = conf.getBatchReceivePolicy();
if (userBatchReceivePolicy.getMaxNumMessages() > conf.getReceiverQueueSize()) {
batchReceivePolicy_ =
BatchReceivePolicy(conf.getReceiverQueueSize(), userBatchReceivePolicy.getMaxNumBytes(),
@@ -107,7 +105,7 @@ void ConsumerImplBase::notifyBatchPendingReceivedCallback() {
}
}
-void ConsumerImplBase::batchReceiveAsync(BatchReceiveCallback callback) {
+void ConsumerImplBase::batchReceiveAsync(const BatchReceiveCallback& callback) {
// fail the callback if consumer is closing or closed
if (state_ != Ready) {
callback(ResultAlreadyClosed, Messages());
diff --git a/lib/ConsumerImplBase.h b/lib/ConsumerImplBase.h
index 79601e4..ffc0e3c 100644
--- a/lib/ConsumerImplBase.h
+++ b/lib/ConsumerImplBase.h
@@ -36,15 +36,15 @@ class OpBatchReceive {
public:
OpBatchReceive();
explicit OpBatchReceive(const BatchReceiveCallback& batchReceiveCallback);
- const BatchReceiveCallback batchReceiveCallback_;
+ BatchReceiveCallback batchReceiveCallback_;
const int64_t createAt_;
};
class ConsumerImplBase : public HandlerBase {
public:
virtual ~ConsumerImplBase(){};
- ConsumerImplBase(ClientImplPtr client, const std::string& topic, Backoff backoff,
- const ConsumerConfiguration& conf, ExecutorServicePtr listenerExecutor);
+ ConsumerImplBase(const ClientImplPtr& client, const std::string& topic, Backoff backoff,
+ const ConsumerConfiguration& conf, const ExecutorServicePtr& listenerExecutor);
std::shared_ptr<ConsumerImplBase> shared_from_this() noexcept {
return std::dynamic_pointer_cast<ConsumerImplBase>(HandlerBase::shared_from_this());
}
@@ -55,13 +55,13 @@ class ConsumerImplBase : public HandlerBase {
virtual const std::string& getSubscriptionName() const = 0;
virtual Result receive(Message& msg) = 0;
virtual Result receive(Message& msg, int timeout) = 0;
- virtual void receiveAsync(ReceiveCallback callback) = 0;
- void batchReceiveAsync(BatchReceiveCallback callback);
- virtual void unsubscribeAsync(ResultCallback callback) = 0;
- virtual void acknowledgeAsync(const MessageId& msgId, ResultCallback callback) = 0;
- virtual void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) = 0;
- virtual void acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) = 0;
- virtual void closeAsync(ResultCallback callback) = 0;
+ virtual void receiveAsync(const ReceiveCallback& callback) = 0;
+ void batchReceiveAsync(const BatchReceiveCallback& callback);
+ virtual void unsubscribeAsync(const ResultCallback& callback) = 0;
+ virtual void acknowledgeAsync(const MessageId& msgId, const ResultCallback& callback) = 0;
+ virtual void acknowledgeAsync(const MessageIdList& messageIdList, const ResultCallback& callback) = 0;
+ virtual void acknowledgeCumulativeAsync(const MessageId& msgId, const ResultCallback& callback) = 0;
+ virtual void closeAsync(const ResultCallback& callback) = 0;
virtual void start() = 0;
virtual void shutdown() = 0;
virtual bool isClosed() = 0;
@@ -71,16 +71,16 @@ class ConsumerImplBase : public HandlerBase {
virtual void redeliverUnacknowledgedMessages() = 0;
virtual void redeliverUnacknowledgedMessages(const std::set<MessageId>& messageIds) = 0;
virtual int getNumOfPrefetchedMessages() const = 0;
- virtual void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) = 0;
- virtual void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) = 0;
- virtual void seekAsync(const MessageId& msgId, ResultCallback callback) = 0;
- virtual void seekAsync(uint64_t timestamp, ResultCallback callback) = 0;
+ virtual void getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback) = 0;
+ virtual void getLastMessageIdAsync(const BrokerGetLastMessageIdCallback& callback) = 0;
+ virtual void seekAsync(const MessageId& msgId, const ResultCallback& callback) = 0;
+ virtual void seekAsync(uint64_t timestamp, const ResultCallback& callback) = 0;
virtual void negativeAcknowledge(const MessageId& msgId) = 0;
virtual bool isConnected() const = 0;
virtual uint64_t getNumberOfConnectedConsumer() = 0;
// overrided methods from HandlerBase
virtual const std::string& getName() const override = 0;
- virtual void hasMessageAvailableAsync(HasMessageAvailableCallback callback) = 0;
+ virtual void hasMessageAvailableAsync(const HasMessageAvailableCallback& callback) = 0;
const std::string& getConsumerName() const noexcept { return consumerName_; }
diff --git a/lib/CryptoKeyReader.cc b/lib/CryptoKeyReader.cc
index b64e865..8889453 100644
--- a/lib/CryptoKeyReader.cc
+++ b/lib/CryptoKeyReader.cc
@@ -47,7 +47,7 @@ DefaultCryptoKeyReader::DefaultCryptoKeyReader(const std::string& publicKeyPath,
DefaultCryptoKeyReader::~DefaultCryptoKeyReader() {}
-void DefaultCryptoKeyReader::readFile(std::string fileName, std::string& fileContents) const {
+void DefaultCryptoKeyReader::readFile(const std::string& fileName, std::string& fileContents) const {
std::ifstream ifs(fileName);
std::stringstream fileStream;
fileStream << ifs.rdbuf();
diff --git a/lib/EncryptionKeyInfo.cc b/lib/EncryptionKeyInfo.cc
index 68c6c0a..e7edd2b 100644
--- a/lib/EncryptionKeyInfo.cc
+++ b/lib/EncryptionKeyInfo.cc
@@ -25,11 +25,11 @@ namespace pulsar {
EncryptionKeyInfo::EncryptionKeyInfo() : impl_(new EncryptionKeyInfoImpl()) {}
-EncryptionKeyInfo::EncryptionKeyInfo(EncryptionKeyInfoImplPtr impl) : impl_(impl) {}
+EncryptionKeyInfo::EncryptionKeyInfo(const EncryptionKeyInfoImplPtr& impl) : impl_(impl) {}
std::string& EncryptionKeyInfo::getKey() { return impl_->getKey(); }
-void EncryptionKeyInfo::setKey(std::string key) { impl_->setKey(key); }
+void EncryptionKeyInfo::setKey(const std::string& key) { impl_->setKey(key); }
EncryptionKeyInfo::StringMap& EncryptionKeyInfo::getMetadata() { return impl_->getMetadata(); }
diff --git a/lib/EncryptionKeyInfoImpl.cc b/lib/EncryptionKeyInfoImpl.cc
index e61ca70..6632832 100644
--- a/lib/EncryptionKeyInfoImpl.cc
+++ b/lib/EncryptionKeyInfoImpl.cc
@@ -22,11 +22,11 @@
namespace pulsar {
EncryptionKeyInfoImpl::EncryptionKeyInfoImpl(std::string key, StringMap& metadata)
- : metadata_(metadata), key_(key) {}
+ : metadata_(metadata), key_(std::move(key)) {}
std::string& EncryptionKeyInfoImpl::getKey() { return key_; }
-void EncryptionKeyInfoImpl::setKey(std::string key) { key_ = key; }
+void EncryptionKeyInfoImpl::setKey(const std::string& key) { key_ = key; }
EncryptionKeyInfoImpl::StringMap& EncryptionKeyInfoImpl::getMetadata() { return metadata_; }
diff --git a/lib/EncryptionKeyInfoImpl.h b/lib/EncryptionKeyInfoImpl.h
index 5ff4ceb..5a7a06c 100644
--- a/lib/EncryptionKeyInfoImpl.h
+++ b/lib/EncryptionKeyInfoImpl.h
@@ -36,7 +36,7 @@ class PULSAR_PUBLIC EncryptionKeyInfoImpl {
std::string& getKey();
- void setKey(std::string key);
+ void setKey(const std::string& key);
StringMap& getMetadata(void);
diff --git a/lib/Future.h b/lib/Future.h
index 22d43cb..3b2b524 100644
--- a/lib/Future.h
+++ b/lib/Future.h
@@ -72,7 +72,8 @@ class InternalState {
cond_.notify_all();
if (!listeners_.empty()) {
- auto listeners = std::move(listeners_);
+ decltype(listeners_) listeners;
+ listeners.swap(listeners_);
lock.unlock();
for (auto &&listener : listeners) {
listener(result, value);
diff --git a/lib/HTTPLookupService.cc b/lib/HTTPLookupService.cc
index 60b5343..79d9e94 100644
--- a/lib/HTTPLookupService.cc
+++ b/lib/HTTPLookupService.cc
@@ -173,8 +173,8 @@ Future<Result, SchemaInfo> HTTPLookupService::getSchema(const TopicNamePtr &topi
return promise.getFuture();
}
-void HTTPLookupService::handleNamespaceTopicsHTTPRequest(NamespaceTopicsPromise promise,
- const std::string completeUrl) {
+void HTTPLookupService::handleNamespaceTopicsHTTPRequest(const NamespaceTopicsPromise &promise,
+ const std::string &completeUrl) {
std::string responseData;
Result result = sendHTTPRequest(completeUrl, responseData);
@@ -185,12 +185,12 @@ void HTTPLookupService::handleNamespaceTopicsHTTPRequest(NamespaceTopicsPromise
}
}
-Result HTTPLookupService::sendHTTPRequest(std::string completeUrl, std::string &responseData) {
+Result HTTPLookupService::sendHTTPRequest(const std::string &completeUrl, std::string &responseData) {
long responseCode = -1;
return sendHTTPRequest(completeUrl, responseData, responseCode);
}
-Result HTTPLookupService::sendHTTPRequest(std::string completeUrl, std::string &responseData,
+Result HTTPLookupService::sendHTTPRequest(const std::string &completeUrl, std::string &responseData,
long &responseCode) {
// Authorization data
AuthenticationDataPtr authDataContent;
@@ -351,7 +351,7 @@ NamespaceTopicsPtr HTTPLookupService::parseNamespaceTopicsData(const std::string
return topicsResultPtr;
}
-void HTTPLookupService::handleLookupHTTPRequest(LookupPromise promise, const std::string completeUrl,
+void HTTPLookupService::handleLookupHTTPRequest(const LookupPromise &promise, const std::string &completeUrl,
RequestType requestType) {
std::string responseData;
Result result = sendHTTPRequest(completeUrl, responseData);
@@ -364,7 +364,8 @@ void HTTPLookupService::handleLookupHTTPRequest(LookupPromise promise, const std
}
}
-void HTTPLookupService::handleGetSchemaHTTPRequest(GetSchemaPromise promise, const std::string completeUrl) {
+void HTTPLookupService::handleGetSchemaHTTPRequest(const GetSchemaPromise &promise,
+ const std::string &completeUrl) {
std::string responseData;
long responseCode = -1;
Result result = sendHTTPRequest(completeUrl, responseData, responseCode);
diff --git a/lib/HTTPLookupService.h b/lib/HTTPLookupService.h
index 17dd110..d17edd5 100644
--- a/lib/HTTPLookupService.h
+++ b/lib/HTTPLookupService.h
@@ -19,6 +19,8 @@
#ifndef PULSAR_CPP_HTTPLOOKUPSERVICE_H
#define PULSAR_CPP_HTTPLOOKUPSERVICE_H
+#include <cstdint>
+
#include "ClientImpl.h"
#include "LookupService.h"
#include "Url.h"
@@ -31,7 +33,7 @@ using NamespaceTopicsPromisePtr = std::shared_ptr<NamespaceTopicsPromise>;
using GetSchemaPromise = Promise<Result, SchemaInfo>;
class HTTPLookupService : public LookupService, public std::enable_shared_from_this<HTTPLookupService> {
- enum RequestType
+ enum RequestType : uint8_t
{
Lookup,
PartitionMetaData
@@ -55,13 +57,14 @@ class HTTPLookupService : public LookupService, public std::enable_shared_from_t
static LookupDataResultPtr parseLookupData(const std::string&);
static NamespaceTopicsPtr parseNamespaceTopicsData(const std::string&);
- void handleLookupHTTPRequest(LookupPromise, const std::string, RequestType);
- void handleNamespaceTopicsHTTPRequest(NamespaceTopicsPromise promise, const std::string completeUrl);
- void handleGetSchemaHTTPRequest(GetSchemaPromise promise, const std::string completeUrl);
+ void handleLookupHTTPRequest(const LookupPromise&, const std::string&, RequestType);
+ void handleNamespaceTopicsHTTPRequest(const NamespaceTopicsPromise& promise,
+ const std::string& completeUrl);
+ void handleGetSchemaHTTPRequest(const GetSchemaPromise& promise, const std::string& completeUrl);
- Result sendHTTPRequest(std::string completeUrl, std::string& responseData);
+ Result sendHTTPRequest(const std::string& completeUrl, std::string& responseData);
- Result sendHTTPRequest(std::string completeUrl, std::string& responseData, long& responseCode);
+ Result sendHTTPRequest(const std::string& completeUrl, std::string& responseData, long& responseCode);
public:
HTTPLookupService(const std::string&, const ClientConfiguration&, const AuthenticationPtr&);
diff --git a/lib/KeyValue.cc b/lib/KeyValue.cc
index e031f52..ff0f5d8 100644
--- a/lib/KeyValue.cc
+++ b/lib/KeyValue.cc
@@ -22,7 +22,7 @@
namespace pulsar {
-KeyValue::KeyValue(KeyValueImplPtr impl) : impl_(impl) {}
+KeyValue::KeyValue(KeyValueImplPtr impl) : impl_(std::move(impl)) {}
KeyValue::KeyValue(std::string &&key, std::string &&value)
: impl_(std::make_shared<KeyValueImpl>(std::move(key), std::move(value))) {}
diff --git a/lib/LogUtils.cc b/lib/LogUtils.cc
index 8739bb8..9b2afc9 100644
--- a/lib/LogUtils.cc
+++ b/lib/LogUtils.cc
@@ -21,7 +21,6 @@
#include <pulsar/ConsoleLoggerFactory.h>
#include <atomic>
-#include <iostream>
namespace pulsar {
@@ -46,8 +45,8 @@ LoggerFactory* LogUtils::getLoggerFactory() {
std::string LogUtils::getLoggerName(const std::string& path) {
// Remove all directories from filename
- int startIdx = path.find_last_of("/");
- int endIdx = path.find_last_of(".");
+ int startIdx = path.find_last_of('/');
+ int endIdx = path.find_last_of('.');
return path.substr(startIdx + 1, endIdx - startIdx - 1);
}
diff --git a/lib/MessageCrypto.cc b/lib/MessageCrypto.cc
index 920e3fd..dc636c0 100644
--- a/lib/MessageCrypto.cc
+++ b/lib/MessageCrypto.cc
@@ -145,7 +145,7 @@ std::string MessageCrypto::stringToHex(const std::string& inputStr, size_t len)
}
Result MessageCrypto::addPublicKeyCipher(const std::set<std::string>& keyNames,
- const CryptoKeyReaderPtr keyReader) {
+ const CryptoKeyReaderPtr& keyReader) {
Lock lock(mutex_);
// Generate data key
@@ -166,7 +166,7 @@ Result MessageCrypto::addPublicKeyCipher(const std::set<std::string>& keyNames,
return result;
}
-Result MessageCrypto::addPublicKeyCipher(const std::string& keyName, const CryptoKeyReaderPtr keyReader) {
+Result MessageCrypto::addPublicKeyCipher(const std::string& keyName, const CryptoKeyReaderPtr& keyReader) {
if (keyName.empty()) {
LOG_ERROR(logCtx_ << "Keyname is empty ");
return ResultCryptoError;
@@ -222,7 +222,7 @@ bool MessageCrypto::removeKeyCipher(const std::string& keyName) {
return true;
}
-bool MessageCrypto::encrypt(const std::set<std::string>& encKeys, const CryptoKeyReaderPtr keyReader,
+bool MessageCrypto::encrypt(const std::set<std::string>& encKeys, const CryptoKeyReaderPtr& keyReader,
proto::MessageMetadata& msgMetadata, SharedBuffer& payload,
SharedBuffer& encryptedPayload) {
if (!encKeys.size()) {
@@ -493,7 +493,7 @@ bool MessageCrypto::getKeyAndDecryptData(const proto::MessageMetadata& msgMetada
}
bool MessageCrypto::decrypt(const proto::MessageMetadata& msgMetadata, SharedBuffer& payload,
- const CryptoKeyReaderPtr keyReader, SharedBuffer& decryptedPayload) {
+ const CryptoKeyReaderPtr& keyReader, SharedBuffer& decryptedPayload) {
// Attempt to decrypt using the existing key
if (getKeyAndDecryptData(msgMetadata, payload, decryptedPayload)) {
return true;
diff --git a/lib/MessageCrypto.h b/lib/MessageCrypto.h
index 90b891e..3e955d9 100644
--- a/lib/MessageCrypto.h
+++ b/lib/MessageCrypto.h
@@ -62,7 +62,7 @@ class MessageCrypto {
* @return ResultOk if succeeded
*
*/
- Result addPublicKeyCipher(const std::set<std::string>& keyNames, const CryptoKeyReaderPtr keyReader);
+ Result addPublicKeyCipher(const std::set<std::string>& keyNames, const CryptoKeyReaderPtr& keyReader);
/*
* Remove a key <p> Remove the key identified by the keyName from the list of keys.<p>
@@ -84,7 +84,7 @@ class MessageCrypto {
*
* @return true if success
*/
- bool encrypt(const std::set<std::string>& encKeys, const CryptoKeyReaderPtr keyReader,
+ bool encrypt(const std::set<std::string>& encKeys, const CryptoKeyReaderPtr& keyReader,
proto::MessageMetadata& msgMetadata, SharedBuffer& payload, SharedBuffer& encryptedPayload);
/*
@@ -98,7 +98,7 @@ class MessageCrypto {
* @return true if success
*/
bool decrypt(const proto::MessageMetadata& msgMetadata, SharedBuffer& payload,
- const CryptoKeyReaderPtr keyReader, SharedBuffer& decryptedPayload);
+ const CryptoKeyReaderPtr& keyReader, SharedBuffer& decryptedPayload);
private:
typedef std::unique_lock<std::mutex> Lock;
@@ -131,7 +131,7 @@ class MessageCrypto {
unsigned char keyDigest[], unsigned int& digestLen);
void removeExpiredDataKey();
- Result addPublicKeyCipher(const std::string& keyName, const CryptoKeyReaderPtr keyReader);
+ Result addPublicKeyCipher(const std::string& keyName, const CryptoKeyReaderPtr& keyReader);
bool decryptDataKey(const proto::EncryptionKeys& encKeys, const CryptoKeyReader& keyReader);
bool decryptData(const std::string& dataKeySecret, const proto::MessageMetadata& msgMetadata,
diff --git a/lib/MessageImpl.cc b/lib/MessageImpl.cc
index 4650b99..3974c4c 100644
--- a/lib/MessageImpl.cc
+++ b/lib/MessageImpl.cc
@@ -122,7 +122,7 @@ void MessageImpl::convertPayloadToKeyValue(const pulsar::SchemaInfo& schemaInfo)
getKeyValueEncodingType(schemaInfo));
}
-KeyValueEncodingType MessageImpl::getKeyValueEncodingType(SchemaInfo schemaInfo) {
+KeyValueEncodingType MessageImpl::getKeyValueEncodingType(const SchemaInfo& schemaInfo) {
if (schemaInfo.getSchemaType() != KEY_VALUE) {
throw std::invalid_argument("Schema not key value type.");
}
diff --git a/lib/MessageImpl.h b/lib/MessageImpl.h
index 55b9612..6467b35 100644
--- a/lib/MessageImpl.h
+++ b/lib/MessageImpl.h
@@ -76,7 +76,7 @@ class MessageImpl {
void setSchemaVersion(const std::string& value);
void convertKeyValueToPayload(const SchemaInfo& schemaInfo);
void convertPayloadToKeyValue(const SchemaInfo& schemaInfo);
- KeyValueEncodingType getKeyValueEncodingType(SchemaInfo schemaInfo);
+ KeyValueEncodingType getKeyValueEncodingType(const SchemaInfo& schemaInfo);
friend class PulsarWrapper;
friend class MessageBuilder;
diff --git a/lib/MultiTopicsBrokerConsumerStatsImpl.cc b/lib/MultiTopicsBrokerConsumerStatsImpl.cc
index 4f96922..159043b 100644
--- a/lib/MultiTopicsBrokerConsumerStatsImpl.cc
+++ b/lib/MultiTopicsBrokerConsumerStatsImpl.cc
@@ -152,7 +152,7 @@ BrokerConsumerStats MultiTopicsBrokerConsumerStatsImpl::getBrokerConsumerStats(i
return statsList_[index];
}
-void MultiTopicsBrokerConsumerStatsImpl::add(BrokerConsumerStats stats, int index) {
+void MultiTopicsBrokerConsumerStatsImpl::add(const BrokerConsumerStats& stats, int index) {
statsList_[index] = stats;
}
diff --git a/lib/MultiTopicsBrokerConsumerStatsImpl.h b/lib/MultiTopicsBrokerConsumerStatsImpl.h
index 481318e..942d3a0 100644
--- a/lib/MultiTopicsBrokerConsumerStatsImpl.h
+++ b/lib/MultiTopicsBrokerConsumerStatsImpl.h
@@ -76,7 +76,7 @@ class PULSAR_PUBLIC MultiTopicsBrokerConsumerStatsImpl : public BrokerConsumerSt
/** Returns the BrokerConsumerStatsImpl at of ith partition */
BrokerConsumerStats getBrokerConsumerStats(int index);
- void add(BrokerConsumerStats stats, int index);
+ void add(const BrokerConsumerStats &stats, int index);
void clear();
diff --git a/lib/MultiTopicsConsumerImpl.cc b/lib/MultiTopicsConsumerImpl.cc
index 8a43173..050b0b1 100644
--- a/lib/MultiTopicsConsumerImpl.cc
+++ b/lib/MultiTopicsConsumerImpl.cc
@@ -40,25 +40,23 @@ using namespace pulsar;
using std::chrono::milliseconds;
using std::chrono::seconds;
-MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(ClientImplPtr client, TopicNamePtr topicName,
+MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(const ClientImplPtr& client, const TopicNamePtr& topicName,
int numPartitions, const std::string& subscriptionName,
const ConsumerConfiguration& conf,
- LookupServicePtr lookupServicePtr,
+ const LookupServicePtr& lookupServicePtr,
const ConsumerInterceptorsPtr& interceptors,
- const Commands::SubscriptionMode subscriptionMode,
- boost::optional<MessageId> startMessageId)
+ Commands::SubscriptionMode subscriptionMode,
+ const boost::optional<MessageId>& startMessageId)
: MultiTopicsConsumerImpl(client, {topicName->toString()}, subscriptionName, topicName, conf,
lookupServicePtr, interceptors, subscriptionMode, startMessageId) {
topicsPartitions_[topicName->toString()] = numPartitions;
}
-MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(ClientImplPtr client, const std::vector<std::string>& topics,
- const std::string& subscriptionName, TopicNamePtr topicName,
- const ConsumerConfiguration& conf,
- LookupServicePtr lookupServicePtr,
- const ConsumerInterceptorsPtr& interceptors,
- const Commands::SubscriptionMode subscriptionMode,
- boost::optional<MessageId> startMessageId)
+MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(
+ const ClientImplPtr& client, const std::vector<std::string>& topics, const std::string& subscriptionName,
+ const TopicNamePtr& topicName, const ConsumerConfiguration& conf,
+ const LookupServicePtr& lookupServicePtr, const ConsumerInterceptorsPtr& interceptors,
+ Commands::SubscriptionMode subscriptionMode, const boost::optional<MessageId>& startMessageId)
: ConsumerImplBase(client, topicName ? topicName->toString() : "EmptyTopics",
Backoff(milliseconds(100), seconds(60), milliseconds(0)), conf,
client->getListenerExecutorProvider()->get()),
@@ -120,7 +118,7 @@ void MultiTopicsConsumerImpl::start() {
// subscribe for each passed in topic
auto weakSelf = weak_from_this();
for (std::vector<std::string>::const_iterator itr = topics_.begin(); itr != topics_.end(); itr++) {
- auto topic = *itr;
+ const auto& topic = *itr;
subscribeOneTopicAsync(topic).addListener(
[this, weakSelf, topic, topicsNeedCreate](Result result, const Consumer& consumer) {
auto self = weakSelf.lock();
@@ -131,9 +129,9 @@ void MultiTopicsConsumerImpl::start() {
}
}
-void MultiTopicsConsumerImpl::handleOneTopicSubscribed(Result result, Consumer consumer,
- const std::string& topic,
- std::shared_ptr<std::atomic<int>> topicsNeedCreate) {
+void MultiTopicsConsumerImpl::handleOneTopicSubscribed(
+ Result result, const Consumer& consumer, const std::string& topic,
+ const std::shared_ptr<std::atomic<int>>& topicsNeedCreate) {
if (result != ResultOk) {
state_ = Failed;
// Use the first failed result
@@ -205,9 +203,9 @@ Future<Result, Consumer> MultiTopicsConsumerImpl::subscribeOneTopicAsync(const s
return topicPromise->getFuture();
}
-void MultiTopicsConsumerImpl::subscribeTopicPartitions(int numPartitions, TopicNamePtr topicName,
- const std::string& consumerName,
- ConsumerSubResultPromisePtr topicSubResultPromise) {
+void MultiTopicsConsumerImpl::subscribeTopicPartitions(
+ int numPartitions, const TopicNamePtr& topicName, const std::string& consumerName,
+ const ConsumerSubResultPromisePtr& topicSubResultPromise) {
std::shared_ptr<ConsumerImpl> consumer;
ConsumerConfiguration config = conf_.clone();
// Pause messageListener until all child topics are subscribed.
@@ -223,7 +221,7 @@ void MultiTopicsConsumerImpl::subscribeTopicPartitions(int numPartitions, TopicN
ExecutorServicePtr internalListenerExecutor = client->getPartitionListenerExecutorProvider()->get();
auto weakSelf = weak_from_this();
- config.setMessageListener([this, weakSelf](Consumer consumer, const Message& msg) {
+ config.setMessageListener([this, weakSelf](const Consumer& consumer, const Message& msg) {
auto self = weakSelf.lock();
if (self) {
messageReceived(consumer, msg);
@@ -295,9 +293,9 @@ void MultiTopicsConsumerImpl::subscribeTopicPartitions(int numPartitions, TopicN
}
void MultiTopicsConsumerImpl::handleSingleConsumerCreated(
- Result result, ConsumerImplBaseWeakPtr consumerImplBaseWeakPtr,
- std::shared_ptr<std::atomic<int>> partitionsNeedCreate,
- ConsumerSubResultPromisePtr topicSubResultPromise) {
+ Result result, const ConsumerImplBaseWeakPtr& consumerImplBaseWeakPtr,
+ const std::shared_ptr<std::atomic<int>>& partitionsNeedCreate,
+ const ConsumerSubResultPromisePtr& topicSubResultPromise) {
if (state_ == Failed) {
// one of the consumer creation failed, and we are cleaning up
topicSubResultPromise->setFailed(ResultAlreadyClosed);
@@ -325,12 +323,12 @@ void MultiTopicsConsumerImpl::handleSingleConsumerCreated(
}
}
-void MultiTopicsConsumerImpl::unsubscribeAsync(ResultCallback originalCallback) {
+void MultiTopicsConsumerImpl::unsubscribeAsync(const ResultCallback& originalCallback) {
LOG_INFO("[ Topics Consumer " << topic() << "," << subscriptionName_ << "] Unsubscribing");
auto callback = [this, originalCallback](Result result) {
if (result == ResultOk) {
- shutdown();
+ internalShutdown();
LOG_INFO(getName() << "Unsubscribed successfully");
} else {
state_ = Ready;
@@ -350,7 +348,7 @@ void MultiTopicsConsumerImpl::unsubscribeAsync(ResultCallback originalCallback)
auto self = get_shared_this_ptr();
consumers_.forEachValue(
- [this, self, callback](const ConsumerImplPtr& consumer, SharedFuture future) {
+ [this, self, callback](const ConsumerImplPtr& consumer, const SharedFuture& future) {
consumer->unsubscribeAsync([this, self, callback, future](Result result) {
if (result != ResultOk) {
state_ = Failed;
@@ -367,7 +365,8 @@ void MultiTopicsConsumerImpl::unsubscribeAsync(ResultCallback originalCallback)
[callback] { callback(ResultOk); });
}
-void MultiTopicsConsumerImpl::unsubscribeOneTopicAsync(const std::string& topic, ResultCallback callback) {
+void MultiTopicsConsumerImpl::unsubscribeOneTopicAsync(const std::string& topic,
+ const ResultCallback& callback) {
Lock lock(mutex_);
std::map<std::string, int>::iterator it = topicsPartitions_.find(topic);
if (it == topicsPartitions_.end()) {
@@ -412,8 +411,8 @@ void MultiTopicsConsumerImpl::unsubscribeOneTopicAsync(const std::string& topic,
}
void MultiTopicsConsumerImpl::handleOneTopicUnsubscribedAsync(
- Result result, std::shared_ptr<std::atomic<int>> consumerUnsubed, int numberPartitions,
- TopicNamePtr topicNamePtr, std::string& topicPartitionName, ResultCallback callback) {
+ Result result, const std::shared_ptr<std::atomic<int>>& consumerUnsubed, int numberPartitions,
+ const TopicNamePtr& topicNamePtr, const std::string& topicPartitionName, const ResultCallback& callback) {
(*consumerUnsubed)++;
if (result != ResultOk) {
@@ -448,12 +447,12 @@ void MultiTopicsConsumerImpl::handleOneTopicUnsubscribedAsync(
}
}
-void MultiTopicsConsumerImpl::closeAsync(ResultCallback originalCallback) {
+void MultiTopicsConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
std::weak_ptr<MultiTopicsConsumerImpl> weakSelf{get_shared_this_ptr()};
auto callback = [weakSelf, originalCallback](Result result) {
auto self = weakSelf.lock();
if (self) {
- self->shutdown();
+ self->internalShutdown();
if (result != ResultOk) {
LOG_WARN(self->getName() << "Failed to close consumer: " << result);
if (result != ResultAlreadyClosed) {
@@ -511,7 +510,7 @@ void MultiTopicsConsumerImpl::closeAsync(ResultCallback originalCallback) {
batchReceiveTimer_->cancel();
}
-void MultiTopicsConsumerImpl::messageReceived(Consumer consumer, const Message& msg) {
+void MultiTopicsConsumerImpl::messageReceived(const Consumer& consumer, const Message& msg) {
if (PULSAR_UNLIKELY(duringSeek_.load(std::memory_order_acquire))) {
return;
}
@@ -555,7 +554,7 @@ void MultiTopicsConsumerImpl::messageReceived(Consumer consumer, const Message&
}
}
-void MultiTopicsConsumerImpl::internalListener(Consumer consumer) {
+void MultiTopicsConsumerImpl::internalListener(const Consumer& consumer) {
Message m;
incomingMessages_.pop(m);
try {
@@ -603,7 +602,7 @@ Result MultiTopicsConsumerImpl::receive(Message& msg, int timeout) {
}
}
-void MultiTopicsConsumerImpl::receiveAsync(ReceiveCallback callback) {
+void MultiTopicsConsumerImpl::receiveAsync(const ReceiveCallback& callback) {
Message msg;
// fail the callback if consumer is closing or closed
@@ -650,7 +649,7 @@ void MultiTopicsConsumerImpl::notifyPendingReceivedCallback(Result result, const
callback(result, msg);
}
-void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callback) {
+void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageId& msgId, const ResultCallback& callback) {
if (state_ != Ready) {
interceptors_->onAcknowledge(Consumer(shared_from_this()), ResultAlreadyClosed, msgId);
callback(ResultAlreadyClosed);
@@ -674,7 +673,8 @@ void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCal
}
}
-void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) {
+void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdList,
+ const ResultCallback& callback) {
if (state_ != Ready) {
callback(ResultAlreadyClosed);
return;
@@ -682,7 +682,7 @@ void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdLis
std::unordered_map<std::string, MessageIdList> topicToMessageId;
for (const MessageId& messageId : messageIdList) {
- auto topicName = messageId.getTopicName();
+ const auto& topicName = messageId.getTopicName();
if (topicName.empty()) {
LOG_ERROR("MessageId without a topic name cannot be acknowledged for a multi-topics consumer");
callback(ResultOperationNotSupported);
@@ -716,7 +716,8 @@ void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdLis
}
}
-void MultiTopicsConsumerImpl::acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) {
+void MultiTopicsConsumerImpl::acknowledgeCumulativeAsync(const MessageId& msgId,
+ const ResultCallback& callback) {
msgId.getTopicName();
auto optConsumer = consumers_.find(msgId.getTopicName());
if (optConsumer) {
@@ -734,7 +735,7 @@ void MultiTopicsConsumerImpl::negativeAcknowledge(const MessageId& msgId) {
}
}
-MultiTopicsConsumerImpl::~MultiTopicsConsumerImpl() { shutdown(); }
+MultiTopicsConsumerImpl::~MultiTopicsConsumerImpl() { internalShutdown(); }
Future<Result, ConsumerImplBaseWeakPtr> MultiTopicsConsumerImpl::getConsumerCreatedFuture() {
return multiTopicsConsumerCreatedPromise_.getFuture();
@@ -745,7 +746,9 @@ const std::string& MultiTopicsConsumerImpl::getTopic() const { return topic(); }
const std::string& MultiTopicsConsumerImpl::getName() const { return consumerStr_; }
-void MultiTopicsConsumerImpl::shutdown() {
+void MultiTopicsConsumerImpl::shutdown() { internalShutdown(); }
+
+void MultiTopicsConsumerImpl::internalShutdown() {
cancelTimers();
incomingMessages_.clear();
topicsPartitions_.clear();
@@ -812,7 +815,7 @@ void MultiTopicsConsumerImpl::redeliverUnacknowledgedMessages(const std::set<Mes
LOG_DEBUG("Sending RedeliverUnacknowledgedMessages command for partitioned consumer.");
std::unordered_map<std::string, std::set<MessageId>> topicToMessageId;
for (const MessageId& messageId : messageIds) {
- auto topicName = messageId.getTopicName();
+ const auto& topicName = messageId.getTopicName();
topicToMessageId[topicName].emplace(messageId);
}
@@ -828,7 +831,7 @@ void MultiTopicsConsumerImpl::redeliverUnacknowledgedMessages(const std::set<Mes
int MultiTopicsConsumerImpl::getNumOfPrefetchedMessages() const { return incomingMessages_.size(); }
-void MultiTopicsConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) {
+void MultiTopicsConsumerImpl::getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback) {
if (state_ != Ready) {
callback(ResultConsumerNotInitialized, BrokerConsumerStats());
return;
@@ -843,24 +846,26 @@ void MultiTopicsConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCal
consumers_.forEachValue([this, &latchPtr, &statsPtr, &i, callback](const ConsumerImplPtr& consumer) {
size_t index = i++;
auto weakSelf = weak_from_this();
- consumer->getBrokerConsumerStatsAsync(
- [this, weakSelf, latchPtr, statsPtr, index, callback](Result result, BrokerConsumerStats stats) {
- auto self = weakSelf.lock();
- if (self) {
- handleGetConsumerStats(result, stats, latchPtr, statsPtr, index, callback);
- }
- });
+ consumer->getBrokerConsumerStatsAsync([this, weakSelf, latchPtr, statsPtr, index, callback](
+ Result result, const BrokerConsumerStats& stats) {
+ auto self = weakSelf.lock();
+ if (self) {
+ handleGetConsumerStats(result, stats, latchPtr, statsPtr, index, callback);
+ }
+ });
});
}
-void MultiTopicsConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) {
+void MultiTopicsConsumerImpl::getLastMessageIdAsync(const BrokerGetLastMessageIdCallback& callback) {
callback(ResultOperationNotSupported, GetLastMessageIdResponse());
}
-void MultiTopicsConsumerImpl::handleGetConsumerStats(Result res, BrokerConsumerStats brokerConsumerStats,
- LatchPtr latchPtr,
- MultiTopicsBrokerConsumerStatsPtr statsPtr, size_t index,
- BrokerConsumerStatsCallback callback) {
+void MultiTopicsConsumerImpl::handleGetConsumerStats(Result res,
+ const BrokerConsumerStats& brokerConsumerStats,
+ const LatchPtr& latchPtr,
+ const MultiTopicsBrokerConsumerStatsPtr& statsPtr,
+ size_t index,
+ const BrokerConsumerStatsCallback& callback) {
Lock lock(mutex_);
if (res == ResultOk) {
latchPtr->countdown();
@@ -907,7 +912,7 @@ void MultiTopicsConsumerImpl::afterSeek() {
});
}
-void MultiTopicsConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
+void MultiTopicsConsumerImpl::seekAsync(const MessageId& msgId, const ResultCallback& callback) {
if (msgId == MessageId::earliest() || msgId == MessageId::latest()) {
return seekAllAsync(msgId, callback);
}
@@ -933,7 +938,7 @@ void MultiTopicsConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback c
});
}
-void MultiTopicsConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
+void MultiTopicsConsumerImpl::seekAsync(uint64_t timestamp, const ResultCallback& callback) {
seekAllAsync(timestamp, callback);
}
@@ -992,7 +997,7 @@ void MultiTopicsConsumerImpl::topicPartitionUpdate() {
});
}
}
-void MultiTopicsConsumerImpl::handleGetPartitions(TopicNamePtr topicName, Result result,
+void MultiTopicsConsumerImpl::handleGetPartitions(const TopicNamePtr& topicName, Result result,
const LookupDataResultPtr& lookupDataResult,
int currentNumPartitions) {
if (state_ != Ready) {
@@ -1024,9 +1029,9 @@ void MultiTopicsConsumerImpl::handleGetPartitions(TopicNamePtr topicName, Result
}
void MultiTopicsConsumerImpl::subscribeSingleNewConsumer(
- int numPartitions, TopicNamePtr topicName, int partitionIndex,
- ConsumerSubResultPromisePtr topicSubResultPromise,
- std::shared_ptr<std::atomic<int>> partitionsNeedCreate) {
+ int numPartitions, const TopicNamePtr& topicName, int partitionIndex,
+ const ConsumerSubResultPromisePtr& topicSubResultPromise,
+ const std::shared_ptr<std::atomic<int>>& partitionsNeedCreate) {
ConsumerConfiguration config = conf_.clone();
auto client = client_.lock();
if (!client) {
@@ -1035,7 +1040,7 @@ void MultiTopicsConsumerImpl::subscribeSingleNewConsumer(
}
ExecutorServicePtr internalListenerExecutor = client->getPartitionListenerExecutorProvider()->get();
auto weakSelf = weak_from_this();
- config.setMessageListener([this, weakSelf](Consumer consumer, const Message& msg) {
+ config.setMessageListener([this, weakSelf](const Consumer& consumer, const Message& msg) {
auto self = weakSelf.lock();
if (self) {
messageReceived(consumer, msg);
@@ -1120,7 +1125,7 @@ void MultiTopicsConsumerImpl::cancelTimers() noexcept {
}
}
-void MultiTopicsConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback) {
+void MultiTopicsConsumerImpl::hasMessageAvailableAsync(const HasMessageAvailableCallback& callback) {
if (incomingMessagesSize_ > 0) {
callback(ResultOk, true);
return;
@@ -1130,24 +1135,25 @@ void MultiTopicsConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallba
auto needCallBack = std::make_shared<std::atomic<int>>(consumers_.size());
auto self = get_shared_this_ptr();
- consumers_.forEachValue([self, needCallBack, callback, hasMessageAvailable](ConsumerImplPtr consumer) {
- consumer->hasMessageAvailableAsync(
- [self, needCallBack, callback, hasMessageAvailable](Result result, bool hasMsg) {
- if (result != ResultOk) {
- LOG_ERROR("Filed when acknowledge list: " << result);
- // set needCallBack is -1 to avoid repeated callback.
- needCallBack->store(-1);
- callback(result, false);
- return;
- }
-
- if (hasMsg) {
- hasMessageAvailable->store(hasMsg);
- }
-
- if (--(*needCallBack) == 0) {
- callback(result, hasMessageAvailable->load() || self->incomingMessagesSize_ > 0);
- }
- });
- });
+ consumers_.forEachValue(
+ [self, needCallBack, callback, hasMessageAvailable](const ConsumerImplPtr& consumer) {
+ consumer->hasMessageAvailableAsync(
+ [self, needCallBack, callback, hasMessageAvailable](Result result, bool hasMsg) {
+ if (result != ResultOk) {
+ LOG_ERROR("Filed when acknowledge list: " << result);
+ // set needCallBack is -1 to avoid repeated callback.
+ needCallBack->store(-1);
+ callback(result, false);
+ return;
+ }
+
+ if (hasMsg) {
+ hasMessageAvailable->store(hasMsg);
+ }
+
+ if (--(*needCallBack) == 0) {
+ callback(result, hasMessageAvailable->load() || self->incomingMessagesSize_ > 0);
+ }
+ });
+ });
}
diff --git a/lib/MultiTopicsConsumerImpl.h b/lib/MultiTopicsConsumerImpl.h
index 6763942..e92ec0e 100644
--- a/lib/MultiTopicsConsumerImpl.h
+++ b/lib/MultiTopicsConsumerImpl.h
@@ -53,18 +53,19 @@ using LookupServicePtr = std::shared_ptr<LookupService>;
class MultiTopicsConsumerImpl;
class MultiTopicsConsumerImpl : public ConsumerImplBase {
public:
- MultiTopicsConsumerImpl(ClientImplPtr client, TopicNamePtr topicName, int numPartitions,
+ MultiTopicsConsumerImpl(const ClientImplPtr& client, const TopicNamePtr& topicName, int numPartitions,
const std::string& subscriptionName, const ConsumerConfiguration& conf,
- LookupServicePtr lookupServicePtr, const ConsumerInterceptorsPtr& interceptors,
+ const LookupServicePtr& lookupServicePtr,
+ const ConsumerInterceptorsPtr& interceptors,
Commands::SubscriptionMode = Commands::SubscriptionModeDurable,
- boost::optional<MessageId> startMessageId = boost::none);
+ const boost::optional<MessageId>& startMessageId = boost::none);
- MultiTopicsConsumerImpl(ClientImplPtr client, const std::vector<std::string>& topics,
- const std::string& subscriptionName, TopicNamePtr topicName,
- const ConsumerConfiguration& conf, LookupServicePtr lookupServicePtr_,
+ MultiTopicsConsumerImpl(const ClientImplPtr& client, const std::vector<std::string>& topics,
+ const std::string& subscriptionName, const TopicNamePtr& topicName,
+ const ConsumerConfiguration& conf, const LookupServicePtr& lookupServicePtr_,
const ConsumerInterceptorsPtr& interceptors,
Commands::SubscriptionMode = Commands::SubscriptionModeDurable,
- boost::optional<MessageId> startMessageId = boost::none);
+ const boost::optional<MessageId>& startMessageId = boost::none);
~MultiTopicsConsumerImpl();
// overrided methods from ConsumerImplBase
@@ -73,14 +74,15 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase {
const std::string& getTopic() const override;
Result receive(Message& msg) override;
Result receive(Message& msg, int timeout) override;
- void receiveAsync(ReceiveCallback callback) override;
- void unsubscribeAsync(ResultCallback callback) override;
- void acknowledgeAsync(const MessageId& msgId, ResultCallback callback) override;
- void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) override;
- void acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) override;
- void closeAsync(ResultCallback callback) override;
+ void receiveAsync(const ReceiveCallback& callback) override;
+ void unsubscribeAsync(const ResultCallback& callback) override;
+ void acknowledgeAsync(const MessageId& msgId, const ResultCallback& callback) override;
+ void acknowledgeAsync(const MessageIdList& messageIdList, const ResultCallback& callback) override;
+ void acknowledgeCumulativeAsync(const MessageId& msgId, const ResultCallback& callback) override;
+ void closeAsync(const ResultCallback& callback) override;
void start() override;
void shutdown() override;
+ void internalShutdown();
bool isClosed() override;
bool isOpen() override;
Result pauseMessageListener() override;
@@ -89,20 +91,21 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase {
void redeliverUnacknowledgedMessages(const std::set<MessageId>& messageIds) override;
const std::string& getName() const override;
int getNumOfPrefetchedMessages() const override;
- void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) override;
- void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) override;
- void seekAsync(const MessageId& msgId, ResultCallback callback) override;
- void seekAsync(uint64_t timestamp, ResultCallback callback) override;
+ void getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback) override;
+ void getLastMessageIdAsync(const BrokerGetLastMessageIdCallback& callback) override;
+ void seekAsync(const MessageId& msgId, const ResultCallback& callback) override;
+ void seekAsync(uint64_t timestamp, const ResultCallback& callback) override;
void negativeAcknowledge(const MessageId& msgId) override;
bool isConnected() const override;
uint64_t getNumberOfConnectedConsumer() override;
- void hasMessageAvailableAsync(HasMessageAvailableCallback callback) override;
+ void hasMessageAvailableAsync(const HasMessageAvailableCallback& callback) override;
- void handleGetConsumerStats(Result, BrokerConsumerStats, LatchPtr, MultiTopicsBrokerConsumerStatsPtr,
- size_t, BrokerConsumerStatsCallback);
+ void handleGetConsumerStats(Result, const BrokerConsumerStats&, const LatchPtr&,
+ const MultiTopicsBrokerConsumerStatsPtr&, size_t,
+ const BrokerConsumerStatsCallback&);
// return first topic name when all topics name valid, or return null pointer
static std::shared_ptr<TopicName> topicNamesValid(const std::vector<std::string>& topics);
- void unsubscribeOneTopicAsync(const std::string& topic, ResultCallback callback);
+ void unsubscribeOneTopicAsync(const std::string& topic, const ResultCallback& callback);
Future<Result, Consumer> subscribeOneTopicAsync(const std::string& topic);
protected:
@@ -135,32 +138,35 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase {
/* methods */
void handleSinglePartitionConsumerCreated(Result result, ConsumerImplBaseWeakPtr consumerImplBaseWeakPtr,
unsigned int partitionIndex);
- void notifyResult(CloseCallback closeCallback);
- void messageReceived(Consumer consumer, const Message& msg);
+ void notifyResult(const CloseCallback& closeCallback);
+ void messageReceived(const Consumer& consumer, const Message& msg);
void messageProcessed(Message& msg);
- void internalListener(Consumer consumer);
+ void internalListener(const Consumer& consumer);
void receiveMessages();
void failPendingReceiveCallback();
void notifyPendingReceivedCallback(Result result, const Message& message,
const ReceiveCallback& callback);
- void handleOneTopicSubscribed(Result result, Consumer consumer, const std::string& topic,
- std::shared_ptr<std::atomic<int>> topicsNeedCreate);
- void subscribeTopicPartitions(int numPartitions, TopicNamePtr topicName, const std::string& consumerName,
- ConsumerSubResultPromisePtr topicSubResultPromise);
- void handleSingleConsumerCreated(Result result, ConsumerImplBaseWeakPtr consumerImplBaseWeakPtr,
- std::shared_ptr<std::atomic<int>> partitionsNeedCreate,
- ConsumerSubResultPromisePtr topicSubResultPromise);
- void handleOneTopicUnsubscribedAsync(Result result, std::shared_ptr<std::atomic<int>> consumerUnsubed,
- int numberPartitions, TopicNamePtr topicNamePtr,
- std::string& topicPartitionName, ResultCallback callback);
+ void handleOneTopicSubscribed(Result result, const Consumer& consumer, const std::string& topic,
+ const std::shared_ptr<std::atomic<int>>& topicsNeedCreate);
+ void subscribeTopicPartitions(int numPartitions, const TopicNamePtr& topicName,
+ const std::string& consumerName,
+ const ConsumerSubResultPromisePtr& topicSubResultPromise);
+ void handleSingleConsumerCreated(Result result, const ConsumerImplBaseWeakPtr& consumerImplBaseWeakPtr,
+ const std::shared_ptr<std::atomic<int>>& partitionsNeedCreate,
+ const ConsumerSubResultPromisePtr& topicSubResultPromise);
+ void handleOneTopicUnsubscribedAsync(Result result,
+ const std::shared_ptr<std::atomic<int>>& consumerUnsubed,
+ int numberPartitions, const TopicNamePtr& topicNamePtr,
+ const std::string& topicPartitionName,
+ const ResultCallback& callback);
void runPartitionUpdateTask();
void topicPartitionUpdate();
- void handleGetPartitions(TopicNamePtr topicName, Result result,
+ void handleGetPartitions(const TopicNamePtr& topicName, Result result,
const LookupDataResultPtr& lookupDataResult, int currentNumPartitions);
- void subscribeSingleNewConsumer(int numPartitions, TopicNamePtr topicName, int partitionIndex,
- ConsumerSubResultPromisePtr topicSubResultPromise,
- std::shared_ptr<std::atomic<int>> partitionsNeedCreate);
+ void subscribeSingleNewConsumer(int numPartitions, const TopicNamePtr& topicName, int partitionIndex,
+ const ConsumerSubResultPromisePtr& topicSubResultPromise,
+ const std::shared_ptr<std::atomic<int>>& partitionsNeedCreate);
// impl consumer base virtual method
bool hasEnoughMessagesForBatchReceive() const override;
void notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) override;
@@ -181,7 +187,7 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase {
requires std::convertible_to<SeekArg, uint64_t> ||
std::same_as<std::remove_cv_t<std::remove_reference_t<SeekArg>>, MessageId>
#endif
- void seekAllAsync(const SeekArg& seekArg, ResultCallback callback);
+ void seekAllAsync(const SeekArg& seekArg, const ResultCallback& callback);
void beforeSeek();
void afterSeek();
@@ -200,7 +206,8 @@ template <typename SeekArg>
requires std::convertible_to<SeekArg, uint64_t> ||
std::same_as<std::remove_cv_t<std::remove_reference_t<SeekArg>>, MessageId>
#endif
- inline void MultiTopicsConsumerImpl::seekAllAsync(const SeekArg& seekArg, ResultCallback callback) {
+ inline void MultiTopicsConsumerImpl::seekAllAsync(const SeekArg& seekArg,
+ const ResultCallback& callback) {
if (state_ != Ready) {
callback(ResultAlreadyClosed);
return;
@@ -209,7 +216,8 @@ template <typename SeekArg>
auto weakSelf = weak_from_this();
auto failed = std::make_shared<std::atomic_bool>(false);
consumers_.forEachValue(
- [this, weakSelf, &seekArg, callback, failed](const ConsumerImplPtr& consumer, SharedFuture future) {
+ [this, weakSelf, &seekArg, callback, failed](const ConsumerImplPtr& consumer,
+ const SharedFuture& future) {
consumer->seekAsync(seekArg, [this, weakSelf, callback, failed, future](Result result) {
auto self = weakSelf.lock();
if (!self || failed->load(std::memory_order_acquire)) {
diff --git a/lib/NegativeAcksTracker.cc b/lib/NegativeAcksTracker.cc
index e443496..7116b1c 100644
--- a/lib/NegativeAcksTracker.cc
+++ b/lib/NegativeAcksTracker.cc
@@ -31,7 +31,7 @@ DECLARE_LOG_OBJECT()
namespace pulsar {
-NegativeAcksTracker::NegativeAcksTracker(ClientImplPtr client, ConsumerImpl &consumer,
+NegativeAcksTracker::NegativeAcksTracker(const ClientImplPtr &client, ConsumerImpl &consumer,
const ConsumerConfiguration &conf)
: consumer_(consumer),
timerInterval_(0),
diff --git a/lib/NegativeAcksTracker.h b/lib/NegativeAcksTracker.h
index 472e976..bf1d931 100644
--- a/lib/NegativeAcksTracker.h
+++ b/lib/NegativeAcksTracker.h
@@ -42,7 +42,8 @@ using ExecutorServicePtr = std::shared_ptr<ExecutorService>;
class NegativeAcksTracker : public std::enable_shared_from_this<NegativeAcksTracker> {
public:
- NegativeAcksTracker(ClientImplPtr client, ConsumerImpl &consumer, const ConsumerConfiguration &conf);
+ NegativeAcksTracker(const ClientImplPtr &client, ConsumerImpl &consumer,
+ const ConsumerConfiguration &conf);
NegativeAcksTracker(const NegativeAcksTracker &) = delete;
diff --git a/lib/PartitionedProducerImpl.cc b/lib/PartitionedProducerImpl.cc
index 4178096..9b2f5c6 100644
--- a/lib/PartitionedProducerImpl.cc
+++ b/lib/PartitionedProducerImpl.cc
@@ -36,8 +36,8 @@ namespace pulsar {
const std::string PartitionedProducerImpl::PARTITION_NAME_SUFFIX = "-partition-";
-PartitionedProducerImpl::PartitionedProducerImpl(ClientImplPtr client, const TopicNamePtr topicName,
- const unsigned int numPartitions,
+PartitionedProducerImpl::PartitionedProducerImpl(const ClientImplPtr& client, const TopicNamePtr& topicName,
+ unsigned int numPartitions,
const ProducerConfiguration& config,
const ProducerInterceptorsPtr& interceptors)
: client_(client),
@@ -79,7 +79,7 @@ MessageRoutingPolicyPtr PartitionedProducerImpl::getMessageRouter() {
}
}
-PartitionedProducerImpl::~PartitionedProducerImpl() { shutdown(); }
+PartitionedProducerImpl::~PartitionedProducerImpl() { internalShutdown(); }
// override
const std::string& PartitionedProducerImpl::getTopic() const { return topic_; }
@@ -144,9 +144,8 @@ void PartitionedProducerImpl::start() {
}
}
-void PartitionedProducerImpl::handleSinglePartitionProducerCreated(Result result,
- ProducerImplBaseWeakPtr producerWeakPtr,
- unsigned int partitionIndex) {
+void PartitionedProducerImpl::handleSinglePartitionProducerCreated(
+ Result result, const ProducerImplBaseWeakPtr& producerWeakPtr, unsigned int partitionIndex) {
// to indicate, we are doing cleanup using closeAsync after producer create
// has failed and the invocation of closeAsync is not from client
const auto numPartitions = getNumPartitionsWithLock();
@@ -234,9 +233,9 @@ void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callbac
} else {
// Wrapping the callback into a lambda has overhead, so we check if the producer is ready first
producer->getProducerCreatedFuture().addListener(
- [msg, callback](Result result, ProducerImplBaseWeakPtr weakProducer) {
+ [msg, callback](Result result, const ProducerImplBaseWeakPtr& weakProducer) {
if (result == ResultOk) {
- weakProducer.lock()->sendAsync(msg, std::move(callback));
+ weakProducer.lock()->sendAsync(msg, callback);
} else if (callback) {
callback(result, {});
}
@@ -245,7 +244,9 @@ void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callbac
}
// override
-void PartitionedProducerImpl::shutdown() {
+void PartitionedProducerImpl::shutdown() { internalShutdown(); }
+
+void PartitionedProducerImpl::internalShutdown() {
cancelTimers();
interceptors_->close();
auto client = client_.lock();
@@ -285,7 +286,7 @@ int64_t PartitionedProducerImpl::getLastSequenceId() const {
void PartitionedProducerImpl::closeAsync(CloseCallback originalCallback) {
auto closeCallback = [this, originalCallback](Result result) {
if (result == ResultOk) {
- shutdown();
+ internalShutdown();
}
if (originalCallback) {
originalCallback(result);
@@ -331,9 +332,8 @@ void PartitionedProducerImpl::closeAsync(CloseCallback originalCallback) {
}
// `callback` is a wrapper of user provided callback, it's not null and will call `shutdown()`
-void PartitionedProducerImpl::handleSinglePartitionProducerClose(Result result,
- const unsigned int partitionIndex,
- CloseCallback callback) {
+void PartitionedProducerImpl::handleSinglePartitionProducerClose(Result result, unsigned int partitionIndex,
+ const CloseCallback& callback) {
if (state_ == Failed) {
// we should have already notified the client by callback
return;
diff --git a/lib/PartitionedProducerImpl.h b/lib/PartitionedProducerImpl.h
index 610c74e..40f2d34 100644
--- a/lib/PartitionedProducerImpl.h
+++ b/lib/PartitionedProducerImpl.h
@@ -20,6 +20,7 @@
#include <pulsar/TopicMetadata.h>
#include <atomic>
+#include <cstdint>
#include <memory>
#include <mutex>
#include <vector>
@@ -47,7 +48,7 @@ using TopicNamePtr = std::shared_ptr<TopicName>;
class PartitionedProducerImpl : public ProducerImplBase,
public std::enable_shared_from_this<PartitionedProducerImpl> {
public:
- enum State
+ enum State : uint8_t
{
Pending,
Ready,
@@ -59,8 +60,9 @@ class PartitionedProducerImpl : public ProducerImplBase,
typedef std::unique_lock<std::mutex> Lock;
- PartitionedProducerImpl(ClientImplPtr ptr, const TopicNamePtr topicName, const unsigned int numPartitions,
- const ProducerConfiguration& config, const ProducerInterceptorsPtr& interceptors);
+ PartitionedProducerImpl(const ClientImplPtr& ptr, const TopicNamePtr& topicName,
+ unsigned int numPartitions, const ProducerConfiguration& config,
+ const ProducerInterceptorsPtr& interceptors);
virtual ~PartitionedProducerImpl();
// overrided methods from ProducerImplBase
@@ -75,6 +77,7 @@ class PartitionedProducerImpl : public ProducerImplBase,
void closeAsync(CloseCallback callback) override;
void start() override;
void shutdown() override;
+ void internalShutdown();
bool isClosed() override;
const std::string& getTopic() const override;
Future<Result, ProducerImplBaseWeakPtr> getProducerCreatedFuture() override;
@@ -82,11 +85,12 @@ class PartitionedProducerImpl : public ProducerImplBase,
void flushAsync(FlushCallback callback) override;
bool isConnected() const override;
uint64_t getNumberOfConnectedProducer() override;
- void handleSinglePartitionProducerCreated(Result result, ProducerImplBaseWeakPtr producerBaseWeakPtr,
+ void handleSinglePartitionProducerCreated(Result result,
+ const ProducerImplBaseWeakPtr& producerBaseWeakPtr,
const unsigned int partitionIndex);
void createLazyPartitionProducer(const unsigned int partitionIndex);
- void handleSinglePartitionProducerClose(Result result, const unsigned int partitionIndex,
- CloseCallback callback);
+ void handleSinglePartitionProducerClose(Result result, unsigned int partitionIndex,
+ const CloseCallback& callback);
void notifyResult(CloseCallback closeCallback);
diff --git a/lib/PatternMultiTopicsConsumerImpl.cc b/lib/PatternMultiTopicsConsumerImpl.cc
index 73aa874..9f3fbb9 100644
--- a/lib/PatternMultiTopicsConsumerImpl.cc
+++ b/lib/PatternMultiTopicsConsumerImpl.cc
@@ -30,10 +30,10 @@ using namespace pulsar;
using std::chrono::seconds;
PatternMultiTopicsConsumerImpl::PatternMultiTopicsConsumerImpl(
- ClientImplPtr client, const std::string pattern, CommandGetTopicsOfNamespace_Mode getTopicsMode,
+ const ClientImplPtr& client, const std::string& pattern, CommandGetTopicsOfNamespace_Mode getTopicsMode,
const std::vector<std::string>& topics, const std::string& subscriptionName,
- const ConsumerConfiguration& conf, const LookupServicePtr lookupServicePtr_,
- const ConsumerInterceptorsPtr interceptors)
+ const ConsumerConfiguration& conf, const LookupServicePtr& lookupServicePtr_,
+ const ConsumerInterceptorsPtr& interceptors)
: MultiTopicsConsumerImpl(client, topics, subscriptionName, TopicName::get(pattern), conf,
lookupServicePtr_, interceptors),
patternString_(pattern),
@@ -89,8 +89,8 @@ void PatternMultiTopicsConsumerImpl::autoDiscoveryTimerTask(const ASIO_ERROR& er
std::placeholders::_1, std::placeholders::_2));
}
-void PatternMultiTopicsConsumerImpl::timerGetTopicsOfNamespace(const Result result,
- const NamespaceTopicsPtr topics) {
+void PatternMultiTopicsConsumerImpl::timerGetTopicsOfNamespace(Result result,
+ const NamespaceTopicsPtr& topics) {
if (result != ResultOk) {
LOG_ERROR("Error in Getting topicsOfNameSpace. result: " << result);
resetAutoDiscoveryTimer();
@@ -132,7 +132,8 @@ void PatternMultiTopicsConsumerImpl::timerGetTopicsOfNamespace(const Result resu
onTopicsAdded(topicsAdded, topicsAddedCallback);
}
-void PatternMultiTopicsConsumerImpl::onTopicsAdded(NamespaceTopicsPtr addedTopics, ResultCallback callback) {
+void PatternMultiTopicsConsumerImpl::onTopicsAdded(const NamespaceTopicsPtr& addedTopics,
+ const ResultCallback& callback) {
// start call subscribeOneTopicAsync for each single topic
if (addedTopics->empty()) {
@@ -152,9 +153,9 @@ void PatternMultiTopicsConsumerImpl::onTopicsAdded(NamespaceTopicsPtr addedTopic
}
}
-void PatternMultiTopicsConsumerImpl::handleOneTopicAdded(const Result result, const std::string& topic,
- std::shared_ptr<std::atomic<int>> topicsNeedCreate,
- ResultCallback callback) {
+void PatternMultiTopicsConsumerImpl::handleOneTopicAdded(
+ Result result, const std::string& topic, const std::shared_ptr<std::atomic<int>>& topicsNeedCreate,
+ const ResultCallback& callback) {
(*topicsNeedCreate)--;
if (result != ResultOk) {
@@ -169,8 +170,8 @@ void PatternMultiTopicsConsumerImpl::handleOneTopicAdded(const Result result, co
}
}
-void PatternMultiTopicsConsumerImpl::onTopicsRemoved(NamespaceTopicsPtr removedTopics,
- ResultCallback callback) {
+void PatternMultiTopicsConsumerImpl::onTopicsRemoved(const NamespaceTopicsPtr& removedTopics,
+ const ResultCallback& callback) {
// start call subscribeOneTopicAsync for each single topic
if (removedTopics->empty()) {
LOG_DEBUG("no topics need unsubscribe");
@@ -208,7 +209,7 @@ NamespaceTopicsPtr PatternMultiTopicsConsumerImpl::topicsPatternFilter(
for (const auto& topicStr : topics) {
auto topic = TopicName::removeDomain(topicStr);
if (PULSAR_REGEX_NAMESPACE::regex_match(topic, pattern)) {
- topicsResultPtr->push_back(std::move(topicStr));
+ topicsResultPtr->push_back(topicStr);
}
}
return topicsResultPtr;
@@ -241,12 +242,12 @@ void PatternMultiTopicsConsumerImpl::start() {
}
}
-void PatternMultiTopicsConsumerImpl::shutdown() {
+PatternMultiTopicsConsumerImpl::~PatternMultiTopicsConsumerImpl() {
cancelTimers();
- MultiTopicsConsumerImpl::shutdown();
+ internalShutdown();
}
-void PatternMultiTopicsConsumerImpl::closeAsync(ResultCallback callback) {
+void PatternMultiTopicsConsumerImpl::closeAsync(const ResultCallback& callback) {
cancelTimers();
MultiTopicsConsumerImpl::closeAsync(callback);
}
diff --git a/lib/PatternMultiTopicsConsumerImpl.h b/lib/PatternMultiTopicsConsumerImpl.h
index f272df2..6352796 100644
--- a/lib/PatternMultiTopicsConsumerImpl.h
+++ b/lib/PatternMultiTopicsConsumerImpl.h
@@ -48,12 +48,13 @@ class PatternMultiTopicsConsumerImpl : public MultiTopicsConsumerImpl {
// which only contains after namespace part.
// when subscribe, client will first get all topics that match given pattern.
// `topics` contains the topics that match `patternString`.
- PatternMultiTopicsConsumerImpl(ClientImplPtr client, const std::string patternString,
+ PatternMultiTopicsConsumerImpl(const ClientImplPtr& client, const std::string& patternString,
CommandGetTopicsOfNamespace_Mode getTopicsMode,
const std::vector<std::string>& topics,
const std::string& subscriptionName, const ConsumerConfiguration& conf,
- const LookupServicePtr lookupServicePtr_,
- const ConsumerInterceptorsPtr interceptors);
+ const LookupServicePtr& lookupServicePtr_,
+ const ConsumerInterceptorsPtr& interceptors);
+ ~PatternMultiTopicsConsumerImpl() override;
const PULSAR_REGEX_NAMESPACE::regex getPattern();
@@ -67,9 +68,8 @@ class PatternMultiTopicsConsumerImpl : public MultiTopicsConsumerImpl {
static NamespaceTopicsPtr topicsListsMinus(std::vector<std::string>& list1,
std::vector<std::string>& list2);
- virtual void closeAsync(ResultCallback callback);
- virtual void start();
- virtual void shutdown();
+ void closeAsync(const ResultCallback& callback) override;
+ void start() override;
private:
const std::string patternString_;
@@ -82,11 +82,12 @@ class PatternMultiTopicsConsumerImpl : public MultiTopicsConsumerImpl {
void cancelTimers() noexcept;
void resetAutoDiscoveryTimer();
- void timerGetTopicsOfNamespace(const Result result, const NamespaceTopicsPtr topics);
- void onTopicsAdded(NamespaceTopicsPtr addedTopics, ResultCallback callback);
- void onTopicsRemoved(NamespaceTopicsPtr removedTopics, ResultCallback callback);
- void handleOneTopicAdded(const Result result, const std::string& topic,
- std::shared_ptr<std::atomic<int>> topicsNeedCreate, ResultCallback callback);
+ void timerGetTopicsOfNamespace(Result result, const NamespaceTopicsPtr& topics);
+ void onTopicsAdded(const NamespaceTopicsPtr& addedTopics, const ResultCallback& callback);
+ void onTopicsRemoved(const NamespaceTopicsPtr& removedTopics, const ResultCallback& callback);
+ void handleOneTopicAdded(Result result, const std::string& topic,
+ const std::shared_ptr<std::atomic<int>>& topicsNeedCreate,
+ const ResultCallback& callback);
std::weak_ptr<PatternMultiTopicsConsumerImpl> weak_from_this() noexcept {
return std::static_pointer_cast<PatternMultiTopicsConsumerImpl>(shared_from_this());
diff --git a/lib/Producer.cc b/lib/Producer.cc
index fc588ba..1adf23c 100644
--- a/lib/Producer.cc
+++ b/lib/Producer.cc
@@ -28,7 +28,7 @@ static const std::string EMPTY_STRING;
Producer::Producer() : impl_() {}
-Producer::Producer(ProducerImplBasePtr impl) : impl_(impl) {}
+Producer::Producer(const ProducerImplBasePtr& impl) : impl_(impl) {}
const std::string& Producer::getTopic() const { return impl_ != NULL ? impl_->getTopic() : EMPTY_STRING; }
@@ -58,7 +58,7 @@ Result Producer::send(const Message& msg, MessageId& messageId) {
return promise.getFuture().get(messageId);
}
-void Producer::sendAsync(const Message& msg, SendCallback callback) {
+void Producer::sendAsync(const Message& msg, const SendCallback& callback) {
if (!impl_) {
callback(ResultProducerNotInitialized, msg.getMessageId());
return;
@@ -82,7 +82,7 @@ Result Producer::close() {
return result;
}
-void Producer::closeAsync(CloseCallback callback) {
+void Producer::closeAsync(const CloseCallback& callback) {
if (!impl_) {
callback(ResultProducerNotInitialized);
return;
@@ -100,7 +100,7 @@ Result Producer::flush() {
return result;
}
-void Producer::flushAsync(FlushCallback callback) {
+void Producer::flushAsync(const FlushCallback& callback) {
if (!impl_) {
callback(ResultProducerNotInitialized);
return;
diff --git a/lib/ProducerConfiguration.cc b/lib/ProducerConfiguration.cc
index 0e141ae..278871c 100644
--- a/lib/ProducerConfiguration.cc
+++ b/lib/ProducerConfiguration.cc
@@ -180,7 +180,7 @@ ProducerConfiguration::BatchingType ProducerConfiguration::getBatchingType() con
const CryptoKeyReaderPtr ProducerConfiguration::getCryptoKeyReader() const { return impl_->cryptoKeyReader; }
ProducerConfiguration& ProducerConfiguration::setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader) {
- impl_->cryptoKeyReader = cryptoKeyReader;
+ impl_->cryptoKeyReader = std::move(cryptoKeyReader);
return *this;
}
@@ -201,7 +201,7 @@ bool ProducerConfiguration::isEncryptionEnabled() const {
return (!impl_->encryptionKeys.empty() && (impl_->cryptoKeyReader != NULL));
}
-ProducerConfiguration& ProducerConfiguration::addEncryptionKey(std::string key) {
+ProducerConfiguration& ProducerConfiguration::addEncryptionKey(const std::string& key) {
impl_->encryptionKeys.insert(key);
return *this;
}
diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
index 4399ce5..cc8f320 100644
--- a/lib/ProducerImpl.cc
+++ b/lib/ProducerImpl.cc
@@ -48,7 +48,7 @@ DECLARE_LOG_OBJECT()
using std::chrono::milliseconds;
-ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName,
+ProducerImpl::ProducerImpl(const ClientImplPtr& client, const TopicName& topicName,
const ProducerConfiguration& conf, const ProducerInterceptorsPtr& interceptors,
int32_t partition, bool retryOnCreationError)
: HandlerBase(client, (partition < 0) ? topicName.toString() : topicName.getTopicPartitionName(partition),
@@ -114,11 +114,11 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName,
}
ProducerImpl::~ProducerImpl() {
- LOG_DEBUG(getName() << "~ProducerImpl");
- shutdown();
+ LOG_DEBUG(producerStr_ << "~ProducerImpl");
+ internalShutdown();
printStats();
if (state_ == Ready || state_ == Pending) {
- LOG_WARN(getName() << "Destroyed producer which was not properly closed");
+ LOG_WARN(producerStr_ << "Destroyed producer which was not properly closed");
}
}
@@ -352,7 +352,7 @@ void ProducerImpl::failPendingMessages(Result result, bool withLock) {
}
}
-void ProducerImpl::resendMessages(ClientConnectionPtr cnx) {
+void ProducerImpl::resendMessages(const ClientConnectionPtr& cnx) {
if (pendingMessagesQueue_.empty()) {
return;
}
@@ -879,7 +879,7 @@ bool ProducerImpl::removeCorruptMessage(uint64_t sequenceId) {
return true;
}
- std::unique_ptr<OpSendMsg> op{std::move(pendingMessagesQueue_.front().release())};
+ std::unique_ptr<OpSendMsg> op{pendingMessagesQueue_.front().release()};
uint64_t expectedSequenceId = op->sendArgs->sequenceId;
if (sequenceId > expectedSequenceId) {
LOG_WARN(getName() << "Got ack failure for msg " << sequenceId //
@@ -993,7 +993,9 @@ void ProducerImpl::start() {
}
}
-void ProducerImpl::shutdown() {
+void ProducerImpl::shutdown() { internalShutdown(); }
+
+void ProducerImpl::internalShutdown() {
resetCnx();
interceptors_->close();
auto client = client_.lock();
diff --git a/lib/ProducerImpl.h b/lib/ProducerImpl.h
index f650b1f..77bd6d1 100644
--- a/lib/ProducerImpl.h
+++ b/lib/ProducerImpl.h
@@ -71,7 +71,7 @@ class MessageMetadata;
class ProducerImpl : public HandlerBase, public ProducerImplBase {
public:
- ProducerImpl(ClientImplPtr client, const TopicName& topic,
+ ProducerImpl(const ClientImplPtr& client, const TopicName& topic,
const ProducerConfiguration& producerConfiguration,
const ProducerInterceptorsPtr& interceptors, int32_t partition = -1,
bool retryOnCreationError = false);
@@ -85,6 +85,7 @@ class ProducerImpl : public HandlerBase, public ProducerImplBase {
void closeAsync(CloseCallback callback) override;
void start() override;
void shutdown() override;
+ void internalShutdown();
bool isClosed() override;
const std::string& getTopic() const override;
Future<Result, ProducerImplBaseWeakPtr> getProducerCreatedFuture() override;
@@ -143,7 +144,7 @@ class ProducerImpl : public HandlerBase, public ProducerImplBase {
Result handleCreateProducer(const ClientConnectionPtr& cnx, Result result,
const ResponseData& responseData);
- void resendMessages(ClientConnectionPtr cnx);
+ void resendMessages(const ClientConnectionPtr& cnx);
void refreshEncryptionKey(const ASIO_ERROR& ec);
bool encryptMessage(proto::MessageMetadata& metadata, SharedBuffer& payload,
diff --git a/lib/ProducerImplBase.h b/lib/ProducerImplBase.h
index 27cfced..25a12c8 100644
--- a/lib/ProducerImplBase.h
+++ b/lib/ProducerImplBase.h
@@ -41,8 +41,8 @@ class ProducerImplBase {
virtual void sendAsync(const Message& msg, SendCallback callback) = 0;
virtual void closeAsync(CloseCallback callback) = 0;
virtual void start() = 0;
- virtual void shutdown() = 0;
virtual bool isClosed() = 0;
+ virtual void shutdown() = 0;
virtual const std::string& getTopic() const = 0;
virtual Future<Result, ProducerImplBaseWeakPtr> getProducerCreatedFuture() = 0;
virtual void triggerFlush() = 0;
diff --git a/lib/ProtobufNativeSchema.cc b/lib/ProtobufNativeSchema.cc
index 5cddf74..632943d 100644
--- a/lib/ProtobufNativeSchema.cc
+++ b/lib/ProtobufNativeSchema.cc
@@ -39,8 +39,8 @@ SchemaInfo createProtobufNativeSchema(const google::protobuf::Descriptor* descri
}
const auto fileDescriptor = descriptor->file();
- const std::string rootMessageTypeName = descriptor->full_name();
- const std::string rootFileDescriptorName = fileDescriptor->name();
+ const std::string& rootMessageTypeName = descriptor->full_name();
+ const std::string& rootFileDescriptorName = fileDescriptor->name();
FileDescriptorSet fileDescriptorSet;
internalCollectFileDescriptors(fileDescriptor, fileDescriptorSet);
diff --git a/lib/Reader.cc b/lib/Reader.cc
index c02fb2e..e7d907a 100644
--- a/lib/Reader.cc
+++ b/lib/Reader.cc
@@ -29,7 +29,7 @@ static const std::string EMPTY_STRING;
Reader::Reader() : impl_() {}
-Reader::Reader(ReaderImplPtr impl) : impl_(impl) {}
+Reader::Reader(const ReaderImplPtr& impl) : impl_(impl) {}
const std::string& Reader::getTopic() const { return impl_ != NULL ? impl_->getTopic() : EMPTY_STRING; }
@@ -49,7 +49,7 @@ Result Reader::readNext(Message& msg, int timeoutMs) {
return impl_->readNext(msg, timeoutMs);
}
-void Reader::readNextAsync(ReadNextCallback callback) {
+void Reader::readNextAsync(const ReadNextCallback& callback) {
if (!impl_) {
return callback(ResultConsumerNotInitialized, {});
}
@@ -66,7 +66,7 @@ Result Reader::close() {
return result;
}
-void Reader::closeAsync(ResultCallback callback) {
+void Reader::closeAsync(const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -75,7 +75,7 @@ void Reader::closeAsync(ResultCallback callback) {
impl_->closeAsync(callback);
}
-void Reader::hasMessageAvailableAsync(HasMessageAvailableCallback callback) {
+void Reader::hasMessageAvailableAsync(const HasMessageAvailableCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized, false);
return;
@@ -91,7 +91,7 @@ Result Reader::hasMessageAvailable(bool& hasMessageAvailable) {
return promise.getFuture().get(hasMessageAvailable);
}
-void Reader::seekAsync(const MessageId& msgId, ResultCallback callback) {
+void Reader::seekAsync(const MessageId& msgId, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -99,7 +99,7 @@ void Reader::seekAsync(const MessageId& msgId, ResultCallback callback) {
impl_->seekAsync(msgId, callback);
}
-void Reader::seekAsync(uint64_t timestamp, ResultCallback callback) {
+void Reader::seekAsync(uint64_t timestamp, const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
@@ -125,7 +125,7 @@ Result Reader::seek(uint64_t timestamp) {
bool Reader::isConnected() const { return impl_ && impl_->isConnected(); }
-void Reader::getLastMessageIdAsync(GetLastMessageIdCallback callback) {
+void Reader::getLastMessageIdAsync(const GetLastMessageIdCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized, MessageId());
return;
diff --git a/lib/ReaderConfiguration.cc b/lib/ReaderConfiguration.cc
index f1dba5e..054010f 100644
--- a/lib/ReaderConfiguration.cc
+++ b/lib/ReaderConfiguration.cc
@@ -41,7 +41,7 @@ ReaderConfiguration& ReaderConfiguration::setSchema(const SchemaInfo& schemaInfo
const SchemaInfo& ReaderConfiguration::getSchema() const { return impl_->schemaInfo; }
ReaderConfiguration& ReaderConfiguration::setReaderListener(ReaderListener readerListener) {
- impl_->readerListener = readerListener;
+ impl_->readerListener = std::move(readerListener);
impl_->hasReaderListener = true;
return *this;
}
@@ -71,7 +71,7 @@ bool ReaderConfiguration::isReadCompacted() const { return impl_->readCompacted;
void ReaderConfiguration::setReadCompacted(bool compacted) { impl_->readCompacted = compacted; }
void ReaderConfiguration::setInternalSubscriptionName(std::string internalSubscriptionName) {
- impl_->internalSubscriptionName = internalSubscriptionName;
+ impl_->internalSubscriptionName = std::move(internalSubscriptionName);
}
const std::string& ReaderConfiguration::getInternalSubscriptionName() const {
@@ -107,7 +107,7 @@ bool ReaderConfiguration::isEncryptionEnabled() const { return impl_->cryptoKeyR
const CryptoKeyReaderPtr ReaderConfiguration::getCryptoKeyReader() const { return impl_->cryptoKeyReader; }
ReaderConfiguration& ReaderConfiguration::setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader) {
- impl_->cryptoKeyReader = cryptoKeyReader;
+ impl_->cryptoKeyReader = std::move(cryptoKeyReader);
return *this;
}
diff --git a/lib/ReaderImpl.cc b/lib/ReaderImpl.cc
index f41106e..7fa7e8b 100644
--- a/lib/ReaderImpl.cc
+++ b/lib/ReaderImpl.cc
@@ -36,9 +36,9 @@ ConsumerConfiguration consumerConfigOfReader;
static ResultCallback emptyCallback;
-ReaderImpl::ReaderImpl(const ClientImplPtr client, const std::string& topic, int partitions,
- const ReaderConfiguration& conf, const ExecutorServicePtr listenerExecutor,
- ReaderCallback readerCreatedCallback)
+ReaderImpl::ReaderImpl(const ClientImplPtr& client, const std::string& topic, int partitions,
+ const ReaderConfiguration& conf, const ExecutorServicePtr& listenerExecutor,
+ const ReaderCallback& readerCreatedCallback)
: topic_(topic),
partitions_(partitions),
client_(client),
@@ -46,7 +46,7 @@ ReaderImpl::ReaderImpl(const ClientImplPtr client, const std::string& topic, int
readerCreatedCallback_(readerCreatedCallback) {}
void ReaderImpl::start(const MessageId& startMessageId,
- std::function<void(const ConsumerImplBaseWeakPtr&)> callback) {
+ const std::function<void(const ConsumerImplBaseWeakPtr&)>& callback) {
ConsumerConfiguration consumerConf;
consumerConf.setConsumerType(ConsumerExclusive);
consumerConf.setReceiverQueueSize(readerConf_.getReceiverQueueSize());
@@ -130,7 +130,7 @@ Result ReaderImpl::readNext(Message& msg, int timeoutMs) {
return res;
}
-void ReaderImpl::readNextAsync(ReceiveCallback callback) {
+void ReaderImpl::readNextAsync(const ReceiveCallback& callback) {
auto self = shared_from_this();
consumer_->receiveAsync([self, callback](Result result, const Message& message) {
self->acknowledgeIfNecessary(result, message);
@@ -138,7 +138,7 @@ void ReaderImpl::readNextAsync(ReceiveCallback callback) {
});
}
-void ReaderImpl::messageListener(Consumer consumer, const Message& msg) {
+void ReaderImpl::messageListener(const Consumer& consumer, const Message& msg) {
readerListener_(Reader(shared_from_this()), msg);
acknowledgeIfNecessary(ResultOk, msg);
}
@@ -156,20 +156,20 @@ void ReaderImpl::acknowledgeIfNecessary(Result result, const Message& msg) {
}
}
-void ReaderImpl::closeAsync(ResultCallback callback) { consumer_->closeAsync(callback); }
+void ReaderImpl::closeAsync(const ResultCallback& callback) { consumer_->closeAsync(callback); }
-void ReaderImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback) {
+void ReaderImpl::hasMessageAvailableAsync(const HasMessageAvailableCallback& callback) {
consumer_->hasMessageAvailableAsync(callback);
}
-void ReaderImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
+void ReaderImpl::seekAsync(const MessageId& msgId, const ResultCallback& callback) {
consumer_->seekAsync(msgId, callback);
}
-void ReaderImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
+void ReaderImpl::seekAsync(uint64_t timestamp, const ResultCallback& callback) {
consumer_->seekAsync(timestamp, callback);
}
-void ReaderImpl::getLastMessageIdAsync(GetLastMessageIdCallback callback) {
+void ReaderImpl::getLastMessageIdAsync(const GetLastMessageIdCallback& callback) {
consumer_->getLastMessageIdAsync([callback](Result result, const GetLastMessageIdResponse& response) {
callback(result, response.getLastMessageId());
});
diff --git a/lib/ReaderImpl.h b/lib/ReaderImpl.h
index 3731990..020a503 100644
--- a/lib/ReaderImpl.h
+++ b/lib/ReaderImpl.h
@@ -58,35 +58,36 @@ extern PULSAR_PUBLIC ConsumerConfiguration consumerConfigOfReader;
class PULSAR_PUBLIC ReaderImpl : public std::enable_shared_from_this<ReaderImpl> {
public:
- ReaderImpl(const ClientImplPtr client, const std::string& topic, int partitions,
- const ReaderConfiguration& conf, const ExecutorServicePtr listenerExecutor,
- ReaderCallback readerCreatedCallback);
+ ReaderImpl(const ClientImplPtr& client, const std::string& topic, int partitions,
+ const ReaderConfiguration& conf, const ExecutorServicePtr& listenerExecutor,
+ const ReaderCallback& readerCreatedCallback);
- void start(const MessageId& startMessageId, std::function<void(const ConsumerImplBaseWeakPtr&)> callback);
+ void start(const MessageId& startMessageId,
+ const std::function<void(const ConsumerImplBaseWeakPtr&)>& callback);
const std::string& getTopic() const;
Result readNext(Message& msg);
Result readNext(Message& msg, int timeoutMs);
- void readNextAsync(ReceiveCallback callback);
+ void readNextAsync(const ReceiveCallback& callback);
- void closeAsync(ResultCallback callback);
+ void closeAsync(const ResultCallback& callback);
Future<Result, ReaderImplWeakPtr> getReaderCreatedFuture();
ConsumerImplBasePtr getConsumer() const noexcept { return consumer_; }
- void hasMessageAvailableAsync(HasMessageAvailableCallback callback);
+ void hasMessageAvailableAsync(const HasMessageAvailableCallback& callback);
- void seekAsync(const MessageId& msgId, ResultCallback callback);
- void seekAsync(uint64_t timestamp, ResultCallback callback);
+ void seekAsync(const MessageId& msgId, const ResultCallback& callback);
+ void seekAsync(uint64_t timestamp, const ResultCallback& callback);
- void getLastMessageIdAsync(GetLastMessageIdCallback callback);
+ void getLastMessageIdAsync(const GetLastMessageIdCallback& callback);
bool isConnected() const;
private:
- void messageListener(Consumer consumer, const Message& msg);
+ void messageListener(const Consumer& consumer, const Message& msg);
void acknowledgeIfNecessary(Result result, const Message& msg);
diff --git a/lib/Schema.cc b/lib/Schema.cc
index c8f8b4d..4777776 100644
--- a/lib/Schema.cc
+++ b/lib/Schema.cc
@@ -53,7 +53,7 @@ PULSAR_PUBLIC const char *strEncodingType(KeyValueEncodingType encodingType) {
return "UnknownSchemaType";
}
-PULSAR_PUBLIC KeyValueEncodingType enumEncodingType(std::string encodingTypeStr) {
+PULSAR_PUBLIC KeyValueEncodingType enumEncodingType(const std::string &encodingTypeStr) {
if (encodingTypeStr == "INLINE") {
return KeyValueEncodingType::INLINE;
} else if (encodingTypeStr == "SEPARATED") {
@@ -104,7 +104,7 @@ PULSAR_PUBLIC const char *strSchemaType(SchemaType schemaType) {
return "UnknownSchemaType";
}
-PULSAR_PUBLIC SchemaType enumSchemaType(std::string schemaTypeStr) {
+PULSAR_PUBLIC SchemaType enumSchemaType(const std::string &schemaTypeStr) {
if (schemaTypeStr == "NONE") {
return NONE;
} else if (schemaTypeStr == "STRING") {
@@ -181,8 +181,8 @@ SchemaInfo::SchemaInfo(const SchemaInfo &keySchema, const SchemaInfo &valueSchem
properties.emplace(VALUE_SCHEMA_PROPS, writeJson(valueSchema.getProperties()));
properties.emplace(KV_ENCODING_TYPE, strEncodingType(keyValueEncodingType));
- std::string keySchemaStr = keySchema.getSchema();
- std::string valueSchemaStr = valueSchema.getSchema();
+ const auto &keySchemaStr = keySchema.getSchema();
+ const auto &valueSchemaStr = valueSchema.getSchema();
impl_ = std::make_shared<SchemaInfoImpl>(KEY_VALUE, "KeyValue",
mergeKeyValueSchema(keySchemaStr, valueSchemaStr), properties);
}
diff --git a/lib/SharedBuffer.h b/lib/SharedBuffer.h
index 26fc59e..033802f 100644
--- a/lib/SharedBuffer.h
+++ b/lib/SharedBuffer.h
@@ -46,8 +46,8 @@ class SharedBuffer {
SharedBuffer& operator=(const SharedBuffer&) = default;
// Move constructor.
- SharedBuffer(SharedBuffer&& right) { *this = std::move(right); }
- SharedBuffer& operator=(SharedBuffer&& right) {
+ SharedBuffer(SharedBuffer&& right) noexcept { *this = std::move(right); }
+ SharedBuffer& operator=(SharedBuffer&& right) noexcept {
this->data_ = std::move(right.data_);
this->ptr_ = right.ptr_;
diff --git a/lib/TableView.cc b/lib/TableView.cc
index 2ff6e62..fc63231 100644
--- a/lib/TableView.cc
+++ b/lib/TableView.cc
@@ -25,7 +25,7 @@ namespace pulsar {
TableView::TableView() {}
-TableView::TableView(TableViewImplPtr impl) : impl_(impl) {}
+TableView::TableView(TableViewImplPtr impl) : impl_(std::move(impl)) {}
bool TableView::retrieveValue(const std::string& key, std::string& value) {
if (impl_) {
@@ -62,19 +62,19 @@ std::size_t TableView::size() const {
return 0;
}
-void TableView::forEach(TableViewAction action) {
+void TableView::forEach(const TableViewAction& action) {
if (impl_) {
impl_->forEach(action);
}
}
-void TableView::forEachAndListen(TableViewAction action) {
+void TableView::forEachAndListen(const TableViewAction& action) {
if (impl_) {
impl_->forEachAndListen(action);
}
}
-void TableView::closeAsync(ResultCallback callback) {
+void TableView::closeAsync(const ResultCallback& callback) {
if (!impl_) {
callback(ResultConsumerNotInitialized);
return;
diff --git a/lib/TableViewImpl.cc b/lib/TableViewImpl.cc
index e434e60..e283a6f 100644
--- a/lib/TableViewImpl.cc
+++ b/lib/TableViewImpl.cc
@@ -28,7 +28,7 @@ namespace pulsar {
DECLARE_LOG_OBJECT()
-TableViewImpl::TableViewImpl(ClientImplPtr client, const std::string& topic,
+TableViewImpl::TableViewImpl(const ClientImplPtr& client, const std::string& topic,
const TableViewConfiguration& conf)
: client_(client), topic_(topic), conf_(conf) {}
@@ -40,7 +40,7 @@ Future<Result, TableViewImplPtr> TableViewImpl::start() {
readerConfiguration.setInternalSubscriptionName(conf_.subscriptionName);
TableViewImplPtr self = shared_from_this();
- ReaderCallback readerCallback = [self, promise](Result res, Reader reader) {
+ ReaderCallback readerCallback = [self, promise](Result res, const Reader& reader) {
if (res == ResultOk) {
self->reader_ = reader.impl_;
self->readAllExistingMessages(promise, TimeUtils::currentTimeMillis(), 0);
@@ -76,15 +76,15 @@ std::unordered_map<std::string, std::string> TableViewImpl::snapshot() { return
std::size_t TableViewImpl::size() const { return data_.size(); }
-void TableViewImpl::forEach(TableViewAction action) { data_.forEach(action); }
+void TableViewImpl::forEach(const TableViewAction& action) { data_.forEach(action); }
-void TableViewImpl::forEachAndListen(TableViewAction action) {
+void TableViewImpl::forEachAndListen(const TableViewAction& action) {
data_.forEach(action);
Lock lock(listenersMutex_);
listeners_.emplace_back(action);
}
-void TableViewImpl::closeAsync(ResultCallback callback) {
+void TableViewImpl::closeAsync(const ResultCallback& callback) {
if (reader_) {
reader_->closeAsync([callback, this](Result result) {
reader_.reset();
@@ -118,7 +118,7 @@ void TableViewImpl::handleMessage(const Message& msg) {
}
}
-void TableViewImpl::readAllExistingMessages(Promise<Result, TableViewImplPtr> promise, long startTime,
+void TableViewImpl::readAllExistingMessages(const Promise<Result, TableViewImplPtr>& promise, long startTime,
long messagesRead) {
std::weak_ptr<TableViewImpl> weakSelf{shared_from_this()};
reader_->hasMessageAvailableAsync(
diff --git a/lib/TableViewImpl.h b/lib/TableViewImpl.h
index 3ca3f75..eda5247 100644
--- a/lib/TableViewImpl.h
+++ b/lib/TableViewImpl.h
@@ -41,7 +41,7 @@ typedef std::shared_ptr<ReaderImpl> ReaderImplPtr;
class TableViewImpl : public std::enable_shared_from_this<TableViewImpl> {
public:
- TableViewImpl(ClientImplPtr client, const std::string& topic, const TableViewConfiguration& conf);
+ TableViewImpl(const ClientImplPtr& client, const std::string& topic, const TableViewConfiguration& conf);
~TableViewImpl(){};
@@ -57,11 +57,11 @@ class TableViewImpl : public std::enable_shared_from_this<TableViewImpl> {
std::size_t size() const;
- void forEach(TableViewAction action);
+ void forEach(const TableViewAction& action);
- void forEachAndListen(TableViewAction action);
+ void forEachAndListen(const TableViewAction& action);
- void closeAsync(ResultCallback callback);
+ void closeAsync(const ResultCallback& callback);
private:
using MutexType = std::mutex;
@@ -77,10 +77,10 @@ class TableViewImpl : public std::enable_shared_from_this<TableViewImpl> {
SynchronizedHashMap<std::string, std::string> data_;
void handleMessage(const Message& msg);
- void readAllExistingMessages(Promise<Result, TableViewImplPtr> promise, long startTime,
+ void readAllExistingMessages(const Promise<Result, TableViewImplPtr>& promise, long startTime,
long messagesRead);
void readTailMessage();
};
} // namespace pulsar
-#endif // PULSAR_CPP_TABLEVIEW_IMPL_H
\ No newline at end of file
+#endif // PULSAR_CPP_TABLEVIEW_IMPL_H
diff --git a/lib/UnAckedMessageTrackerEnabled.cc b/lib/UnAckedMessageTrackerEnabled.cc
index e371af9..b61c4c9 100644
--- a/lib/UnAckedMessageTrackerEnabled.cc
+++ b/lib/UnAckedMessageTrackerEnabled.cc
@@ -32,7 +32,11 @@ namespace pulsar {
void UnAckedMessageTrackerEnabled::timeoutHandler() {
timeoutHandlerHelper();
- ExecutorServicePtr executorService = client_->getIOExecutorProvider()->get();
+ auto client = client_.lock();
+ if (client == nullptr) {
+ return;
+ }
+ ExecutorServicePtr executorService = client->getIOExecutorProvider()->get();
timer_ = executorService->createDeadlineTimer();
timer_->expires_from_now(std::chrono::milliseconds(tickDurationInMs_));
std::weak_ptr<UnAckedMessageTrackerEnabled> weakSelf{shared_from_this()};
@@ -73,12 +77,12 @@ void UnAckedMessageTrackerEnabled::timeoutHandlerHelper() {
}
}
-UnAckedMessageTrackerEnabled::UnAckedMessageTrackerEnabled(long timeoutMs, const ClientImplPtr client,
+UnAckedMessageTrackerEnabled::UnAckedMessageTrackerEnabled(long timeoutMs, const ClientImplPtr& client,
ConsumerImplBase& consumer)
: UnAckedMessageTrackerEnabled(timeoutMs, timeoutMs, client, consumer) {}
UnAckedMessageTrackerEnabled::UnAckedMessageTrackerEnabled(long timeoutMs, long tickDurationInMs,
- const ClientImplPtr client,
+ const ClientImplPtr& client,
ConsumerImplBase& consumer)
: consumerReference_(consumer),
client_(client),
diff --git a/lib/UnAckedMessageTrackerEnabled.h b/lib/UnAckedMessageTrackerEnabled.h
index c5479a7..60625ab 100644
--- a/lib/UnAckedMessageTrackerEnabled.h
+++ b/lib/UnAckedMessageTrackerEnabled.h
@@ -33,12 +33,13 @@ namespace pulsar {
class ClientImpl;
class ConsumerImplBase;
using ClientImplPtr = std::shared_ptr<ClientImpl>;
+using ClientImplWeakPtr = std::weak_ptr<ClientImpl>;
class UnAckedMessageTrackerEnabled : public std::enable_shared_from_this<UnAckedMessageTrackerEnabled>,
public UnAckedMessageTrackerInterface {
public:
- UnAckedMessageTrackerEnabled(long timeoutMs, ClientImplPtr, ConsumerImplBase&);
- UnAckedMessageTrackerEnabled(long timeoutMs, long tickDuration, ClientImplPtr, ConsumerImplBase&);
+ UnAckedMessageTrackerEnabled(long timeoutMs, const ClientImplPtr&, ConsumerImplBase&);
+ UnAckedMessageTrackerEnabled(long timeoutMs, long tickDuration, const ClientImplPtr&, ConsumerImplBase&);
void start() override;
void stop() override;
bool add(const MessageId& msgId) override;
@@ -58,7 +59,7 @@ class UnAckedMessageTrackerEnabled : public std::enable_shared_from_this<UnAcked
std::deque<std::set<MessageId>> timePartitions;
std::recursive_mutex lock_;
ConsumerImplBase& consumerReference_;
- ClientImplPtr client_;
+ ClientImplWeakPtr client_;
DeadlineTimerPtr timer_; // DO NOT place this before client_!
long timeoutMs_;
long tickDurationInMs_;
diff --git a/lib/Utils.h b/lib/Utils.h
index d6303a3..5aab715 100644
--- a/lib/Utils.h
+++ b/lib/Utils.h
@@ -31,7 +31,7 @@ namespace pulsar {
struct WaitForCallback {
Promise<bool, Result> m_promise;
- WaitForCallback(Promise<bool, Result> promise) : m_promise(promise) {}
+ WaitForCallback(Promise<bool, Result> promise) : m_promise(std::move(promise)) {}
void operator()(Result result) { m_promise.setValue(result); }
};
diff --git a/lib/auth/AuthOauth2.cc b/lib/auth/AuthOauth2.cc
index c074557..9573496 100644
--- a/lib/auth/AuthOauth2.cc
+++ b/lib/auth/AuthOauth2.cc
@@ -87,7 +87,7 @@ CachedToken::~CachedToken() {}
// Oauth2CachedToken
-Oauth2CachedToken::Oauth2CachedToken(Oauth2TokenResultPtr token) {
+Oauth2CachedToken::Oauth2CachedToken(const Oauth2TokenResultPtr& token) {
latest_ = token;
int64_t expiredIn = token->getExpiresIn();
@@ -124,7 +124,7 @@ KeyFile KeyFile::fromParamMap(ParamMap& params) {
if (endPos == std::string::npos) {
return "";
}
- const auto prefix = url.substr(startPos, endPos - startPos);
+ auto prefix = url.substr(startPos, endPos - startPos);
startPos = endPos + 1;
return prefix;
};
diff --git a/lib/auth/AuthOauth2.h b/lib/auth/AuthOauth2.h
index e65d0e2..035ad08 100644
--- a/lib/auth/AuthOauth2.h
+++ b/lib/auth/AuthOauth2.h
@@ -79,7 +79,7 @@ class Oauth2CachedToken : public CachedToken {
public:
using Clock = std::chrono::high_resolution_clock;
- Oauth2CachedToken(Oauth2TokenResultPtr token);
+ Oauth2CachedToken(const Oauth2TokenResultPtr& token);
~Oauth2CachedToken();
bool isExpired();
AuthenticationDataPtr getAuthData();
diff --git a/lib/auth/AuthToken.cc b/lib/auth/AuthToken.cc
index 367a26d..1bdfc1a 100644
--- a/lib/auth/AuthToken.cc
+++ b/lib/auth/AuthToken.cc
@@ -90,8 +90,7 @@ AuthenticationPtr AuthToken::create(const std::string &authParamsString) {
std::string envVarName = authParamsString.substr(strlen("env:"));
params["env"] = envVarName;
} else {
- std::string token = authParamsString;
- params["token"] = token;
+ params["token"] = authParamsString;
}
return create(params);
diff --git a/lib/stats/ConsumerStatsImpl.cc b/lib/stats/ConsumerStatsImpl.cc
index 0eefabd..74c7117 100644
--- a/lib/stats/ConsumerStatsImpl.cc
+++ b/lib/stats/ConsumerStatsImpl.cc
@@ -19,9 +19,6 @@
#include "ConsumerStatsImpl.h"
-#include <functional>
-
-#include "lib/ExecutorService.h"
#include "lib/LogUtils.h"
#include "lib/Utils.h"
@@ -30,11 +27,9 @@ DECLARE_LOG_OBJECT();
using Lock = std::unique_lock<std::mutex>;
-ConsumerStatsImpl::ConsumerStatsImpl(std::string consumerStr, ExecutorServicePtr executor,
+ConsumerStatsImpl::ConsumerStatsImpl(const std::string& consumerStr, DeadlineTimerPtr timer,
unsigned int statsIntervalInSeconds)
- : consumerStr_(consumerStr),
- timer_(executor->createDeadlineTimer()),
- statsIntervalInSeconds_(statsIntervalInSeconds) {}
+ : consumerStr_(consumerStr), timer_(std::move(timer)), statsIntervalInSeconds_(statsIntervalInSeconds) {}
ConsumerStatsImpl::ConsumerStatsImpl(const ConsumerStatsImpl& stats)
: consumerStr_(stats.consumerStr_),
diff --git a/lib/stats/ConsumerStatsImpl.h b/lib/stats/ConsumerStatsImpl.h
index 3333ea8..0ce0ef1 100644
--- a/lib/stats/ConsumerStatsImpl.h
+++ b/lib/stats/ConsumerStatsImpl.h
@@ -27,12 +27,8 @@
#include "ConsumerStatsBase.h"
#include "lib/AsioTimer.h"
-#include "lib/ExecutorService.h"
namespace pulsar {
-class ExecutorService;
-using ExecutorServicePtr = std::shared_ptr<ExecutorService>;
-
class ConsumerStatsImpl : public std::enable_shared_from_this<ConsumerStatsImpl>, public ConsumerStatsBase {
private:
std::string consumerStr_;
@@ -55,7 +51,7 @@ class ConsumerStatsImpl : public std::enable_shared_from_this<ConsumerStatsImpl>
friend class PulsarFriend;
public:
- ConsumerStatsImpl(std::string, ExecutorServicePtr, unsigned int);
+ ConsumerStatsImpl(const std::string&, DeadlineTimerPtr, unsigned int);
ConsumerStatsImpl(const ConsumerStatsImpl& stats);
void flushAndReset(const ASIO_ERROR&);
void start() override;
diff --git a/lib/stats/ProducerStatsImpl.cc b/lib/stats/ProducerStatsImpl.cc
index 15e9e67..8aae280 100644
--- a/lib/stats/ProducerStatsImpl.cc
+++ b/lib/stats/ProducerStatsImpl.cc
@@ -45,9 +45,9 @@ std::string ProducerStatsImpl::latencyToString(const LatencyAccumulator& obj) {
return os.str();
}
-ProducerStatsImpl::ProducerStatsImpl(std::string producerStr, ExecutorServicePtr executor,
+ProducerStatsImpl::ProducerStatsImpl(std::string producerStr, const ExecutorServicePtr& executor,
unsigned int statsIntervalInSeconds)
- : producerStr_(producerStr),
+ : producerStr_(std::move(producerStr)),
latencyAccumulator_(boost::accumulators::tag::extended_p_square::probabilities = probs),
totalLatencyAccumulator_(boost::accumulators::tag::extended_p_square::probabilities = probs),
timer_(executor->createDeadlineTimer()),
diff --git a/lib/stats/ProducerStatsImpl.h b/lib/stats/ProducerStatsImpl.h
index 5d445c6..495a89c 100644
--- a/lib/stats/ProducerStatsImpl.h
+++ b/lib/stats/ProducerStatsImpl.h
@@ -75,7 +75,7 @@ class ProducerStatsImpl : public std::enable_shared_from_this<ProducerStatsImpl>
void scheduleTimer();
public:
- ProducerStatsImpl(std::string, ExecutorServicePtr, unsigned int);
+ ProducerStatsImpl(std::string, const ExecutorServicePtr&, unsigned int);
ProducerStatsImpl(const ProducerStatsImpl& stats);
diff --git a/tests/AcknowledgeTest.cc b/tests/AcknowledgeTest.cc
index 9ca1310..0e2183e 100644
--- a/tests/AcknowledgeTest.cc
+++ b/tests/AcknowledgeTest.cc
@@ -147,7 +147,7 @@ TEST_P(AcknowledgeTest, testAckMsgListWithMultiConsumer) {
// assert stats
unsigned long totalAck = 0;
auto consumerStatsList = PulsarFriend::getConsumerStatsPtrList(consumer);
- for (auto consumerStats : consumerStatsList) {
+ for (auto&& consumerStats : consumerStatsList) {
auto ackMap = consumerStats->getAckedMsgMap();
totalAck += ackMap[std::make_pair(ResultOk, CommandAck_AckType_Individual)];
}
diff --git a/tests/AuthPluginTest.cc b/tests/AuthPluginTest.cc
index 24549d7..e48dbfb 100644
--- a/tests/AuthPluginTest.cc
+++ b/tests/AuthPluginTest.cc
@@ -20,7 +20,9 @@
#include <pulsar/Authentication.h>
#include <pulsar/Client.h>
+#include <array>
#include <boost/algorithm/string.hpp>
+#include <sstream>
#ifdef USE_ASIO
#include <asio.hpp>
#else
@@ -305,39 +307,106 @@ TEST(AuthPluginTest, testTlsDetectClientCertSignedByICA) {
ASSERT_EQ(ResultOk, res);
}
+// There is a bug that clang-tidy could report memory leak for boost::algorithm::split, see
+// https://github.com/boostorg/algorithm/issues/63.
+static std::vector<std::string> split(const std::string& s, char separator) {
+ std::vector<std::string> tokens;
+ size_t startPos = 0;
+ while (startPos < s.size()) {
+ auto pos = s.find(separator, startPos);
+ if (pos == std::string::npos) {
+ tokens.emplace_back(s.substr(startPos));
+ break;
+ }
+ tokens.emplace_back(s.substr(startPos, pos - startPos));
+ startPos = pos + 1;
+ }
+ return tokens;
+}
+
namespace testAthenz {
std::string principalToken;
+
+// ASIO::ip::tcp::iostream could call a virtual function during destruction, so the clang-tidy will fail by
+// clang-analyzer-optin.cplusplus.VirtualCall. Here we write a simple stream to read lines from socket.
+class SocketStream {
+ public:
+ SocketStream(ASIO::ip::tcp::socket& socket) : socket_(socket) {}
+
+ bool getline(std::string& line) {
+ auto pos = buffer_.find('\n', bufferPos_);
+ if (pos != std::string::npos) {
+ line = buffer_.substr(bufferPos_, pos - bufferPos_);
+ bufferPos_ = pos + 1;
+ return true;
+ }
+
+ std::array<char, 1024> buffer;
+ ASIO_ERROR error;
+ auto length = socket_.read_some(ASIO::buffer(buffer.data(), buffer.size()), error);
+ if (error == ASIO::error::eof) {
+ return false;
+ } else if (error) {
+ LOG_ERROR("Failed to read from socket: " << error.message());
+ return false;
+ }
+ buffer_.append(buffer.data(), length);
+
+ pos = buffer_.find('\n', bufferPos_);
+ if (pos != std::string::npos) {
+ line = buffer_.substr(bufferPos_, pos - bufferPos_);
+ bufferPos_ = pos + 1;
+ } else {
+ line = "";
+ }
+ return true;
+ }
+
+ private:
+ ASIO::ip::tcp::socket& socket_;
+ std::string buffer_;
+ size_t bufferPos_{0};
+};
+
void mockZTS(Latch& latch, int port) {
LOG_INFO("-- MockZTS started");
ASIO::io_service io;
- ASIO::ip::tcp::iostream stream;
ASIO::ip::tcp::acceptor acceptor(io, ASIO::ip::tcp::endpoint(ASIO::ip::tcp::v4(), port));
LOG_INFO("-- MockZTS waiting for connnection");
latch.countdown();
- acceptor.accept(*stream.rdbuf());
+ ASIO::ip::tcp::socket socket(io);
+ acceptor.accept(socket);
LOG_INFO("-- MockZTS got connection");
std::string headerLine;
- while (getline(stream, headerLine)) {
- std::vector<std::string> kv;
- boost::algorithm::split(kv, headerLine, boost::is_any_of(" "));
+ SocketStream stream(socket);
+ while (stream.getline(headerLine)) {
+ if (headerLine.empty()) {
+ continue;
+ }
+ auto kv = split(headerLine, ' ');
if (kv[0] == "Athenz-Principal-Auth:") {
principalToken = kv[1];
}
- if (headerLine == "\r" || headerLine == "\n" || headerLine == "\r\n") {
+ if (headerLine == "\r") {
+ std::ostringstream stream;
std::string mockToken = "{\"token\":\"mockToken\",\"expiryTime\":4133980800}";
- stream << "HTTP/1.1 200 OK" << std::endl;
- stream << "Host: localhost" << std::endl;
- stream << "Content-Type: application/json" << std::endl;
- stream << "Content-Length: " << mockToken.size() << std::endl;
- stream << std::endl;
- stream << mockToken << std::endl;
+ stream << "HTTP/1.1 200 OK" << '\n';
+ stream << "Host: localhost" << '\n';
+ stream << "Content-Type: application/json" << '\n';
+ stream << "Content-Length: " << mockToken.size() << '\n';
+ stream << '\n';
+ stream << mockToken << '\n';
+ auto response = stream.str();
+ socket.send(ASIO::const_buffer(response.c_str(), response.length()));
break;
}
}
+ socket.close();
+ acceptor.close();
LOG_INFO("-- MockZTS exiting");
}
} // namespace testAthenz
@@ -365,11 +434,9 @@ TEST(AuthPluginTest, testAthenz) {
ASSERT_EQ(data->getHttpHeaders(), "Athenz-Role-Auth: mockToken");
ASSERT_EQ(data->getCommandData(), "mockToken");
zts.join();
- std::vector<std::string> kvs;
- boost::algorithm::split(kvs, testAthenz::principalToken, boost::is_any_of(";"));
+ auto kvs = split(testAthenz::principalToken, ';');
for (std::vector<std::string>::iterator itr = kvs.begin(); itr != kvs.end(); itr++) {
- std::vector<std::string> kv;
- boost::algorithm::split(kv, *itr, boost::is_any_of("="));
+ auto kv = split(*itr, '=');
if (kv[0] == "d") {
ASSERT_EQ(kv[1], "pulsar.test.tenant");
} else if (kv[0] == "n") {
@@ -440,11 +507,9 @@ TEST(AuthPluginTest, testAuthFactoryAthenz) {
zts.join();
LOG_INFO("Done zts.join()");
- std::vector<std::string> kvs;
- boost::algorithm::split(kvs, testAthenz::principalToken, boost::is_any_of(";"));
+ auto kvs = split(testAthenz::principalToken, ';');
for (std::vector<std::string>::iterator itr = kvs.begin(); itr != kvs.end(); itr++) {
- std::vector<std::string> kv;
- boost::algorithm::split(kv, *itr, boost::is_any_of("="));
+ auto kv = split(*itr, '=');
if (kv[0] == "d") {
ASSERT_EQ(kv[1], "pulsar.test2.tenant");
} else if (kv[0] == "n") {
diff --git a/tests/BasicEndToEndTest.cc b/tests/BasicEndToEndTest.cc
index 994bedd..e3c7039 100644
--- a/tests/BasicEndToEndTest.cc
+++ b/tests/BasicEndToEndTest.cc
@@ -78,14 +78,14 @@ static void messageListenerFunction(Consumer consumer, const Message &msg) {
consumer.acknowledge(msg);
}
-static void messageListenerFunctionWithoutAck(Consumer consumer, const Message &msg, Latch &latch,
+static void messageListenerFunctionWithoutAck(const Consumer &consumer, const Message &msg, Latch &latch,
const std::string &content) {
globalCount++;
ASSERT_EQ(content, msg.getDataAsString());
latch.countdown();
}
-static void sendCallBack(Result r, const MessageId &msgId, std::string prefix, int *count) {
+static void sendCallBack(Result r, const MessageId &msgId, const std::string &prefix, int *count) {
static std::mutex sendMutex_;
sendMutex_.lock();
ASSERT_EQ(r, ResultOk);
@@ -111,8 +111,8 @@ static void receiveCallBack(Result r, const Message &msg, std::string &messageCo
receiveMutex_.unlock();
}
-static void sendCallBackWithDelay(Result r, const MessageId &msgId, std::string prefix, double percentage,
- uint64_t delayInMicros, int *count) {
+static void sendCallBackWithDelay(Result r, const MessageId &msgId, const std::string &prefix,
+ double percentage, uint64_t delayInMicros, int *count) {
if ((rand() % 100) <= percentage) {
std::this_thread::sleep_for(std::chrono::microseconds(delayInMicros));
}
@@ -194,7 +194,7 @@ TEST(BasicEndToEndTest, testBatchMessages) {
ASSERT_EQ(i, numOfMessages);
}
-void resendMessage(Result r, const MessageId msgId, Producer producer) {
+void resendMessage(Result r, const MessageId &msgId, Producer &producer) {
std::unique_lock<std::mutex> lock(mutex_);
if (r != ResultOk) {
LOG_DEBUG("globalResendMessageCount" << globalResendMessageCount);
@@ -425,6 +425,7 @@ TEST(BasicEndToEndTest, testProduceAndConsumeAfterClientClose) {
Consumer consumer;
result = client.subscribe(topicName, "my-sub-name", consumer);
+ ASSERT_EQ(ResultOk, result);
// Clean dangling subscription
consumer.unsubscribe();
@@ -494,6 +495,7 @@ TEST(BasicEndToEndTest, testSubscribeCloseUnsubscribeSherpaScenario) {
Consumer consumer1;
result = client.subscribe(topicName, subName, consumer1);
+ ASSERT_EQ(ResultOk, result);
result = consumer1.unsubscribe();
ASSERT_EQ(ResultOk, result);
}
@@ -506,7 +508,7 @@ TEST(BasicEndToEndTest, testInvalidUrlPassed) {
EXPECT_THROW({ Client{"Dream of the day when this will be a valid URL"}; }, std::invalid_argument);
}
-void testPartitionedProducerConsumer(bool lazyStartPartitionedProducers, std::string topicName) {
+void testPartitionedProducerConsumer(bool lazyStartPartitionedProducers, const std::string &topicName) {
Client client(lookupUrl);
// call admin api to make it partitioned
@@ -709,6 +711,7 @@ TEST(BasicEndToEndTest, testSinglePartitionRoutingPolicy) {
ProducerConfiguration producerConfiguration;
producerConfiguration.setPartitionsRoutingMode(ProducerConfiguration::UseSinglePartition);
Result result = client.createProducer(topicName, producerConfiguration, producer);
+ ASSERT_EQ(ResultOk, result);
Consumer consumer;
result = client.subscribe(topicName, "subscription-A", consumer);
@@ -884,6 +887,7 @@ TEST(BasicEndToEndTest, testMessageListener) {
ProducerConfiguration producerConfiguration;
producerConfiguration.setPartitionsRoutingMode(ProducerConfiguration::UseSinglePartition);
Result result = client.createProducer(topicName, producerConfiguration, producer);
+ ASSERT_EQ(ResultOk, result);
// Initializing global Count
globalCount = 0;
@@ -927,6 +931,7 @@ TEST(BasicEndToEndTest, testMessageListenerPause) {
ProducerConfiguration producerConfiguration;
producerConfiguration.setPartitionsRoutingMode(ProducerConfiguration::UseSinglePartition);
Result result = client.createProducer(topicName, producerConfiguration, producer);
+ ASSERT_EQ(ResultOk, result);
// Initializing global Count
globalCount = 0;
@@ -937,6 +942,7 @@ TEST(BasicEndToEndTest, testMessageListenerPause) {
Consumer consumer;
// Removing dangling subscription from previous test failures
result = client.subscribe(topicName, "subscription-name", consumerConfig, consumer);
+ ASSERT_EQ(ResultOk, result);
consumer.unsubscribe();
result = client.subscribe(topicName, "subscription-name", consumerConfig, consumer);
@@ -983,6 +989,7 @@ void testStartPaused(bool isPartitioned) {
Producer producer;
Result result = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultOk, result);
// Initializing global Count
globalCount = 0;
@@ -994,6 +1001,7 @@ void testStartPaused(bool isPartitioned) {
Consumer consumer;
// Removing dangling subscription from previous test failures
result = client.subscribe(topicName, subName, consumerConfig, consumer);
+ ASSERT_EQ(ResultOk, result);
consumer.unsubscribe();
result = client.subscribe(topicName, subName, consumerConfig, consumer);
@@ -1164,7 +1172,7 @@ TEST(BasicEndToEndTest, testStatsLatencies) {
ASSERT_EQ(expectedMessageContent, receivedMsg.getDataAsString());
ASSERT_EQ(ResultOk, consumer.acknowledge(receivedMsg));
- auto msgId = receivedMsg.getMessageId();
+ const auto &msgId = receivedMsg.getMessageId();
if (msgId.batchIndex() < 0) {
numAcks++;
} else if (msgId.batchIndex() + 1 == msgId.batchSize()) {
@@ -1556,6 +1564,7 @@ TEST(BasicEndToEndTest, testEncryptionFailure) {
client.subscribeAsync(topicName, subName, consConfig, WaitForCallbackValue<Consumer>(consumerPromise3));
consumerFuture = consumerPromise3.getFuture();
result = consumerFuture.get(consumer);
+ ASSERT_EQ(ResultOk, result);
for (; msgNum < totalMsgs - 1; msgNum++) {
ASSERT_EQ(ResultOk, consumer.receive(msgReceived, 1000));
@@ -1575,6 +1584,7 @@ TEST(BasicEndToEndTest, testEncryptionFailure) {
client.subscribeAsync(topicName, subName, consConfig2, WaitForCallbackValue<Consumer>(consumerPromise4));
consumerFuture = consumerPromise4.getFuture();
result = consumerFuture.get(consumer);
+ ASSERT_EQ(ResultOk, result);
// Since messag is discarded, no message will be received.
ASSERT_EQ(ResultTimeout, consumer.receive(msgReceived, 5000));
@@ -1793,7 +1803,7 @@ TEST(BasicEndToEndTest, testUnAckedMessageTimeout) {
static long messagesReceived = 0;
-static void unackMessageListenerFunction(Consumer consumer, const Message &msg) { messagesReceived++; }
+static void unackMessageListenerFunction(const Consumer &consumer, const Message &msg) { messagesReceived++; }
TEST(BasicEndToEndTest, testPartitionTopicUnAckedMessageTimeout) {
Client client(lookupUrl);
@@ -2831,6 +2841,7 @@ void testFlushInPartitionedProducer(bool lazyStartPartitionedProducers) {
partitionedConsumerId << consumerId << i;
subscribeResult = client.subscribe(partitionedTopicName.str(), partitionedConsumerId.str(),
consConfig, consumer[i]);
+ ASSERT_EQ(ResultOk, subscribeResult);
consumer[i].unsubscribe();
subscribeResult = client.subscribe(partitionedTopicName.str(), partitionedConsumerId.str(),
consConfig, consumer[i]);
@@ -3240,7 +3251,7 @@ TEST(BasicEndToEndTest, testNegativeAcksWithPartitions) {
static long regexTestMessagesReceived = 0;
-static void regexMessageListenerFunction(Consumer consumer, const Message &msg) {
+static void regexMessageListenerFunction(const Consumer &consumer, const Message &msg) {
regexTestMessagesReceived++;
}
@@ -3417,7 +3428,7 @@ TEST(BasicEndToEndTest, testDelayedMessages) {
ASSERT_EQ("msg-2", msgReceived.getDataAsString());
auto result1 = client.close();
- std::cout << "closed with " << result1 << std::endl;
+ std::cout << "closed with " << result1 << '\n';
ASSERT_EQ(ResultOk, result1);
}
@@ -3919,7 +3930,7 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerEnabledCumulativeAck) {
false, ackGroupingTimeMs, ackGroupingMaxSize, clientImplPtr->getIOExecutorProvider()->get());
tracker1->start();
tracker1->addAcknowledgeCumulative(recvMsgId[numMsg - 1], nullptr);
- tracker1->close();
+ tracker1.reset();
consumer.close();
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
@@ -3929,7 +3940,7 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerEnabledCumulativeAck) {
class UnAckedMessageTrackerEnabledMock : public UnAckedMessageTrackerEnabled {
public:
- UnAckedMessageTrackerEnabledMock(long timeoutMs, const ClientImplPtr client, ConsumerImplBase &consumer)
+ UnAckedMessageTrackerEnabledMock(long timeoutMs, const ClientImplPtr &client, ConsumerImplBase &consumer)
: UnAckedMessageTrackerEnabled(timeoutMs, timeoutMs, client, consumer) {}
const long getUnAckedMessagesTimeoutMs() { return this->timeoutMs_; }
const long getTickDurationInMs() { return this->tickDurationInMs_; }
@@ -4202,8 +4213,8 @@ void testBatchReceive(bool multiConsumer) {
ASSERT_EQ(ResultOk, producer.flush());
for (int i = 0; i < numOfMessages / batchReceiveMaxNumMessages; i++) {
Latch latch(1);
- BatchReceiveCallback batchReceiveCallback = [&latch, batchReceiveMaxNumMessages](Result result,
- Messages messages) {
+ BatchReceiveCallback batchReceiveCallback = [&latch, batchReceiveMaxNumMessages](
+ Result result, const Messages &messages) {
ASSERT_EQ(result, ResultOk);
ASSERT_EQ(messages.size(), batchReceiveMaxNumMessages);
latch.countdown();
@@ -4265,7 +4276,8 @@ void testBatchReceiveTimeout(bool multiConsumer) {
}
Latch latch(1);
- BatchReceiveCallback batchReceiveCallback = [&latch, numOfMessages](Result result, Messages messages) {
+ BatchReceiveCallback batchReceiveCallback = [&latch, numOfMessages](Result result,
+ const Messages &messages) {
ASSERT_EQ(result, ResultOk);
ASSERT_EQ(messages.size(), numOfMessages);
latch.countdown();
@@ -4310,7 +4322,7 @@ void testBatchReceiveClose(bool multiConsumer) {
ASSERT_EQ(ResultOk, result);
Latch latch(1);
- BatchReceiveCallback batchReceiveCallback = [&latch](Result result, Messages messages) {
+ BatchReceiveCallback batchReceiveCallback = [&latch](Result result, const Messages &messages) {
ASSERT_EQ(result, ResultAlreadyClosed);
latch.countdown();
};
diff --git a/tests/BatchMessageTest.cc b/tests/BatchMessageTest.cc
index 99e61f8..0b61de1 100644
--- a/tests/BatchMessageTest.cc
+++ b/tests/BatchMessageTest.cc
@@ -307,7 +307,7 @@ TEST(BatchMessageTest, testSmallReceiverQueueSize) {
ProducerStatsImplPtr producerStatsImplPtr = PulsarFriend::getProducerStatsPtr(producer);
// Send Asynchronously
std::atomic_int numOfMessagesProduced{0};
- std::string prefix = testName;
+ const auto& prefix = testName;
for (int i = 0; i < numOfMessages; i++) {
std::string messageContent = prefix + std::to_string(i);
Message msg =
@@ -400,7 +400,7 @@ TEST(BatchMessageTest, testIndividualAck) {
// Send Asynchronously
std::atomic_int numOfMessagesProduced{0};
- std::string prefix = testName;
+ const auto& prefix = testName;
for (int i = 0; i < numOfMessages; i++) {
std::string messageContent = prefix + std::to_string(i);
Message msg =
@@ -552,7 +552,7 @@ TEST(BatchMessageTest, testCumulativeAck) {
// Send Asynchronously
std::atomic_int numOfMessagesProduced{0};
- std::string prefix = testName;
+ const auto& prefix = testName;
for (int i = 0; i < numOfMessages; i++) {
std::string messageContent = prefix + std::to_string(i);
Message msg =
@@ -671,7 +671,7 @@ TEST(BatchMessageTest, testMixedAck) {
// Send Asynchronously
std::atomic_int numOfMessagesProduced{0};
- std::string prefix = testName;
+ const auto& prefix = testName;
for (int i = 0; i < numOfMessages; i++) {
std::string messageContent = prefix + std::to_string(i);
Message msg =
@@ -779,7 +779,7 @@ TEST(BatchMessageTest, testPermits) {
// Send Asynchronously
std::atomic_int numOfMessagesProduced{0};
- std::string prefix = testName;
+ const auto& prefix = testName;
for (int i = 0; i < numOfMessages; i++) {
std::string messageContent = prefix + std::to_string(i);
Message msg =
diff --git a/tests/ClientTest.cc b/tests/ClientTest.cc
index b514286..dd89268 100644
--- a/tests/ClientTest.cc
+++ b/tests/ClientTest.cc
@@ -116,11 +116,12 @@ TEST(ClientTest, testConnectTimeout) {
std::promise<Result> promiseLow;
clientLow.createProducerAsync(
- topic, [&promiseLow](Result result, Producer producer) { promiseLow.set_value(result); });
+ topic, [&promiseLow](Result result, const Producer &producer) { promiseLow.set_value(result); });
std::promise<Result> promiseDefault;
- clientDefault.createProducerAsync(
- topic, [&promiseDefault](Result result, Producer producer) { promiseDefault.set_value(result); });
+ clientDefault.createProducerAsync(topic, [&promiseDefault](Result result, const Producer &producer) {
+ promiseDefault.set_value(result);
+ });
auto futureLow = promiseLow.get_future();
ASSERT_EQ(futureLow.wait_for(std::chrono::milliseconds(1500)), std::future_status::ready);
@@ -334,7 +335,8 @@ class PulsarWrapper {
// When `subscription` is empty, get client versions of the producers.
// Otherwise, get client versions of the consumers under the subscribe.
-static std::vector<std::string> getClientVersions(const std::string &topic, std::string subscription = "") {
+static std::vector<std::string> getClientVersions(const std::string &topic,
+ const std::string &subscription = "") {
boost::property_tree::ptree root;
const auto error = getTopicStats(topic, root);
if (!error.empty()) {
@@ -403,7 +405,7 @@ TEST(ClientTest, testConnectionClose) {
const auto topic = "client-test-connection-close";
for (auto &client : clients) {
- auto testClose = [&client](ClientConnectionWeakPtr weakCnx) {
+ auto testClose = [&client](const ClientConnectionWeakPtr &weakCnx) {
auto cnx = weakCnx.lock();
ASSERT_TRUE(cnx);
diff --git a/tests/ConsumerConfigurationTest.cc b/tests/ConsumerConfigurationTest.cc
index 482f0fc..e44543d 100644
--- a/tests/ConsumerConfigurationTest.cc
+++ b/tests/ConsumerConfigurationTest.cc
@@ -90,7 +90,7 @@ TEST(ConsumerConfigurationTest, testCustomConfig) {
conf.setConsumerType(ConsumerKeyShared);
ASSERT_EQ(conf.getConsumerType(), ConsumerKeyShared);
- conf.setMessageListener([](Consumer consumer, const Message& msg) {});
+ conf.setMessageListener([](const Consumer& consumer, const Message& msg) {});
ASSERT_EQ(conf.hasMessageListener(), true);
conf.setConsumerEventListener(std::make_shared<DummyEventListener>());
diff --git a/tests/ConsumerStatsTest.cc b/tests/ConsumerStatsTest.cc
index 3c1959a..d1ad139 100644
--- a/tests/ConsumerStatsTest.cc
+++ b/tests/ConsumerStatsTest.cc
@@ -24,7 +24,6 @@
#include "ConsumerTest.h"
#include "HttpHelper.h"
-#include "PulsarFriend.h"
#include "lib/Future.h"
#include "lib/Latch.h"
#include "lib/LogUtils.h"
@@ -37,8 +36,8 @@ using namespace pulsar;
static std::string lookupUrl = "pulsar://localhost:6650";
static std::string adminUrl = "http://localhost:8080/";
-void partitionedCallbackFunction(Result result, BrokerConsumerStats brokerConsumerStats, long expectedBacklog,
- Latch& latch, int index, bool accurate) {
+void partitionedCallbackFunction(Result result, const BrokerConsumerStats& brokerConsumerStats,
+ long expectedBacklog, Latch& latch, int index, bool accurate) {
ASSERT_EQ(result, ResultOk);
MultiTopicsBrokerConsumerStatsImpl* statsPtr =
(MultiTopicsBrokerConsumerStatsImpl*)(brokerConsumerStats.getImpl().get());
@@ -51,8 +50,9 @@ void partitionedCallbackFunction(Result result, BrokerConsumerStats brokerConsum
latch.countdown();
}
-void simpleCallbackFunction(Result result, BrokerConsumerStats brokerConsumerStats, Result expectedResult,
- uint64_t expectedBacklog, ConsumerType expectedConsumerType) {
+void simpleCallbackFunction(Result result, const BrokerConsumerStats& brokerConsumerStats,
+ Result expectedResult, uint64_t expectedBacklog,
+ ConsumerType expectedConsumerType) {
LOG_DEBUG(brokerConsumerStats);
ASSERT_EQ(result, expectedResult);
ASSERT_EQ(brokerConsumerStats.getMsgBacklog(), expectedBacklog);
diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc
index 105e5f6..ad4ee81 100644
--- a/tests/ConsumerTest.cc
+++ b/tests/ConsumerTest.cc
@@ -61,7 +61,7 @@ namespace pulsar {
class ConsumerStateEventListener : public ConsumerEventListener {
public:
- ConsumerStateEventListener(std::string name) { name_ = name; }
+ ConsumerStateEventListener(std::string name) { name_ = std::move(name); }
void becameActive(Consumer consumer, int partitionId) override {
LOG_INFO("Received consumer active event, partitionId:" << partitionId << ", name: " << name_);
diff --git a/tests/CustomLoggerTest.cc b/tests/CustomLoggerTest.cc
index 22fbb4e..b57ede6 100644
--- a/tests/CustomLoggerTest.cc
+++ b/tests/CustomLoggerTest.cc
@@ -38,7 +38,7 @@ class MyTestLogger : public Logger {
void log(Level level, int line, const std::string &message) override {
std::stringstream ss;
ss << std::this_thread::get_id() << " " << level << " " << fileName_ << ":" << line << " " << message
- << std::endl;
+ << '\n';
logLines.emplace_back(ss.str());
}
diff --git a/tests/InterceptorsTest.cc b/tests/InterceptorsTest.cc
index db87d8f..becbdf8 100644
--- a/tests/InterceptorsTest.cc
+++ b/tests/InterceptorsTest.cc
@@ -21,6 +21,7 @@
#include <pulsar/ConsumerInterceptor.h>
#include <pulsar/ProducerInterceptor.h>
+#include <cstdint>
#include <random>
#include <utility>
@@ -109,7 +110,7 @@ class ProducerPartitionsChangeInterceptor : public ProducerInterceptor {
Latch latch_;
};
-void createPartitionedTopic(std::string topic) {
+void createPartitionedTopic(const std::string& topic) {
std::string topicOperateUrl = adminUrl + "admin/v2/persistent/public/default/" + topic + "/partitions";
int res = makePutRequest(topicOperateUrl, "2");
@@ -229,7 +230,7 @@ class ConsumerExceptionInterceptor : public ConsumerInterceptor {
Latch latch_;
};
-enum TopicType
+enum TopicType : uint8_t
{
Single,
Partitioned,
@@ -294,7 +295,7 @@ class ConsumerInterceptorsTest : public ::testing::TestWithParam<std::tuple<Topi
consumerConf_.setReceiverQueueSize(std::get<1>(GetParam()));
}
- void createPartitionedTopic(std::string topic) {
+ void createPartitionedTopic(const std::string& topic) {
std::string topicOperateUrl = adminUrl + "admin/v2/persistent/" +
topic.substr(std::string("persistent://").length()) + "/partitions";
diff --git a/tests/LookupServiceTest.cc b/tests/LookupServiceTest.cc
index 924acd4..ff3a7e0 100644
--- a/tests/LookupServiceTest.cc
+++ b/tests/LookupServiceTest.cc
@@ -90,14 +90,14 @@ TEST(LookupServiceTest, basicLookup) {
Future<Result, LookupDataResultPtr> partitionFuture = lookupService.getPartitionMetadataAsync(topicName);
LookupDataResultPtr lookupData;
- Result result = partitionFuture.get(lookupData);
+ partitionFuture.get(lookupData);
ASSERT_TRUE(lookupData != NULL);
ASSERT_EQ(0, lookupData->getPartitions());
const auto topicNamePtr = TopicName::get("topic");
auto future = lookupService.getBroker(*topicNamePtr);
LookupService::LookupResult lookupResult;
- result = future.get(lookupResult);
+ auto result = future.get(lookupResult);
ASSERT_EQ(ResultOk, result);
ASSERT_EQ(url, lookupResult.logicalAddress);
diff --git a/tests/MultiTopicsConsumerTest.cc b/tests/MultiTopicsConsumerTest.cc
index 5aae1eb..0694fa4 100644
--- a/tests/MultiTopicsConsumerTest.cc
+++ b/tests/MultiTopicsConsumerTest.cc
@@ -89,7 +89,7 @@ TEST(MultiTopicsConsumerTest, testSeekToNewerPosition) {
consumer.close();
// Test message listener
- conf.setMessageListener([&messages](Consumer consumer, Message msg) { messages.add(msg); });
+ conf.setMessageListener([&messages](const Consumer& consumer, const Message& msg) { messages.add(msg); });
messages.clear();
messages.setMinNumMsgs(4);
ASSERT_EQ(ResultOk, client.subscribe(topics, "sub-4", conf, consumer));
diff --git a/tests/PartitionsUpdateTest.cc b/tests/PartitionsUpdateTest.cc
index 4d2fa27..609b91f 100644
--- a/tests/PartitionsUpdateTest.cc
+++ b/tests/PartitionsUpdateTest.cc
@@ -50,7 +50,7 @@ class PartitionsSet {
public:
size_t size() const { return names_.size(); }
- Result initProducer(std::string topicName, bool enablePartitionsUpdate,
+ Result initProducer(const std::string& topicName, bool enablePartitionsUpdate,
bool lazyStartPartitionedProducers) {
clientForProducer_.reset(new Client(serviceUrl, newClientConfig(enablePartitionsUpdate)));
const auto producerConfig = ProducerConfiguration()
@@ -59,7 +59,7 @@ class PartitionsSet {
return clientForProducer_->createProducer(topicName, producerConfig, producer_);
}
- Result initConsumer(std::string topicName, bool enablePartitionsUpdate) {
+ Result initConsumer(const std::string& topicName, bool enablePartitionsUpdate) {
clientForConsumer_.reset(new Client(serviceUrl, newClientConfig(enablePartitionsUpdate)));
return clientForConsumer_->subscribe(topicName, "SubscriptionName", consumer_);
}
@@ -120,7 +120,7 @@ TEST(PartitionsUpdateTest, testConfigPartitionsUpdateInterval) {
ASSERT_EQ(static_cast<unsigned int>(-1), clientConfig.getPartitionsUpdateInterval());
}
-void testPartitionsUpdate(bool lazyStartPartitionedProducers, std::string topicNameSuffix) {
+void testPartitionsUpdate(bool lazyStartPartitionedProducers, const std::string& topicNameSuffix) {
std::string topicName = "persistent://" + topicNameSuffix;
std::string topicOperateUrl = adminUrl + "admin/v2/persistent/" + topicNameSuffix + "/partitions";
diff --git a/tests/ProducerTest.cc b/tests/ProducerTest.cc
index d0d9eb4..edb79e4 100644
--- a/tests/ProducerTest.cc
+++ b/tests/ProducerTest.cc
@@ -322,7 +322,7 @@ TEST(ProducerTest, testWaitForExclusiveProducer) {
Latch latch(1);
client.createProducerAsync(topicName, producerConfiguration2,
- [&latch, &producer2](Result res, Producer producer) {
+ [&latch, &producer2](Result res, const Producer& producer) {
ASSERT_EQ(ResultOk, res);
producer2 = producer;
latch.countdown();
@@ -357,7 +357,7 @@ TEST(ProducerTest, testExclusiveWithFencingProducer) {
Latch latch(1);
client.createProducerAsync(topicName, producerConfiguration2,
- [&latch, &producer2](Result res, Producer producer) {
+ [&latch, &producer2](Result res, const Producer& producer) {
// producer2 will be fenced
ASSERT_EQ(ResultProducerFenced, res);
latch.countdown();
@@ -385,7 +385,7 @@ TEST(ProducerTest, testExclusiveWithFencingProducer) {
producerConfiguration2.setAccessMode(ProducerConfiguration::WaitForExclusive);
Latch latch2(1);
client.createProducerAsync(topicName, producerConfiguration2,
- [&latch2, &producer4](Result res, Producer producer) {
+ [&latch2, &producer4](Result res, const Producer& producer) {
// producer4 will be success
ASSERT_EQ(ResultOk, res);
producer4 = producer;
diff --git a/tests/ProtobufNativeSchemaTest.cc b/tests/ProtobufNativeSchemaTest.cc
index 32719bd..cf3c8dc 100644
--- a/tests/ProtobufNativeSchemaTest.cc
+++ b/tests/ProtobufNativeSchemaTest.cc
@@ -131,7 +131,7 @@ TEST(ProtobufNativeSchemaTest, testEndToEnd) {
TEST(ProtobufNativeSchemaTest, testBase64WithPadding) {
const auto schemaInfo = createProtobufNativeSchema(::padding::demo::Person::descriptor());
- const auto schemaJson = schemaInfo.getSchema();
+ const auto& schemaJson = schemaInfo.getSchema();
size_t pos = schemaJson.find(R"(","rootMessageTypeName":)");
ASSERT_NE(pos, std::string::npos);
ASSERT_TRUE(pos > 0);
diff --git a/tests/ReaderConfigurationTest.cc b/tests/ReaderConfigurationTest.cc
index 5783ac9..cd043bd 100644
--- a/tests/ReaderConfigurationTest.cc
+++ b/tests/ReaderConfigurationTest.cc
@@ -78,7 +78,7 @@ TEST(ReaderConfigurationTest, testCustomConfig) {
ReaderConfiguration readerConf;
readerConf.setSchema(schema);
- readerConf.setReaderListener([](Reader, const Message&) {});
+ readerConf.setReaderListener([](const Reader&, const Message&) {});
readerConf.setReceiverQueueSize(2000);
readerConf.setReaderName("my-reader");
readerConf.setReadCompacted(true);
diff --git a/tests/ReaderTest.cc b/tests/ReaderTest.cc
index ad81949..3a35dad 100644
--- a/tests/ReaderTest.cc
+++ b/tests/ReaderTest.cc
@@ -39,7 +39,7 @@ static const std::string adminUrl = "http://localhost:8080/";
class ReaderTest : public ::testing::TestWithParam<bool> {
public:
- void initTopic(std::string topicName) {
+ void initTopic(const std::string& topicName) {
if (isMultiTopic_) {
// call admin api to make it partitioned
std::string url = adminUrl + "admin/v2/persistent/public/default/" + topicName + "/partitions";
@@ -903,7 +903,7 @@ TEST_F(ReaderSeekTest, testSeekInclusiveChunkMessage) {
MessageId secondMsgId;
ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeValue).build(), secondMsgId));
- auto assertStartMessageId = [&](bool inclusive, MessageId expectedMsgId) {
+ auto assertStartMessageId = [&](bool inclusive, const MessageId& expectedMsgId) {
Reader reader;
ReaderConfiguration readerConf;
readerConf.setStartMessageIdInclusive(inclusive);
diff --git a/tests/ServiceURITest.cc b/tests/ServiceURITest.cc
index 4e01ab2..8b03511 100644
--- a/tests/ServiceURITest.cc
+++ b/tests/ServiceURITest.cc
@@ -25,7 +25,7 @@ using namespace pulsar;
static void verifyServiceURIFailure(const std::string& uriString, const std::string& errorMsg) {
try {
ServiceURI uri{uriString};
- std::cerr << uriString << " should be invalid" << std::endl;
+ std::cerr << uriString << " should be invalid" << '\n';
FAIL();
} catch (const std::invalid_argument& e) {
EXPECT_EQ(errorMsg, e.what());
diff --git a/tests/SynchronizedHashMapTest.cc b/tests/SynchronizedHashMapTest.cc
index 3dba6bf..9bc1c52 100644
--- a/tests/SynchronizedHashMapTest.cc
+++ b/tests/SynchronizedHashMapTest.cc
@@ -96,21 +96,21 @@ TEST(SynchronizedHashMapTest, testForEach) {
m.clear();
int result = 0;
values.clear();
- m.forEachValue([&values](int value, SharedFuture) { values.emplace_back(value); },
+ m.forEachValue([&values](int value, const SharedFuture&) { values.emplace_back(value); },
[&result] { result = 1; });
ASSERT_TRUE(values.empty());
ASSERT_EQ(result, 1);
ASSERT_EQ(m.putIfAbsent(1, 100), boost::none);
ASSERT_EQ(m.putIfAbsent(1, 101), boost::optional<int>(100));
- m.forEachValue([&values](int value, SharedFuture) { values.emplace_back(value); },
+ m.forEachValue([&values](int value, const SharedFuture&) { values.emplace_back(value); },
[&result] { result = 2; });
ASSERT_EQ(values, (std::vector<int>({100})));
ASSERT_EQ(result, 1);
m.put(1, 102);
values.clear();
- m.forEachValue([&values](int value, SharedFuture) { values.emplace_back(value); },
+ m.forEachValue([&values](int value, const SharedFuture&) { values.emplace_back(value); },
[&result] { result = 2; });
ASSERT_EQ(values, (std::vector<int>({102})));
ASSERT_EQ(result, 1);
@@ -118,7 +118,7 @@ TEST(SynchronizedHashMapTest, testForEach) {
values.clear();
ASSERT_EQ(m.putIfAbsent(2, 200), boost::none);
ASSERT_EQ(m.putIfAbsent(2, 201), boost::optional<int>(200));
- m.forEachValue([&values](int value, SharedFuture) { values.emplace_back(value); },
+ m.forEachValue([&values](int value, const SharedFuture&) { values.emplace_back(value); },
[&result] { result = 2; });
std::sort(values.begin(), values.end());
ASSERT_EQ(values, (std::vector<int>({102, 200})));
diff --git a/tests/ZeroQueueSizeTest.cc b/tests/ZeroQueueSizeTest.cc
index b3ed066..35ea58b 100644
--- a/tests/ZeroQueueSizeTest.cc
+++ b/tests/ZeroQueueSizeTest.cc
@@ -19,11 +19,9 @@
#include <gtest/gtest.h>
#include <pulsar/Client.h>
-#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
-#include <memory>
#include <mutex>
#include "ConsumerTest.h"
@@ -41,7 +39,7 @@ static std::string lookupUrl = "pulsar://localhost:6650";
static std::string adminUrl = "http://localhost:8080";
static std::string contentBase = "msg-";
-static void messageListenerFunction(Consumer consumer, const Message& msg, Latch& latch) {
+static void messageListenerFunction(const Consumer& consumer, const Message& msg, Latch& latch) {
ASSERT_EQ(0, ConsumerTest::getNumOfMessagesInQueue(consumer));
std::ostringstream ss;
ss << contentBase << globalCount;
@@ -128,12 +126,12 @@ TEST(ZeroQueueSizeTest, testMessageListener) {
}
static ConsumerConfiguration zeroQueueSharedConsumerConf(
- const std::string& name, std::function<void(Consumer, const Message&)> callback) {
+ const std::string& name, const std::function<void(Consumer, const Message&)>& callback) {
ConsumerConfiguration conf;
conf.setConsumerType(ConsumerShared);
conf.setReceiverQueueSize(0);
conf.setSubscriptionInitialPosition(InitialPositionEarliest);
- conf.setMessageListener([name, callback](Consumer consumer, const Message& msg) {
+ conf.setMessageListener([name, callback](const Consumer& consumer, const Message& msg) {
LOG_INFO(name << " received " << msg.getDataAsString() << " from " << msg.getMessageId());
callback(consumer, msg);
});
diff --git a/tests/c/c_BasicEndToEndTest.cc b/tests/c/c_BasicEndToEndTest.cc
index e3ff19a..b319727 100644
--- a/tests/c/c_BasicEndToEndTest.cc
+++ b/tests/c/c_BasicEndToEndTest.cc
@@ -42,7 +42,7 @@ static void send_callback(pulsar_result async_result, pulsar_message_id_t *msg_i
send_ctx->result = async_result;
if (async_result == pulsar_result_Ok) {
const char *msg_id_str = pulsar_message_id_str(msg_id);
- send_ctx->msg_id = (char *)malloc(strlen(msg_id_str) * sizeof(char));
+ send_ctx->msg_id = (char *)malloc(strlen(msg_id_str) * sizeof(char) + 1);
strcpy(send_ctx->msg_id, msg_id_str);
}
send_ctx->promise->set_value();
@@ -55,7 +55,7 @@ static void receive_callback(pulsar_result async_result, pulsar_message_t *msg,
if (async_result == pulsar_result_Ok &&
pulsar_consumer_acknowledge(receive_ctx->consumer, msg) == pulsar_result_Ok) {
const char *data = (const char *)pulsar_message_get_data(msg);
- receive_ctx->data = (char *)malloc(strlen(data) * sizeof(char));
+ receive_ctx->data = (char *)malloc(strlen(data) * sizeof(char) + 1);
strcpy(receive_ctx->data, data);
}
receive_ctx->promise->set_value();
diff --git a/tests/oauth2/Oauth2Test.cc b/tests/oauth2/Oauth2Test.cc
index 158065e..216e355 100644
--- a/tests/oauth2/Oauth2Test.cc
+++ b/tests/oauth2/Oauth2Test.cc
@@ -86,7 +86,7 @@ TEST(Oauth2Test, testTlsTrustFilePath) {
}
int main(int argc, char* argv[]) {
- std::cout << "Load Oauth2 configs from " << gKeyPath << "..." << std::endl;
+ std::cout << "Load Oauth2 configs from " << gKeyPath << "..." << '\n';
boost::property_tree::ptree root;
boost::property_tree::read_json(gKeyPath, root);
gClientId = root.get<std::string>("client_id");
diff --git a/tests/unix/ConnectionFailTest.cc b/tests/unix/ConnectionFailTest.cc
index 1d49869..a05e6a3 100644
--- a/tests/unix/ConnectionFailTest.cc
+++ b/tests/unix/ConnectionFailTest.cc
@@ -53,7 +53,7 @@ class ConnectionFailTest : public ::testing::TestWithParam<int> {
limit.rlim_cur = fdLimit;
limit.rlim_max = maxFdCount_;
ASSERT_EQ(setrlimit(RLIMIT_NOFILE, &limit), 0);
- std::cout << "Updated fd limit to " << limit.rlim_cur << std::endl;
+ std::cout << "Updated fd limit to " << limit.rlim_cur << '\n';
}
};
@@ -88,7 +88,7 @@ TEST_P(ConnectionFailTest, test) {
break;
}
}
- std::cout << "Create producer, consumer and reader successfully when fdLimit is " << fdLimit << std::endl;
+ std::cout << "Create producer, consumer and reader successfully when fdLimit is " << fdLimit << '\n';
ASSERT_TRUE(fdLimit < 100);
}