Define event consumption interface for the application side event ser…

…vice

implementation.
Implement event pass-through to the applicaiton.

Refs #174

include/wsrep/event_service.hpp

@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2020 Codership Oy <[email protected]>
+ *
+ * This file is part of wsrep-lib.
+ *
+ * Wsrep-lib is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * Wsrep-lib is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+
+/** @file event_service.hpp
+ *
+ * Service interface for providing events to DBMS.
+ */
+
+#ifndef WSREP_EVENT_SERVICE_HPP
+#define WSREP_EVENT_SERVICE_HPP
+
+#include <string>
+
+namespace wsrep
+{
+
+    /** @class event_service
+     *
+     * Event service interface. This provides an interface corresponding
+     * to wsrep-API event service. For details see
+     * wsrep-API/wsrep_event_service.h
+     */
+    class event_service
+    {
+    public:
+        virtual ~event_service() { }
+
+        /**
+         * Process event with name name and value value.
+         */
+        virtual void process_event(const std::string& name,
+                                   const std::string& value) = 0;
+    };
+}
+
+#endif // WSREP_EVENT_SERVICE_HPP

include/wsrep/provider.hpp

@@ -47,6 +47,7 @@ namespace wsrep
     class thread_service;
     class tls_service;
     class allowlist_service;
+    class event_service;
 
     class stid
     {
@@ -424,10 +425,29 @@ namespace wsrep
             wsrep::thread_service* thread_service;
             wsrep::tls_service* tls_service;
             wsrep::allowlist_service* allowlist_service;
+            wsrep::event_service* event_service;
+
+            // some GCC and clang versions don't support C++11 default
+            // initializers fully, so we need to use explicit constructors
+            // instead:
+            // https://gcc.gnu.org/bugzilla/show_bug.cgi?id=88165
+            // https://bugs.llvm.org/show_bug.cgi?id=36684
             services()
                 : thread_service()
                 , tls_service()
                 , allowlist_service()
+                , event_service()
+            {
+            }
+
+            services(wsrep::thread_service* thr,
+                     wsrep::tls_service*    tls,
+                     wsrep::allowlist_service* all,
+                     wsrep::event_service*  event)
+                : thread_service(thr)
+                , tls_service(tls)
+                , allowlist_service(all)
+                , event_service(event)
             {
             }
         };

src/CMakeLists.txt

@@ -17,6 +17,7 @@ add_library(wsrep-lib
   thread.cpp
   thread_service_v1.cpp
   tls_service_v1.cpp
+  event_service_v1.cpp
   transaction.cpp
   uuid.cpp
   reporter.cpp

src/event_service_v1.cpp

@@ -0,0 +1,104 @@
+/*
+ * Copyright (C) 2020 Codership Oy <[email protected]>
+ *
+ * This file is part of wsrep-lib.
+ *
+ * Wsrep-lib is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * Wsrep-lib is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "event_service_v1.hpp"
+
+#include "wsrep/event_service.hpp"
+#include "wsrep/reporter.hpp"
+#include "wsrep/logger.hpp"
+#include "v26/wsrep_event_service.h"
+#include "service_helpers.hpp"
+
+#include <cassert>
+
+namespace wsrep_event_service_v1
+{
+    static std::atomic_flag initialized = ATOMIC_FLAG_INIT;
+
+    static void callback(
+        wsrep_event_context_t* ctx,
+        const char* name,
+        const char* value)
+    {
+        if (ctx)
+        {
+            wsrep::event_service* const impl
+                (reinterpret_cast<wsrep::event_service*>(ctx));
+            impl->process_event(name, value);
+        }
+    }
+
+    static const char* const log_string = "event service v1";
+}
+
+int wsrep::event_service_v1_probe(void* dlh)
+{
+    typedef int (*init_fn)(wsrep_event_service_v1_t*);
+    typedef void (*deinit_fn)();
+    if (wsrep_impl::service_probe<init_fn>(
+            dlh, WSREP_EVENT_SERVICE_INIT_FUNC_V1,
+            wsrep_event_service_v1::log_string) ||
+        wsrep_impl::service_probe<deinit_fn>(
+            dlh, WSREP_EVENT_SERVICE_DEINIT_FUNC_V1,
+            wsrep_event_service_v1::log_string))
+    {
+        // diagnostic message was logged by wsrep_impl::service_probe()
+        return 1;
+    }
+    return 0;
+}
+
+int wsrep::event_service_v1_init(void* dlh,
+                                 wsrep::event_service* event_service)
+{
+    if (not (dlh && event_service)) return EINVAL;
+
+    if (wsrep_event_service_v1::initialized.test_and_set()) return EALREADY;
+
+    wsrep_event_service_v1_t service =
+    {
+        wsrep_event_service_v1::callback,
+        reinterpret_cast<wsrep_event_context_t*>(event_service)
+    };
+
+    typedef int (*init_fn)(wsrep_event_service_v1_t*);
+    int const ret(wsrep_impl::service_init<init_fn>(
+                      dlh, WSREP_EVENT_SERVICE_INIT_FUNC_V1, &service,
+                      wsrep_event_service_v1::log_string));
+    if (ret)
+    {
+        wsrep_event_service_v1::initialized.clear();
+    }
+
+    return ret;
+}
+
+void wsrep::event_service_v1_deinit(void* dlh)
+{
+    if (wsrep_event_service_v1::initialized.test_and_set())
+    {
+        // service was initialized
+        typedef int (*deinit_fn)();
+        wsrep_impl::service_deinit<deinit_fn>(
+            dlh, WSREP_EVENT_SERVICE_DEINIT_FUNC_V1,
+            wsrep_event_service_v1::log_string);
+    }
+
+    wsrep_event_service_v1::initialized.clear();
+}

src/event_service_v1.hpp

@@ -0,0 +1,54 @@
+/*
+ * Copyright (C) 2020 Codership Oy <[email protected]>
+ *
+ * This file is part of wsrep-lib.
+ *
+ * Wsrep-lib is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * Wsrep-lib is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef WSREP_EVENT_SERVICE_V1_HPP
+#define WSREP_EVENT_SERVICE_V1_HPP
+
+namespace wsrep
+{
+    class event_service;
+    /**
+     * Probe event_service_v1 support in loaded library.
+     *
+     * @param dlh Handle returned by dlopen().
+     *
+     * @return Zero on success, non-zero system error code on failure.
+     */
+    int event_service_v1_probe(void *dlh);
+
+    /**
+     * Initialize event service.
+     *
+     * @param dlh Handle returned by dlopen().
+     * @params event_service Pointer to wsrep::event_service implementation.
+     *
+     * @return Zero on success, non-zero system error code on failure.
+     */
+    int event_service_v1_init(void* dlh,
+                              wsrep::event_service* event_service);
+
+    /**
+     * Deinitialize event service.
+     *
+     * @param dlh Handler returned by dlopen().
+     */
+    void event_service_v1_deinit(void* dlh);
+}
+
+#endif // WSREP_EVENT_SERVICE_V1_HPP

src/tls_service_v1.hpp

@@ -24,7 +24,7 @@ namespace wsrep
 {
     class tls_service;
     /**
-     * Probe thread_service_v1 support in loaded library.
+     * Probe tls_service_v1 support in loaded library.
      *
      * @param dlh Handle returned by dlopen().
      *
@@ -36,12 +36,12 @@ namespace wsrep
      * Initialize TLS service.
      *
      * @param dlh Handle returned by dlopen().
-     * @params thread_service Pointer to wsrep::thread_service implementation.
+     * @params tls_service Pointer to wsrep::thread_service implementation.
      *
      * @return Zero on success, non-zero system error code on failure.
      */
     int tls_service_v1_init(void* dlh,
-                            wsrep::tls_service* thread_service);
+                            wsrep::tls_service* tls_service);
 
     /**
      * Deinitialize TLS service.

src/wsrep_provider_v26.cpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2018 Codership Oy <[email protected]>
+ * Copyright (C) 2018-2021 Codership Oy <[email protected]>
  *
  * This file is part of wsrep-lib.
  *
@@ -32,6 +32,7 @@
 #include "thread_service_v1.hpp"
 #include "tls_service_v1.hpp"
 #include "allowlist_service_v1.hpp"
+#include "event_service_v1.hpp"
 #include "v26/wsrep_api.h"
 
 
@@ -460,7 +461,7 @@ namespace
         wsrep::const_buffer key(enc_ctx->key->ptr, enc_ctx->key->len);
         wsrep::const_buffer in(input->ptr, input->len);
         try
-        {   
+        {
             return server_state.encryption_service()->do_crypt(&enc_ctx->ctx,
                                                               key,
                                                               enc_ctx->iv,
@@ -654,8 +655,26 @@ namespace
         // assert(not wsrep::allowlist_service_v1_probe(dlh));
         wsrep::allowlist_service_v1_deinit(dlh);
     }
+
+    static int init_event_service(void* dlh,
+                                  wsrep::event_service* service)
+    {
+        assert(service);
+        if (not wsrep::event_service_v1_probe(dlh))
+        {
+            return wsrep::event_service_v1_init(dlh, service);
+        }
+        return 1;
+    }
+
+    static void deinit_event_service(void* dlh)
+    {
+        wsrep::event_service_v1_deinit(dlh);
+    }
 }
 
+
+
 void wsrep::wsrep_provider_v26::init_services(
     const wsrep::provider::services& services)
 {
@@ -683,10 +702,24 @@ void wsrep::wsrep_provider_v26::init_services(
         }
         services_enabled_.allowlist_service = services.allowlist_service;
     }
+    if (services.event_service)
+    {
+        if (init_event_service(wsrep_->dlh, services.event_service))
+        {
+            wsrep::log_warning() << "Failed to initialize event service";
+            // provider does not produce events, ignore
+        }
+        else
+        {
+            services_enabled_.event_service = services.event_service;
+        }
+    }
 }
 
 void wsrep::wsrep_provider_v26::deinit_services()
 {
+    if (services_enabled_.event_service)
+        deinit_event_service(wsrep_->dlh);
     if (services_enabled_.tls_service)
         deinit_tls_service(wsrep_->dlh);
     if (services_enabled_.thread_service)