Improve C++/Java Aeron tests and coverage

Author: cuihairu

.gitignore

@@ -55,6 +55,7 @@ dkms.conf
 
 # debug information files
 *.dwo
+*.gcov
 
 # macOS
 .DS_Store

cpp/CMakeLists.txt

@@ -4,6 +4,7 @@ project(epoch_cpp VERSION 0.1.0 LANGUAGES C CXX)
 add_library(epoch_cpp src/epoch.cpp src/engine.cpp src/actor_id.cpp src/aeron_transport.cpp)
 
 target_include_directories(epoch_cpp PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
+target_compile_definitions(epoch_cpp PRIVATE EPOCH_TESTING)
 
 target_compile_features(epoch_cpp PUBLIC cxx_std_17)
 
@@ -41,10 +42,21 @@ enable_testing()
 add_executable(epoch_cpp_test tests/epoch_vector_test.cpp)
 target_link_libraries(epoch_cpp_test PRIVATE epoch_cpp)
 add_test(NAME epoch_cpp_vector COMMAND epoch_cpp_test)
+target_compile_definitions(epoch_cpp_test PRIVATE EPOCH_TESTING)
+
+add_executable(epoch_cpp_core_test tests/core_test.cpp)
+target_link_libraries(epoch_cpp_core_test PRIVATE epoch_cpp)
+add_test(NAME epoch_cpp_core COMMAND epoch_cpp_core_test)
+target_compile_definitions(epoch_cpp_core_test PRIVATE EPOCH_TESTING)
+
+add_executable(epoch_cpp_aeron_test tests/aeron_transport_test.cpp)
+target_link_libraries(epoch_cpp_aeron_test PRIVATE epoch_cpp)
+add_test(NAME epoch_cpp_aeron COMMAND epoch_cpp_aeron_test)
+target_compile_definitions(epoch_cpp_aeron_test PRIVATE EPOCH_TESTING)
 
 option(EPOCH_COVERAGE "Enable coverage instrumentation" OFF)
 if (EPOCH_COVERAGE)
-    foreach(target epoch_cpp epoch_cpp_test)
+    foreach(target epoch_cpp epoch_cpp_test epoch_cpp_core_test epoch_cpp_aeron_test)
         target_compile_options(${target} PRIVATE -O0 -g --coverage)
         target_link_options(${target} PRIVATE --coverage)
     endforeach()

cpp/include/epoch/aeron_transport.h

@@ -55,4 +55,39 @@ class AeronTransport final : public Transport {
     aeron_subscription_t *subscription_ = nullptr;
 };
 
+namespace detail {
+
+struct AeronHooks {
+    int (*context_init)(aeron_context_t **);
+    int (*context_set_dir)(aeron_context_t *, const char *);
+    int (*init)(aeron_t **, aeron_context_t *);
+    int (*start)(aeron_t *);
+    int (*async_add_publication)(aeron_async_add_publication_t **, aeron_t *, const char *, int32_t);
+    int (*async_add_publication_poll)(aeron_publication_t **, aeron_async_add_publication_t *);
+    int (*async_add_subscription)(aeron_async_add_subscription_t **, aeron_t *, const char *, int32_t,
+                                  aeron_on_available_image_t, void *, aeron_on_unavailable_image_t, void *);
+    int (*async_add_subscription_poll)(aeron_subscription_t **, aeron_async_add_subscription_t *);
+    int64_t (*publication_offer)(aeron_publication_t *, const uint8_t *, size_t,
+                                 aeron_reserved_value_supplier_t, void *);
+    int (*subscription_poll)(aeron_subscription_t *, aeron_fragment_handler_t, void *, size_t);
+    int (*publication_close)(aeron_publication_t *, aeron_notification_t, void *);
+    int (*subscription_close)(aeron_subscription_t *, aeron_notification_t, void *);
+    int (*close)(aeron_t *);
+    int (*context_close)(aeron_context_t *);
+    const char *(*errmsg)();
+};
+
+AeronHooks &aeron_hooks();
+
+} // namespace detail
+
+#ifdef EPOCH_TESTING
+namespace test {
+
+detail::AeronHooks &aeron_hooks();
+void reset_aeron_hooks();
+
+} // namespace test
+#endif
+
 } // namespace epoch

cpp/src/aeron_transport.cpp

@@ -76,20 +76,78 @@ void throw_if_error(int result, const char *context)
 {
     if (result < 0)
     {
-        throw std::runtime_error(std::string(context) + ": " + aeron_errmsg());
+        throw std::runtime_error(std::string(context) + ": " + epoch::detail::aeron_hooks().errmsg());
     }
 }
 
 void throw_if_null(void *ptr, const char *context)
 {
     if (ptr == nullptr)
     {
-        throw std::runtime_error(std::string(context) + ": " + aeron_errmsg());
+        throw std::runtime_error(std::string(context) + ": " + epoch::detail::aeron_hooks().errmsg());
     }
 }
 
 } // namespace
 
+namespace detail {
+
+AeronHooks &aeron_hooks()
+{
+    static AeronHooks hooks{
+        aeron_context_init,
+        aeron_context_set_dir,
+        aeron_init,
+        aeron_start,
+        aeron_async_add_publication,
+        aeron_async_add_publication_poll,
+        aeron_async_add_subscription,
+        aeron_async_add_subscription_poll,
+        aeron_publication_offer,
+        aeron_subscription_poll,
+        aeron_publication_close,
+        aeron_subscription_close,
+        aeron_close,
+        aeron_context_close,
+        aeron_errmsg,
+    };
+    return hooks;
+}
+
+} // namespace detail
+
+#ifdef EPOCH_TESTING
+namespace test {
+
+detail::AeronHooks &aeron_hooks()
+{
+    return detail::aeron_hooks();
+}
+
+void reset_aeron_hooks()
+{
+    detail::aeron_hooks() = detail::AeronHooks{
+        aeron_context_init,
+        aeron_context_set_dir,
+        aeron_init,
+        aeron_start,
+        aeron_async_add_publication,
+        aeron_async_add_publication_poll,
+        aeron_async_add_subscription,
+        aeron_async_add_subscription_poll,
+        aeron_publication_offer,
+        aeron_subscription_poll,
+        aeron_publication_close,
+        aeron_subscription_close,
+        aeron_close,
+        aeron_context_close,
+        aeron_errmsg,
+    };
+}
+
+} // namespace test
+#endif
+
 AeronTransport::AeronTransport(AeronConfig config) : config_(std::move(config))
 {
     if (config_.fragment_limit <= 0)
@@ -104,25 +162,27 @@ AeronTransport::AeronTransport(AeronConfig config) : config_(std::move(config))
     try
     {
         aeron_context_t *context = nullptr;
-        throw_if_error(aeron_context_init(&context), "aeron_context_init failed");
+        throw_if_error(detail::aeron_hooks().context_init(&context), "aeron_context_init failed");
         if (!config_.aeron_directory.empty())
         {
-            throw_if_error(aeron_context_set_dir(context, config_.aeron_directory.c_str()),
+            throw_if_error(detail::aeron_hooks().context_set_dir(context, config_.aeron_directory.c_str()),
                            "aeron_context_set_dir failed");
         }
         context_ = context;
 
         aeron_t *client = nullptr;
-        throw_if_error(aeron_init(&client, context_), "aeron_init failed");
-        throw_if_error(aeron_start(client), "aeron_start failed");
+        throw_if_error(detail::aeron_hooks().init(&client, context_), "aeron_init failed");
+        throw_if_error(detail::aeron_hooks().start(client), "aeron_start failed");
         client_ = client;
 
         aeron_async_add_publication_t *pub_async = nullptr;
-        throw_if_error(aeron_async_add_publication(&pub_async, client_, config_.channel.c_str(), config_.stream_id),
+        throw_if_error(
+            detail::aeron_hooks().async_add_publication(
+                &pub_async, client_, config_.channel.c_str(), config_.stream_id),
                        "aeron_async_add_publication failed");
         while (true)
         {
-            int poll_result = aeron_async_add_publication_poll(&publication_, pub_async);
+            int poll_result = detail::aeron_hooks().async_add_publication_poll(&publication_, pub_async);
             if (poll_result == 1)
             {
                 break;
@@ -134,12 +194,19 @@ AeronTransport::AeronTransport(AeronConfig config) : config_(std::move(config))
 
         aeron_async_add_subscription_t *sub_async = nullptr;
         throw_if_error(
-            aeron_async_add_subscription(
-                &sub_async, client_, config_.channel.c_str(), config_.stream_id, nullptr, nullptr, nullptr, nullptr),
+            detail::aeron_hooks().async_add_subscription(
+                &sub_async,
+                client_,
+                config_.channel.c_str(),
+                config_.stream_id,
+                nullptr,
+                nullptr,
+                nullptr,
+                nullptr),
             "aeron_async_add_subscription failed");
         while (true)
         {
-            int poll_result = aeron_async_add_subscription_poll(&subscription_, sub_async);
+            int poll_result = detail::aeron_hooks().async_add_subscription_poll(&subscription_, sub_async);
             if (poll_result == 1)
             {
                 break;
@@ -174,7 +241,8 @@ void AeronTransport::send(const Message &message)
     std::int64_t result = 0;
     do
     {
-        result = aeron_publication_offer(publication_, buffer.data(), buffer.size(), nullptr, nullptr);
+        result = detail::aeron_hooks().publication_offer(
+            publication_, buffer.data(), buffer.size(), nullptr, nullptr);
         if (result >= 0)
         {
             stats_.sent_count++;
@@ -238,7 +306,7 @@ std::vector<Message> AeronTransport::poll(std::size_t max)
         ctx->stats->received_count++;
     };
 
-    int fragments = aeron_subscription_poll(subscription_, handler, &context, limit);
+    int fragments = detail::aeron_hooks().subscription_poll(subscription_, handler, &context, limit);
     throw_if_error(fragments, "aeron_subscription_poll failed");
     return out;
 }
@@ -253,22 +321,22 @@ void AeronTransport::close()
 
     if (publication_ != nullptr)
     {
-        aeron_publication_close(publication_, nullptr, nullptr);
+        detail::aeron_hooks().publication_close(publication_, nullptr, nullptr);
         publication_ = nullptr;
     }
     if (subscription_ != nullptr)
     {
-        aeron_subscription_close(subscription_, nullptr, nullptr);
+        detail::aeron_hooks().subscription_close(subscription_, nullptr, nullptr);
         subscription_ = nullptr;
     }
     if (client_ != nullptr)
     {
-        aeron_close(client_);
+        detail::aeron_hooks().close(client_);
         client_ = nullptr;
     }
     if (context_ != nullptr)
     {
-        aeron_context_close(context_);
+        detail::aeron_hooks().context_close(context_);
         context_ = nullptr;
     }
 }