You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/08/05 01:48:35 UTC
[pulsar] 01/04: [C++][Python] Add connection timeout configuration
(#11029)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7d401a589ab5d4142cac53510daa27930a52b5fb
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sat Jun 26 02:36:50 2021 +0800
[C++][Python] Add connection timeout configuration (#11029)
Fixes #10747
### Motivation
This PR is a catchup of https://github.com/apache/pulsar/pull/2852 and adds connection timeout configuration to C++ and Python client.
### Modifications
- Add a `PeriodicTask` class to execute tasks periodically and the relate unit tests: `PeriodicTastTest`.
- Use `PeriodicTask` to register a timer before connecting to broker asynchronously, if the connection was not established when the timer is triggered, close the socket so that `handleTcpConnected` can be triggered immediately with a failure.
- Add connection timeout (in milliseconds) to both C++ and Python clients.
- Add `ClientTest.testConnectTimeout` (C++) and `test_connect_timeout` (Python) and to verify the connection timeout works.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- PeriodicTaskTest
- ClientTest.testConnectTimeout
- test_connect_timeout
(cherry picked from commit 6062d2fd70936a691b5a2369580eaab01fe033f3)
---
.../include/pulsar/ClientConfiguration.h | 16 +++++
pulsar-client-cpp/lib/ClientConfiguration.cc | 8 +++
pulsar-client-cpp/lib/ClientConfigurationImpl.h | 1 +
pulsar-client-cpp/lib/ClientConnection.cc | 28 +++++++-
pulsar-client-cpp/lib/ClientConnection.h | 2 +
pulsar-client-cpp/lib/ClientImpl.cc | 3 +-
pulsar-client-cpp/lib/ExecutorService.h | 2 +
pulsar-client-cpp/lib/PeriodicTask.cc | 60 +++++++++++++++++
pulsar-client-cpp/lib/PeriodicTask.h | 76 ++++++++++++++++++++++
pulsar-client-cpp/python/pulsar/__init__.py | 7 +-
pulsar-client-cpp/python/pulsar_test.py | 18 ++++-
pulsar-client-cpp/python/src/config.cc | 2 +
pulsar-client-cpp/tests/ClientTest.cc | 28 ++++++++
pulsar-client-cpp/tests/PeriodicTaskTest.cc | 75 +++++++++++++++++++++
14 files changed, 322 insertions(+), 4 deletions(-)
diff --git a/pulsar-client-cpp/include/pulsar/ClientConfiguration.h b/pulsar-client-cpp/include/pulsar/ClientConfiguration.h
index 11bfc43..2f2c461 100644
--- a/pulsar-client-cpp/include/pulsar/ClientConfiguration.h
+++ b/pulsar-client-cpp/include/pulsar/ClientConfiguration.h
@@ -244,6 +244,22 @@ class PULSAR_PUBLIC ClientConfiguration {
*/
unsigned int getPartitionsUpdateInterval() const;
+ /**
+ * Set the duration of time to wait for a connection to a broker to be established. If the duration passes
+ * without a response from the broker, the connection attempt is dropped.
+ *
+ * Default: 10000
+ *
+ * @param timeoutMs the duration in milliseconds
+ * @return
+ */
+ ClientConfiguration& setConnectionTimeout(int timeoutMs);
+
+ /**
+ * The getter associated with setConnectionTimeout().
+ */
+ int getConnectionTimeout() const;
+
friend class ClientImpl;
friend class PulsarWrapper;
diff --git a/pulsar-client-cpp/lib/ClientConfiguration.cc b/pulsar-client-cpp/lib/ClientConfiguration.cc
index 52072a3..188980a 100644
--- a/pulsar-client-cpp/lib/ClientConfiguration.cc
+++ b/pulsar-client-cpp/lib/ClientConfiguration.cc
@@ -140,4 +140,12 @@ ClientConfiguration& ClientConfiguration::setListenerName(const std::string& lis
}
const std::string& ClientConfiguration::getListenerName() const { return impl_->listenerName; }
+
+ClientConfiguration& ClientConfiguration::setConnectionTimeout(int timeoutMs) {
+ impl_->connectionTimeoutMs = timeoutMs;
+ return *this;
+}
+
+int ClientConfiguration::getConnectionTimeout() const { return impl_->connectionTimeoutMs; }
+
} // namespace pulsar
diff --git a/pulsar-client-cpp/lib/ClientConfigurationImpl.h b/pulsar-client-cpp/lib/ClientConfigurationImpl.h
index 631e8ae..887ecf2 100644
--- a/pulsar-client-cpp/lib/ClientConfigurationImpl.h
+++ b/pulsar-client-cpp/lib/ClientConfigurationImpl.h
@@ -39,6 +39,7 @@ struct ClientConfigurationImpl {
bool validateHostName{false};
unsigned int partitionsUpdateInterval{60}; // 1 minute
std::string listenerName;
+ int connectionTimeoutMs{10000}; // 10 seconds
std::unique_ptr<LoggerFactory> takeLogger() { return std::move(loggerFactory); }
};
diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc
index 58a3cce..3b07115 100644
--- a/pulsar-client-cpp/lib/ClientConnection.cc
+++ b/pulsar-client-cpp/lib/ClientConnection.cc
@@ -175,6 +175,8 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
error_(boost::system::error_code()),
incomingBuffer_(SharedBuffer::allocate(DefaultBufferSize)),
incomingCmd_(),
+ connectTimeoutTask_(std::make_shared<PeriodicTask>(executor_->getIOService(),
+ clientConfiguration.getConnectionTimeout())),
pendingWriteBuffers_(),
pendingWriteOperations_(0),
outgoingBuffer_(SharedBuffer::allocate(DefaultBufferSize)),
@@ -374,6 +376,7 @@ void ClientConnection::handleTcpConnected(const boost::system::error_code& err,
LOG_INFO(cnxString_ << "Connected to broker through proxy. Logical broker: " << logicalAddress_);
}
state_ = TcpConnected;
+ connectTimeoutTask_->stop();
socket_->set_option(tcp::no_delay(true));
socket_->set_option(tcp::socket::keep_alive(true));
@@ -414,7 +417,13 @@ void ClientConnection::handleTcpConnected(const boost::system::error_code& err,
}
} else if (endpointIterator != tcp::resolver::iterator()) {
// The connection failed. Try the next endpoint in the list.
- socket_->close();
+ boost::system::error_code err;
+ socket_->close(err); // ignore the error of close
+ if (err) {
+ LOG_WARN(cnxString_ << "Failed to close socket: " << err.message());
+ }
+ connectTimeoutTask_->stop();
+ connectTimeoutTask_->start();
tcp::endpoint endpoint = *endpointIterator;
socket_->async_connect(endpoint, std::bind(&ClientConnection::handleTcpConnected, shared_from_this(),
std::placeholders::_1, ++endpointIterator));
@@ -500,6 +509,18 @@ void ClientConnection::handleResolve(const boost::system::error_code& err,
return;
}
+ auto self = shared_from_this();
+ connectTimeoutTask_->setCallback([this, self](const PeriodicTask::ErrorCode& ec) {
+ if (state_ != TcpConnected) {
+ LOG_ERROR(cnxString_ << "Connection was not established in " << connectTimeoutTask_->getPeriodMs()
+ << " ms, close the socket");
+ PeriodicTask::ErrorCode ignoredError;
+ socket_->close(ignoredError);
+ }
+ connectTimeoutTask_->stop();
+ });
+
+ connectTimeoutTask_->start();
if (endpointIterator != tcp::resolver::iterator()) {
LOG_DEBUG(cnxString_ << "Resolved hostname " << endpointIterator->host_name() //
<< " to " << endpointIterator->endpoint());
@@ -1412,6 +1433,9 @@ void ClientConnection::close() {
state_ = Disconnected;
boost::system::error_code err;
socket_->close(err);
+ if (err) {
+ LOG_WARN(cnxString_ << "Failed to close socket: " << err.message());
+ }
if (tlsSocket_) {
tlsSocket_->lowest_layer().close();
@@ -1442,6 +1466,8 @@ void ClientConnection::close() {
consumerStatsRequestTimer_.reset();
}
+ connectTimeoutTask_->stop();
+
lock.unlock();
LOG_INFO(cnxString_ << "Connection closed");
diff --git a/pulsar-client-cpp/lib/ClientConnection.h b/pulsar-client-cpp/lib/ClientConnection.h
index 71db1ad..a20219c 100644
--- a/pulsar-client-cpp/lib/ClientConnection.h
+++ b/pulsar-client-cpp/lib/ClientConnection.h
@@ -45,6 +45,7 @@
#include <pulsar/Client.h>
#include <set>
#include <lib/BrokerConsumerStatsImpl.h>
+#include "lib/PeriodicTask.h"
using namespace pulsar;
@@ -283,6 +284,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
proto::BaseCommand incomingCmd_;
Promise<Result, ClientConnectionWeakPtr> connectPromise_;
+ std::shared_ptr<PeriodicTask> connectTimeoutTask_;
typedef std::map<long, PendingRequestData> PendingRequestsMap;
PendingRequestsMap pendingRequests_;
diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc
index d9a2b57..b93ad2d 100644
--- a/pulsar-client-cpp/lib/ClientImpl.cc
+++ b/pulsar-client-cpp/lib/ClientImpl.cc
@@ -480,8 +480,9 @@ void ClientImpl::closeAsync(CloseCallback callback) {
state_ = Closing;
lock.unlock();
- LOG_INFO("Closing Pulsar client");
SharedInt numberOfOpenHandlers = std::make_shared<int>(producers.size() + consumers.size());
+ LOG_INFO("Closing Pulsar client with " << producers.size() << " producers and " << consumers.size()
+ << " consumers");
for (ProducersList::iterator it = producers.begin(); it != producers.end(); ++it) {
ProducerImplBasePtr producer = it->lock();
diff --git a/pulsar-client-cpp/lib/ExecutorService.h b/pulsar-client-cpp/lib/ExecutorService.h
index b673b79..d0ffc23 100644
--- a/pulsar-client-cpp/lib/ExecutorService.h
+++ b/pulsar-client-cpp/lib/ExecutorService.h
@@ -47,6 +47,8 @@ class PULSAR_PUBLIC ExecutorService : private boost::noncopyable {
void postWork(std::function<void(void)> task);
void close();
+ boost::asio::io_service &getIOService() { return *io_service_; }
+
private:
/*
* only called once and within lock so no need to worry about thread-safety
diff --git a/pulsar-client-cpp/lib/PeriodicTask.cc b/pulsar-client-cpp/lib/PeriodicTask.cc
new file mode 100644
index 0000000..f25a175
--- /dev/null
+++ b/pulsar-client-cpp/lib/PeriodicTask.cc
@@ -0,0 +1,60 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "lib/PeriodicTask.h"
+#include <boost/date_time/posix_time/posix_time.hpp>
+
+namespace pulsar {
+
+void PeriodicTask::start() {
+ if (state_ != Pending) {
+ return;
+ }
+ state_ = Ready;
+ if (periodMs_ >= 0) {
+ auto self = shared_from_this();
+ timer_.expires_from_now(boost::posix_time::millisec(periodMs_));
+ timer_.async_wait([this, self](const ErrorCode& ec) { handleTimeout(ec); });
+ }
+}
+
+void PeriodicTask::stop() {
+ State state = Ready;
+ if (!state_.compare_exchange_strong(state, Closing)) {
+ return;
+ }
+ timer_.cancel();
+ state_ = Pending;
+}
+
+void PeriodicTask::handleTimeout(const ErrorCode& ec) {
+ if (state_ != Ready) {
+ return;
+ }
+
+ callback_(ec);
+
+ // state_ may be changed in handleTimeout, so we check state_ again
+ if (state_ == Ready) {
+ auto self = shared_from_this();
+ timer_.expires_from_now(boost::posix_time::millisec(periodMs_));
+ timer_.async_wait([this, self](const ErrorCode& ec) { handleTimeout(ec); });
+ }
+}
+
+} // namespace pulsar
diff --git a/pulsar-client-cpp/lib/PeriodicTask.h b/pulsar-client-cpp/lib/PeriodicTask.h
new file mode 100644
index 0000000..57d0734
--- /dev/null
+++ b/pulsar-client-cpp/lib/PeriodicTask.h
@@ -0,0 +1,76 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+#include <functional>
+#include <memory>
+
+#include <boost/asio.hpp>
+
+namespace pulsar {
+
+/**
+ * A task that is executed periodically.
+ *
+ * After the `start()` method is called, it will trigger `callback_` method periodically whose interval is
+ * `periodMs` in the constructor. After the `stop()` method is called, the timer will be cancelled and
+ * `callback()` will never be called again unless `start()` was called again.
+ *
+ * If you don't want to execute the task infinitely, you can call `stop()` in the implementation of
+ * `callback()` method.
+ *
+ * NOTE: If the `periodMs` is negative, the `callback()` will never be called.
+ */
+class PeriodicTask : public std::enable_shared_from_this<PeriodicTask> {
+ public:
+ using ErrorCode = boost::system::error_code;
+ using CallbackType = std::function<void(const ErrorCode&)>;
+
+ enum State : std::uint8_t
+ {
+ Pending,
+ Ready,
+ Closing
+ };
+
+ PeriodicTask(boost::asio::io_service& ioService, int periodMs) : timer_(ioService), periodMs_(periodMs) {}
+
+ void start();
+
+ void stop();
+
+ void setCallback(CallbackType callback) noexcept { callback_ = callback; }
+
+ State getState() const noexcept { return state_; }
+ int getPeriodMs() const noexcept { return periodMs_; }
+
+ private:
+ std::atomic<State> state_{Pending};
+ boost::asio::deadline_timer timer_;
+ const int periodMs_;
+ CallbackType callback_{trivialCallback};
+
+ void handleTimeout(const ErrorCode& ec);
+
+ static void trivialCallback(const ErrorCode&) {}
+};
+
+} // namespace pulsar
diff --git a/pulsar-client-cpp/python/pulsar/__init__.py b/pulsar-client-cpp/python/pulsar/__init__.py
index 93df3b5..514ca11 100644
--- a/pulsar-client-cpp/python/pulsar/__init__.py
+++ b/pulsar-client-cpp/python/pulsar/__init__.py
@@ -363,7 +363,8 @@ class Client:
tls_trust_certs_file_path=None,
tls_allow_insecure_connection=False,
tls_validate_hostname=False,
- logger=None
+ logger=None,
+ connection_timeout_ms=10000,
):
"""
Create a new Pulsar client instance.
@@ -409,10 +410,13 @@ class Client:
the endpoint.
* `logger`:
Set a Python logger for this Pulsar client. Should be an instance of `logging.Logger`.
+ * `connection_timeout_ms`:
+ Set timeout in milliseconds on TCP connections.
"""
_check_type(str, service_url, 'service_url')
_check_type_or_none(Authentication, authentication, 'authentication')
_check_type(int, operation_timeout_seconds, 'operation_timeout_seconds')
+ _check_type(int, connection_timeout_ms, 'connection_timeout_ms')
_check_type(int, io_threads, 'io_threads')
_check_type(int, message_listener_threads, 'message_listener_threads')
_check_type(int, concurrent_lookup_requests, 'concurrent_lookup_requests')
@@ -427,6 +431,7 @@ class Client:
if authentication:
conf.authentication(authentication.auth)
conf.operation_timeout_seconds(operation_timeout_seconds)
+ conf.connection_timeout(connection_timeout_ms)
conf.io_threads(io_threads)
conf.message_listener_threads(message_listener_threads)
conf.concurrent_lookup_requests(concurrent_lookup_requests)
diff --git a/pulsar-client-cpp/python/pulsar_test.py b/pulsar-client-cpp/python/pulsar_test.py
index dd2b228..fc17c72 100755
--- a/pulsar-client-cpp/python/pulsar_test.py
+++ b/pulsar-client-cpp/python/pulsar_test.py
@@ -30,7 +30,7 @@ from pulsar import Client, MessageId, \
AuthenticationTLS, Authentication, AuthenticationToken, InitialPosition, \
CryptoKeyReader
-from _pulsar import ProducerConfiguration, ConsumerConfiguration
+from _pulsar import ProducerConfiguration, ConsumerConfiguration, ConnectError
from schema_test import *
@@ -1148,6 +1148,22 @@ class PulsarTest(TestCase):
consumer.receive(100)
client.close()
+ def test_connect_timeout(self):
+ client = pulsar.Client(
+ service_url='pulsar://192.0.2.1:1234',
+ connection_timeout_ms=1000, # 1 second
+ )
+ t1 = time.time()
+ try:
+ producer = client.create_producer('test_connect_timeout')
+ self.fail('create_producer should not succeed')
+ except ConnectError as expected:
+ print('expected error: {} when create producer'.format(expected))
+ t2 = time.time()
+ self.assertGreater(t2 - t1, 1.0)
+ self.assertLess(t2 - t1, 1.5) # 1.5 seconds is long enough
+ client.close()
+
def _check_value_error(self, fun):
with self.assertRaises(ValueError):
fun()
diff --git a/pulsar-client-cpp/python/src/config.cc b/pulsar-client-cpp/python/src/config.cc
index 5838208..b665ec7 100644
--- a/pulsar-client-cpp/python/src/config.cc
+++ b/pulsar-client-cpp/python/src/config.cc
@@ -195,6 +195,8 @@ void export_config() {
.def("authentication", &ClientConfiguration_setAuthentication, return_self<>())
.def("operation_timeout_seconds", &ClientConfiguration::getOperationTimeoutSeconds)
.def("operation_timeout_seconds", &ClientConfiguration::setOperationTimeoutSeconds, return_self<>())
+ .def("connection_timeout", &ClientConfiguration::getConnectionTimeout)
+ .def("connection_timeout", &ClientConfiguration::setConnectionTimeout, return_self<>())
.def("io_threads", &ClientConfiguration::getIOThreads)
.def("io_threads", &ClientConfiguration::setIOThreads, return_self<>())
.def("message_listener_threads", &ClientConfiguration::getMessageListenerThreads)
diff --git a/pulsar-client-cpp/tests/ClientTest.cc b/pulsar-client-cpp/tests/ClientTest.cc
index 91232dc..129a322 100644
--- a/pulsar-client-cpp/tests/ClientTest.cc
+++ b/pulsar-client-cpp/tests/ClientTest.cc
@@ -18,6 +18,7 @@
*/
#include <gtest/gtest.h>
+#include <future>
#include <pulsar/Client.h>
#include "../lib/checksum/ChecksumProvider.h"
@@ -86,3 +87,30 @@ TEST(ClientTest, testServerConnectError) {
ASSERT_EQ(ResultConnectError, client.createReader(topic, MessageId::earliest(), readerConf, reader));
client.close();
}
+
+TEST(ClientTest, testConnectTimeout) {
+ // 192.0.2.0/24 is assigned for documentation, should be a deadend
+ const std::string blackHoleBroker = "pulsar://192.0.2.1:1234";
+ const std::string topic = "test-connect-timeout";
+
+ Client clientLow(blackHoleBroker, ClientConfiguration().setConnectionTimeout(1000));
+ Client clientDefault(blackHoleBroker);
+
+ std::promise<Result> promiseLow;
+ clientLow.createProducerAsync(
+ topic, [&promiseLow](Result result, Producer producer) { promiseLow.set_value(result); });
+
+ std::promise<Result> promiseDefault;
+ clientDefault.createProducerAsync(
+ topic, [&promiseDefault](Result result, Producer producer) { promiseDefault.set_value(result); });
+
+ auto futureLow = promiseLow.get_future();
+ ASSERT_EQ(futureLow.wait_for(std::chrono::milliseconds(1500)), std::future_status::ready);
+ ASSERT_EQ(futureLow.get(), ResultConnectError);
+
+ auto futureDefault = promiseDefault.get_future();
+ ASSERT_EQ(futureDefault.wait_for(std::chrono::milliseconds(10)), std::future_status::timeout);
+
+ clientLow.close();
+ clientDefault.close();
+}
diff --git a/pulsar-client-cpp/tests/PeriodicTaskTest.cc b/pulsar-client-cpp/tests/PeriodicTaskTest.cc
new file mode 100644
index 0000000..11c1c62
--- /dev/null
+++ b/pulsar-client-cpp/tests/PeriodicTaskTest.cc
@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include <gtest/gtest.h>
+#include <atomic>
+#include <chrono>
+#include <thread>
+#include "lib/ExecutorService.h"
+#include "lib/LogUtils.h"
+#include "lib/PeriodicTask.h"
+
+DECLARE_LOG_OBJECT()
+
+using namespace pulsar;
+
+TEST(PeriodicTaskTest, testCountdownTask) {
+ ExecutorService executor;
+
+ std::atomic_int count{5};
+
+ auto task = std::make_shared<PeriodicTask>(executor.getIOService(), 200);
+ task->setCallback([task, &count](const PeriodicTask::ErrorCode& ec) {
+ if (--count <= 0) {
+ task->stop();
+ }
+ LOG_INFO("Now count is " << count << ", error code: " << ec.message());
+ });
+
+ // Wait for 2 seconds to verify callback won't be triggered after 1 second (200 ms * 5)
+ task->start();
+ std::this_thread::sleep_for(std::chrono::seconds(2));
+ LOG_INFO("Now count is " << count);
+ ASSERT_EQ(count.load(), 0);
+ task->stop(); // it's redundant, just to verify multiple stop() is idempotent
+
+ // Test start again
+ count = 1;
+ task->start();
+ std::this_thread::sleep_for(std::chrono::milliseconds(800));
+ LOG_INFO("Now count is " << count);
+ ASSERT_EQ(count.load(), 0);
+ task->stop();
+
+ executor.close();
+}
+
+TEST(PeriodicTaskTest, testNegativePeriod) {
+ ExecutorService executor;
+
+ auto task = std::make_shared<PeriodicTask>(executor.getIOService(), -1);
+ std::atomic_bool callbackTriggered{false};
+ task->setCallback([&callbackTriggered](const PeriodicTask::ErrorCode& ec) { callbackTriggered = true; });
+
+ task->start();
+ std::this_thread::sleep_for(std::chrono::seconds(1));
+ ASSERT_EQ(callbackTriggered.load(), false);
+ task->stop();
+
+ executor.close();
+}