ApiListener: listen(2) via Boost ASIO

diff --git a/lib/remote/apilistener.cpp b/lib/remote/apilistener.cpp
index a26846d38bfc0ef0689e051fe915428439e55921..d8e1832eed3fd5a0751801039cda74e23e5fc9a7 100644 (file)
--- a/lib/remote/apilistener.cpp
+++ b/lib/remote/apilistener.cpp
@@ -7,6 +7,7 @@
 #include "remote/jsonrpc.hpp"
 #include "remote/apifunction.hpp"
 #include "base/convert.hpp"
+#include "base/io-engine.hpp"
 #include "base/netstring.hpp"
 #include "base/json.hpp"
 #include "base/configtype.hpp"
@@ -18,7 +19,12 @@
 #include "base/context.hpp"
 #include "base/statsfunction.hpp"
 #include "base/exception.hpp"
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/ip/v6_only.hpp>
+#include <boost/asio/spawn.hpp>
+#include <climits>
 #include <fstream>
+#include <memory>
 
 using namespace icinga;
 
@@ -326,6 +332,10 @@ bool ApiListener::IsMaster() const
  */
 bool ApiListener::AddListener(const String& node, const String& service)
 {
+       namespace asio = boost::asio;
+       namespace ip = asio::ip;
+       using ip::tcp;
+
        ObjectLock olock(this);
 
        std::shared_ptr<SSL_CTX> sslContext = m_SSLContext;
@@ -335,45 +345,38 @@ bool ApiListener::AddListener(const String& node, const String& service)
                return false;
        }
 
-       TcpSocket::Ptr server = new TcpSocket();
+       auto& io (IoEngine::Get().GetIoService());
+       auto acceptor (std::make_shared<tcp::acceptor>(io));
 
        try {
-               server->Bind(node, service, AF_UNSPEC);
+               tcp::resolver resolver (io);
+               tcp::resolver::query query (node, service, tcp::resolver::query::passive);
+               auto endpoint (resolver.resolve(query)->endpoint());
+
+               acceptor->open(endpoint.protocol());
+               acceptor->set_option(ip::v6_only(false));
+               acceptor->set_option(tcp::acceptor::reuse_address(true));
+               acceptor->bind(endpoint);
        } catch (const std::exception&) {
                Log(LogCritical, "ApiListener")
                        << "Cannot bind TCP socket for host '" << node << "' on port '" << service << "'.";
                return false;
        }
 
-       Log(LogInformation, "ApiListener")
-               << "Started new listener on '" << server->GetClientAddress() << "'";
-
-       std::thread thread(std::bind(&ApiListener::ListenerThreadProc, this, server));
-       thread.detach();
+       acceptor->listen(INT_MAX);
 
-       m_Servers.insert(server);
+       auto localEndpoint (acceptor->local_endpoint());
 
-       UpdateStatusFile(server);
-
-       return true;
-}
-
-void ApiListener::ListenerThreadProc(const Socket::Ptr& server)
-{
-       Utility::SetThreadName("API Listener");
+       Log(LogInformation, "ApiListener")
+               << "Started new listener on '[" << localEndpoint.address() << "]:" << localEndpoint.port() << "'";
 
-       server->Listen();
+       asio::spawn(io, [acceptor](asio::yield_context yc) {
+               // TODO
+       });
 
-       for (;;) {
-               try {
-                       Socket::Ptr client = server->Accept();
+       UpdateStatusFile(localEndpoint);
 
-                       /* Use dynamic thread pool with additional on demand resources with fast throughput. */
-                       EnqueueAsyncCallback(std::bind(&ApiListener::NewClientHandler, this, client, String(), RoleServer), LowLatencyScheduler);
-               } catch (const std::exception&) {
-                       Log(LogCritical, "ApiListener", "Cannot accept new connection.");
-               }
-       }
+       return true;
 }
 
 /**
@@ -1513,14 +1516,13 @@ String ApiListener::GetFromZoneName(const Zone::Ptr& fromZone)
        return fromZoneName;
 }
 
-void ApiListener::UpdateStatusFile(TcpSocket::Ptr socket)
+void ApiListener::UpdateStatusFile(boost::asio::ip::tcp::endpoint localEndpoint)
 {
        String path = Configuration::CacheDir + "/api-state.json";
-       std::pair<String, String> details = socket->GetClientAddressDetails();
 
        Utility::SaveJsonFile(path, 0644, new Dictionary({
-               {"host", details.first},
-               {"port", Convert::ToLong(details.second)}
+               {"host", String(localEndpoint.address().to_string())},
+               {"port", localEndpoint.port()}
        }));
 }
 

diff --git a/lib/remote/apilistener.hpp b/lib/remote/apilistener.hpp
index 54b96dee5d2e40b74c3cdc788c78cd654a9eadbe..96861d74b75b7cf653b7366a5b16e553cc65e00a 100644 (file)
--- a/lib/remote/apilistener.hpp
+++ b/lib/remote/apilistener.hpp
@@ -14,6 +14,7 @@
 #include "base/tcpsocket.hpp"
 #include "base/tlsstream.hpp"
 #include "base/threadpool.hpp"
+#include <boost/asio/ip/tcp.hpp>
 #include <set>
 
 namespace icinga
@@ -106,7 +107,6 @@ protected:
 
 private:
        std::shared_ptr<SSL_CTX> m_SSLContext;
-       std::set<TcpSocket::Ptr> m_Servers;
 
        mutable boost::mutex m_AnonymousClientsLock;
        mutable boost::mutex m_HttpClientsLock;
@@ -130,7 +130,6 @@ private:
 
        void NewClientHandler(const Socket::Ptr& client, const String& hostname, ConnectionRole role);
        void NewClientHandlerInternal(const Socket::Ptr& client, const String& hostname, ConnectionRole role);
-       void ListenerThreadProc(const Socket::Ptr& server);
 
        static ThreadPool& GetTP();
        static void EnqueueAsyncCallback(const std::function<void ()>& callback, SchedulerPolicy policy = DefaultScheduler);
@@ -154,7 +153,7 @@ private:
 
        static void CopyCertificateFile(const String& oldCertPath, const String& newCertPath);
 
-       void UpdateStatusFile(TcpSocket::Ptr socket);
+       void UpdateStatusFile(boost::asio::ip::tcp::endpoint localEndpoint);
        void RemoveStatusFile();
 
        /* filesync */