Add event admin remote provider based on mqtt

bundles/event_admin/CMakeLists.txt

@@ -18,7 +18,9 @@
 celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" ON)
 if (EVENT_ADMIN)
     add_subdirectory(event_admin_api)
+    add_subdirectory(event_admin_spi)
     add_subdirectory(event_admin)
+    add_subdirectory(remote_provider)
     add_subdirectory(examples)
 endif()
 

bundles/event_admin/README.md

@@ -39,4 +39,5 @@ If we want to build the event admin examples, the cmake option `BUILD_EVENT_ADMI
 
 ## Event Admin Bundles
 
-* [EventAdmin](event_admin/README.md) - The event admin implementation.
+* [EventAdmin](event_admin/README.md) - The event admin implementation.
+* [RemoteProviders](remote_provider/README.md) - The remote providers implementation for the event admin. It is used to deliver events to remote frameworks. It is not a part of the OSGi Event Admin specification.

bundles/event_admin/event_admin/CMakeLists.txt

@@ -24,6 +24,7 @@ set(EVENT_ADMIN_SRC
 
 set(EVENT_ADMIN_DEPS
         Celix::event_admin_api
+        Celix::event_admin_spi
         Celix::log_helper
         Celix::framework
         Celix::utils

bundles/event_admin/event_admin/README.md

@@ -38,9 +38,10 @@ event admin pubsub model, and events are delivered asynchronously.
 
 ### Properties/Configuration
 
-| **Properties**                         | **Type** | **Description**                                               | **Default value** |
-|----------------------------------------|----------|---------------------------------------------------------------|-----|
-| **CELIX_EVENT_ADMIN_HANDLER_THREADS**  | long     | The number of event handler threads. Its maximum value is 20. | 5   |
+| **Properties**                                           | **Type** | **Description**                                               | **Default value** |
+|----------------------------------------------------------|----------|---------------------------------------------------------------|-------------------|
+| **CELIX_EVENT_ADMIN_HANDLER_THREADS**                    | long     | The number of event handler threads. Its maximum value is 20. | 5                 |
+| **CELIX_EVENT_ADMIN_EVENT_SEQID_CACHE_CLEANUP_INTERVAL** | long     | The event sequence id cache will be cleaned up when it has not been used for this interval. The unit is seconds. The event sequence id cache is used to prevent duplicate events. | (60*60)s          |
 
 ### Software Design
 
@@ -63,6 +64,17 @@ at most one event-delivery thread at a time, so that events can be delivered in
 "event.delivery" property to "async.unordered", the event handler can hold multiple event-delivery threads at the same 
 time, so that events can be delivered in parallel.
 
+#### Remote Event Delivery
+
+If the event property "celix.event.remote.enable" is set to true, the event will be delivered to the local event handlers
+and remote event handlers. For delivering events to local event handlers, it can refer to the section of synchronous delivery
+and asynchronous delivery. For delivering events to remote event handlers, event admin will forward the event to the 
+[remote provider](../remote_provider/README.md). The remote provider will serialize the event and send it to the remote framework.
+The remote framework will deserialize the event and deliver it to the remote event handler. The diagram of remote event delivery
+is as follows:
+
+![remote_delivery_seq.png](diagrams/remote_event_delivery_seq.png)
+
 
 #### Event Adapter
 

bundles/event_admin/event_admin/diagrams/remote_event_delivery_seq.puml

@@ -0,0 +1,39 @@
+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.
+
+@startuml
+'https://plantuml.com/sequence-diagram
+
+box FrameworkA
+participant "Event Admin" as EventAdmin1
+participant "Remote Provider" as RemoteProvider1
+end box
+
+box FrameworkB
+participant "Remote Provider" as RemoteProvider2
+participant "Event Admin" as EventAdmin2
+end box
+
+-\EventAdmin1:postEvent/sendEvent
+EventAdmin1->EventAdmin1:Delivery event to local event handlers
+alt "celix.event.remote.enable" is true
+    EventAdmin1->EventAdmin1:Unset the "celix.event.remote.enable" property
+    EventAdmin1->RemoteProvider1:postEvent/sendEvent
+    RemoteProvider1->RemoteProvider2:IPC or Network
+    RemoteProvider2->EventAdmin2:postEvent/sendEvent
+    EventAdmin2 -> EventAdmin2:Delivery event to event handlers in FrameworkB
+end alt
+
+@enduml

bundles/event_admin/event_admin/gtest/src/CelixEventAdminActivatorErrorInjectionTestSuite.cc

@@ -103,6 +103,30 @@ TEST_F(CelixEventAdminActTestSuite, FailedToAddEventHandlerDependencyToEventAdmi
     });
 }
 
+TEST_F(CelixEventAdminActTestSuite, FailedToCreateRemoteProviderDependencyForEventAdminTest) {
+    TestEventAdminActivator([](void *act, celix_bundle_context_t *ctx) {
+        celix_ei_expect_celix_dmServiceDependency_create((void*)&celix_bundleActivator_start, 1, nullptr, 2);
+        auto status = celix_bundleActivator_start(act, ctx);
+        ASSERT_EQ(CELIX_ENOMEM, status);
+    });
+}
+
+TEST_F(CelixEventAdminActTestSuite, FailedToSetServiceToRemoteProviderDependencyTest) {
+    TestEventAdminActivator([](void *act, celix_bundle_context_t *ctx) {
+        celix_ei_expect_celix_dmServiceDependency_setService((void*)&celix_bundleActivator_start, 1, CELIX_ENOMEM, 2);
+        auto status = celix_bundleActivator_start(act, ctx);
+        ASSERT_EQ(CELIX_ENOMEM, status);
+    });
+}
+
+TEST_F(CelixEventAdminActTestSuite, FailedToAddRemoteProviderDependencyToEventAdminComponentTest) {
+    TestEventAdminActivator([](void *act, celix_bundle_context_t *ctx) {
+        celix_ei_expect_celix_dmComponent_addServiceDependency((void*)&celix_bundleActivator_start, 1, CELIX_ENOMEM, 2);
+        auto status = celix_bundleActivator_start(act, ctx);
+        ASSERT_EQ(CELIX_ENOMEM, status);
+    });
+}
+
 TEST_F(CelixEventAdminActTestSuite, FailedToAddEventAdminServiceToComponentTest) {
     TestEventAdminActivator([](void *act, celix_bundle_context_t *ctx) {
         celix_ei_expect_celix_dmComponent_addInterface((void*)&celix_bundleActivator_start, 1, CELIX_ENOMEM);
@@ -129,23 +153,23 @@ TEST_F(CelixEventAdminActTestSuite, FailedToCreateEventAdapterTest) {
 
 TEST_F(CelixEventAdminActTestSuite, FailedToCreateEventAdminDependencyForEventAdapterTest) {
     TestEventAdminActivator([](void *act, celix_bundle_context_t *ctx) {
-        celix_ei_expect_celix_dmServiceDependency_create((void*)&celix_bundleActivator_start, 1, nullptr, 2);
+        celix_ei_expect_celix_dmServiceDependency_create((void*)&celix_bundleActivator_start, 1, nullptr, 3);
         auto status = celix_bundleActivator_start(act, ctx);
         ASSERT_EQ(CELIX_ENOMEM, status);
     });
 }
 
 TEST_F(CelixEventAdminActTestSuite, FailedToSetServiceToEventAdminDependencyTest) {
     TestEventAdminActivator([](void *act, celix_bundle_context_t *ctx) {
-        celix_ei_expect_celix_dmServiceDependency_setService((void*)&celix_bundleActivator_start, 1, CELIX_ENOMEM, 2);
+        celix_ei_expect_celix_dmServiceDependency_setService((void*)&celix_bundleActivator_start, 1, CELIX_ENOMEM, 3);
         auto status = celix_bundleActivator_start(act, ctx);
         ASSERT_EQ(CELIX_ENOMEM, status);
     });
 }
 
 TEST_F(CelixEventAdminActTestSuite, FailedToAddEventAdminDependencyToEventAdapterComponentTest) {
     TestEventAdminActivator([](void *act, celix_bundle_context_t *ctx) {
-        celix_ei_expect_celix_dmComponent_addServiceDependency((void*)&celix_bundleActivator_start, 1, CELIX_ENOMEM, 2);
+        celix_ei_expect_celix_dmComponent_addServiceDependency((void*)&celix_bundleActivator_start, 1, CELIX_ENOMEM, 3);
         auto status = celix_bundleActivator_start(act, ctx);
         ASSERT_EQ(CELIX_ENOMEM, status);
     });

bundles/event_admin/event_admin/gtest/src/CelixEventAdminErrorInjectionTestSuite.cc

@@ -16,6 +16,8 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+#include <cerrno>
+
 #include "CelixEventAdminTestSuiteBaseClass.h"
 #include "celix_event_admin.h"
 #include "celix_event.h"
@@ -27,6 +29,8 @@
 #include "celix_long_hash_map_ei.h"
 #include "celix_threads_ei.h"
 #include "celix_utils_ei.h"
+#include "celix_properties_ei.h"
+#include "celix_bundle_context_ei.h"
 #include "malloc_ei.h"
 #include <gtest/gtest.h>
 
@@ -51,6 +55,11 @@ class CelixEventAdminErrorInjectionTestSuite : public CelixEventAdminTestSuiteBa
         celix_ei_expect_celix_arrayList_addLong(nullptr, 0, 0);
         celix_ei_expect_celix_elapsedtime(nullptr, 0, 0);
         celix_ei_expect_celix_arrayList_createWithOptions(nullptr, 0, nullptr);
+        celix_ei_expect_celix_properties_copy(nullptr, 0, nullptr);
+        celix_ei_expect_celix_properties_set(nullptr, 0, 0);
+        celix_ei_expect_celix_properties_setLong(nullptr, 0, 0);
+        celix_ei_expect_celix_bundleContext_getProperty(nullptr, 0, nullptr);
+        celix_ei_expect_celix_stringHashMap_createWithOptions(nullptr, 0, nullptr);
     }
 };
 
@@ -89,6 +98,12 @@ TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToCreateLogHelperForEventAd
     EXPECT_EQ(nullptr, ea);
 }
 
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToGetFrameworkUUIDForEventAdminTest) {
+    celix_ei_expect_celix_bundleContext_getProperty((void*)&celix_eventAdmin_create, 0, nullptr);
+    auto ea = celix_eventAdmin_create(ctx.get());
+    EXPECT_EQ(nullptr, ea);
+}
+
 TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToCreateLockForEventAdminTest) {
     celix_ei_expect_celixThreadRwlock_create((void*)&celix_eventAdmin_create, 0, CELIX_ENOMEM);
     auto ea = celix_eventAdmin_create(ctx.get());
@@ -119,6 +134,18 @@ TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToCreateEventHandlersMapFor
     EXPECT_EQ(nullptr, ea);
 }
 
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToCreateEventSeqIdCacheForEventAdminTest) {
+    celix_ei_expect_celix_stringHashMap_createWithOptions((void*)&celix_eventAdmin_create, 0, nullptr);
+    auto ea = celix_eventAdmin_create(ctx.get());
+    EXPECT_EQ(nullptr, ea);
+}
+
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToCreateRemoteProviderMapForEventAdminTest) {
+    celix_ei_expect_celix_longHashMap_create((void*)&celix_eventAdmin_create, 0, nullptr, 2);
+    auto ea = celix_eventAdmin_create(ctx.get());
+    EXPECT_EQ(nullptr, ea);
+}
+
 TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToCreateMutexForEventAdminTest) {
     celix_ei_expect_celixThreadMutex_create((void*)&celix_eventAdmin_create, 0, CELIX_ENOMEM);
     auto ea = celix_eventAdmin_create(ctx.get());
@@ -339,4 +366,128 @@ TEST_F(CelixEventAdminErrorInjectionTestSuite, PostEventToBlacklistHandlerTest)
         EXPECT_STRNE("org/celix/test1", topic);
         return CELIX_SUCCESS;
     });
+}
+
+TEST_F(CelixEventAdminErrorInjectionTestSuite, RetrieveEventSeqIdCacheTest) {
+    int receivedEventCount = 0;
+    TestPublishEvent("org/celix/test", nullptr, [](celix_event_admin_t *ea) {
+        for (int i = 0; i < 32; i++) {
+            celix_autoptr(celix_properties_t) eventProps = celix_properties_create();
+            std::string uuid = "9748f803-5766-49f1-a2e9-" + std::to_string(i);
+            celix_properties_set(eventProps, CELIX_EVENT_REMOTE_FRAMEWORK_UUID, uuid.c_str());
+            celix_properties_setLong(eventProps, CELIX_EVENT_REMOTE_SEQ_ID, 1);
+            auto status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+            EXPECT_EQ(CELIX_SUCCESS, status);
+        }
+        celix_ei_expect_celix_elapsedtime(CELIX_EI_UNKNOWN_CALLER, 0, 60*60+1);
+        celix_autoptr(celix_properties_t) eventProps = celix_properties_create();
+        celix_properties_set(eventProps, CELIX_EVENT_REMOTE_FRAMEWORK_UUID, "9748f803-5766-49f1-a2e9-9bbb522e874b");
+        celix_properties_setLong(eventProps, CELIX_EVENT_REMOTE_SEQ_ID, 1);
+        auto status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(CELIX_SUCCESS, status);
+    }, [&receivedEventCount](void*, const char*, const celix_properties_t*) {
+        receivedEventCount ++;
+        return CELIX_SUCCESS;
+    });
+    EXPECT_EQ(33, receivedEventCount);
+}
+
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToCopyRemoteEnableEventPropertiesTest) {
+    celix_event_remote_provider_service_t remoteProviderService;
+    remoteProviderService.handle = nullptr;
+    remoteProviderService.sendEvent = [](void*, const char*, const celix_properties_t*) {
+        ADD_FAILURE() << "Should not be called";
+        return CELIX_SUCCESS;
+    };
+    remoteProviderService.postEvent = [](void*, const char*, const celix_properties_t*) {
+        ADD_FAILURE() << "Should not be called";
+        return CELIX_SUCCESS;
+    };
+    TestPublishEventToRemote([](celix_event_admin_t *ea) {
+        celix_ei_expect_celix_properties_copy((void*)&celix_eventAdmin_sendEvent, 1, nullptr);
+        celix_autoptr(celix_properties_t) eventProps = celix_properties_create();
+        celix_properties_setBool(eventProps, CELIX_EVENT_REMOTE_ENABLE, true);
+        auto status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(ENOMEM, status);
+    }, &remoteProviderService);
+}
+
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToSetRemoteFrameworkUUIDToRemoteEnableEventTest) {
+    celix_event_remote_provider_service_t remoteProviderService;
+    remoteProviderService.handle = nullptr;
+    remoteProviderService.sendEvent = [](void*, const char*, const celix_properties_t*) {
+        ADD_FAILURE() << "Should not be called";
+        return CELIX_SUCCESS;
+    };
+    remoteProviderService.postEvent = [](void*, const char*, const celix_properties_t*) {
+        ADD_FAILURE() << "Should not be called";
+        return CELIX_SUCCESS;
+    };
+    TestPublishEventToRemote([](celix_event_admin_t *ea) {
+        celix_ei_expect_celix_properties_set((void*)&celix_eventAdmin_sendEvent, 1, ENOMEM);
+        celix_autoptr(celix_properties_t) eventProps = celix_properties_create();
+        celix_properties_setBool(eventProps, CELIX_EVENT_REMOTE_ENABLE, true);
+        auto status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(ENOMEM, status);
+    }, &remoteProviderService);
+}
+
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedToSetSeqIdToRemoteEnableEventTest) {
+    celix_event_remote_provider_service_t remoteProviderService;
+    remoteProviderService.handle = nullptr;
+    remoteProviderService.sendEvent = [](void*, const char*, const celix_properties_t*) {
+        ADD_FAILURE() << "Should not be called";
+        return CELIX_SUCCESS;
+    };
+    remoteProviderService.postEvent = [](void*, const char*, const celix_properties_t*) {
+        ADD_FAILURE() << "Should not be called";
+        return CELIX_SUCCESS;
+    };
+    TestPublishEventToRemote([](celix_event_admin_t *ea) {
+        celix_ei_expect_celix_properties_setLong((void*)&celix_eventAdmin_sendEvent, 1, ENOMEM);
+        celix_autoptr(celix_properties_t) eventProps = celix_properties_create();
+        celix_properties_setBool(eventProps, CELIX_EVENT_REMOTE_ENABLE, true);
+        auto status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(ENOMEM, status);
+    }, &remoteProviderService);
+}
+
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedAllocMemoryForSeqIdCacheWhenSendRemoteEventTest) {
+    int receivedEventCount = 0;
+    TestPublishEvent("org/celix/test", nullptr, [](celix_event_admin_t *ea) {
+        celix_ei_expect_calloc((void*)&celix_eventAdmin_sendEvent, 2, nullptr);
+        celix_autoptr(celix_properties_t) eventProps = celix_properties_create();
+        celix_properties_set(eventProps, CELIX_EVENT_REMOTE_FRAMEWORK_UUID, "9748f803-5766-49f1-a2e9-9bbb522e874a");
+        celix_properties_setLong(eventProps, CELIX_EVENT_REMOTE_SEQ_ID, 1);
+        auto status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(CELIX_SUCCESS, status);
+
+        celix_properties_setLong(eventProps, CELIX_EVENT_REMOTE_SEQ_ID, 1);
+        status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(CELIX_SUCCESS, status);
+    }, [&receivedEventCount](void*, const char*, const celix_properties_t*) {
+        receivedEventCount ++;
+        return CELIX_SUCCESS;
+    });
+    EXPECT_EQ(2, receivedEventCount);
+}
+
+TEST_F(CelixEventAdminErrorInjectionTestSuite, FailedAddSeqIdCacheWhenSendRemoteEventTest) {
+    int receivedEventCount = 0;
+    TestPublishEvent("org/celix/test", nullptr, [](celix_event_admin_t *ea) {
+        celix_ei_expect_celix_stringHashMap_put((void*)&celix_eventAdmin_sendEvent, 2, ENOMEM);
+        celix_autoptr(celix_properties_t) eventProps = celix_properties_create();
+        celix_properties_set(eventProps, CELIX_EVENT_REMOTE_FRAMEWORK_UUID, "9748f803-5766-49f1-a2e9-9bbb522e874a");
+        celix_properties_setLong(eventProps, CELIX_EVENT_REMOTE_SEQ_ID, 1);
+        auto status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(CELIX_SUCCESS, status);
+
+        celix_properties_setLong(eventProps, CELIX_EVENT_REMOTE_SEQ_ID, 1);
+        status = celix_eventAdmin_sendEvent(ea, "org/celix/test", eventProps);
+        EXPECT_EQ(CELIX_SUCCESS, status);
+    }, [&receivedEventCount](void*, const char*, const celix_properties_t*) {
+        receivedEventCount ++;
+        return CELIX_SUCCESS;
+    });
+    EXPECT_EQ(2, receivedEventCount);
 }