fix some io_context interface changes

Author: BewareMyPower

lib/ClientConnection.h

@@ -26,13 +26,13 @@
 #include <cstdint>
 #ifdef USE_ASIO
 #include <asio/bind_executor.hpp>
-#include <asio/io_service.hpp>
+#include <asio/io_context.hpp>
 #include <asio/ip/tcp.hpp>
 #include <asio/ssl/stream.hpp>
 #include <asio/strand.hpp>
 #else
 #include <boost/asio/bind_executor.hpp>
-#include <boost/asio/io_service.hpp>
+#include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/ssl/stream.hpp>
 #include <boost/asio/strand.hpp>
@@ -325,7 +325,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
      */
     SocketPtr socket_;
     TlsSocketPtr tlsSocket_;
-    ASIO::strand<ASIO::io_service::executor_type> strand_;
+    ASIO::strand<ASIO::io_context::executor_type> strand_;
 
     const std::string logicalAddress_;
     /*

lib/ExecutorService.cc

@@ -31,17 +31,16 @@ ExecutorService::~ExecutorService() { close(0); }
 void ExecutorService::start() {
     auto self = shared_from_this();
     std::thread t{[this, self] {
-        LOG_DEBUG("Run io_service in a single thread");
-        ASIO_ERROR ec;
+        LOG_DEBUG("Run io_context in a single thread");
         while (!closed_) {
-            io_service_.restart();
-            IOService::work work{getIOService()};
-            io_service_.run(ec);
-        }
-        if (ec) {
-            LOG_ERROR("Failed to run io_service: " << ec.message());
-        } else {
-            LOG_DEBUG("Event loop of ExecutorService exits successfully");
+            io_context_.restart();
+            auto work_guard = ASIO::make_work_guard(io_context_);
+            try {
+                io_context_.run();
+                LOG_DEBUG("Event loop of ExecutorService exits successfully");
+            } catch (const ASIO_ERROR &e) {
+                LOG_ERROR("Failed to run io_context: " << e.message());
+            }
         }
         {
             std::lock_guard<std::mutex> lock{mutex_};
@@ -63,12 +62,12 @@ ExecutorServicePtr ExecutorService::create() {
 }
 
 /*
- *  factory method of ASIO::ip::tcp::socket associated with io_service_ instance
+ *  factory method of ASIO::ip::tcp::socket associated with io_context_ instance
  *  @ returns shared_ptr to this socket
  */
 SocketPtr ExecutorService::createSocket() {
     try {
-        return SocketPtr(new ASIO::ip::tcp::socket(io_service_));
+        return SocketPtr(new ASIO::ip::tcp::socket(io_context_));
     } catch (const ASIO_SYSTEM_ERROR &e) {
         restart();
         auto error = std::string("Failed to create socket: ") + e.what();
@@ -82,12 +81,12 @@ TlsSocketPtr ExecutorService::createTlsSocket(SocketPtr &socket, ASIO::ssl::cont
 }
 
 /*
- *  factory method of Resolver object associated with io_service_ instance
+ *  factory method of Resolver object associated with io_context_ instance
  *  @returns shraed_ptr to resolver object
  */
 TcpResolverPtr ExecutorService::createTcpResolver() {
     try {
-        return TcpResolverPtr(new ASIO::ip::tcp::resolver(io_service_));
+        return TcpResolverPtr(new ASIO::ip::tcp::resolver(io_context_));
     } catch (const ASIO_SYSTEM_ERROR &e) {
         restart();
         auto error = std::string("Failed to create resolver: ") + e.what();
@@ -97,36 +96,36 @@ TcpResolverPtr ExecutorService::createTcpResolver() {
 
 DeadlineTimerPtr ExecutorService::createDeadlineTimer() {
     try {
-        return DeadlineTimerPtr(new ASIO::steady_timer(io_service_));
+        return DeadlineTimerPtr(new ASIO::steady_timer(io_context_));
     } catch (const ASIO_SYSTEM_ERROR &e) {
         restart();
         auto error = std::string("Failed to create steady_timer: ") + e.what();
         throw std::runtime_error(error);
     }
 }
 
-void ExecutorService::restart() { io_service_.stop(); }
+void ExecutorService::restart() { io_context_.stop(); }
 
 void ExecutorService::close(long timeoutMs) {
     bool expectedState = false;
     if (!closed_.compare_exchange_strong(expectedState, true)) {
         return;
     }
     if (timeoutMs == 0) {  // non-blocking
-        io_service_.stop();
+        io_context_.stop();
         return;
     }
 
     std::unique_lock<std::mutex> lock{mutex_};
-    io_service_.stop();
+    io_context_.stop();
     if (timeoutMs > 0) {
         cond_.wait_for(lock, std::chrono::milliseconds(timeoutMs), [this] { return ioServiceDone_; });
     } else {  // < 0
         cond_.wait(lock, [this] { return ioServiceDone_; });
     }
 }
 
-void ExecutorService::postWork(std::function<void(void)> task) { io_service_.post(task); }
+void ExecutorService::postWork(std::function<void(void)> task) { ASIO::post(io_context_, std::move(task)); }
 
 /////////////////////
 

lib/ExecutorService.h

@@ -23,11 +23,11 @@
 
 #include <atomic>
 #ifdef USE_ASIO
-#include <asio/io_service.hpp>
+#include <asio/io_context.hpp>
 #include <asio/ip/tcp.hpp>
 #include <asio/ssl.hpp>
 #else
-#include <boost/asio/io_service.hpp>
+#include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/ssl.hpp>
 #endif
@@ -46,7 +46,7 @@ typedef std::shared_ptr<ASIO::ssl::stream<ASIO::ip::tcp::socket &> > TlsSocketPt
 typedef std::shared_ptr<ASIO::ip::tcp::resolver> TcpResolverPtr;
 class PULSAR_PUBLIC ExecutorService : public std::enable_shared_from_this<ExecutorService> {
    public:
-    using IOService = ASIO::io_service;
+    using IOService = ASIO::io_context;
     using SharedPtr = std::shared_ptr<ExecutorService>;
 
     static SharedPtr create();
@@ -67,14 +67,14 @@ class PULSAR_PUBLIC ExecutorService : public std::enable_shared_from_this<Execut
     // See TimeoutProcessor for the semantics of the parameter.
     void close(long timeoutMs = 3000);
 
-    IOService &getIOService() { return io_service_; }
+    IOService &getIOService() { return io_context_; }
     bool isClosed() const noexcept { return closed_; }
 
    private:
     /*
-     * io_service is our interface to os, io object schedule async ops on this object
+     * io_context is our interface to os, io object schedule async ops on this object
      */
-    IOService io_service_;
+    IOService io_context_;
 
     std::atomic_bool closed_{false};
     std::mutex mutex_;