fix: simplifying the ability to test using from_current (#242)

Enables `#[tokio::test]` via `Runtime::from_current()`

This uses the current handle as both the primary and secondary.

runtime/rust/src/component/namespace.rs

@@ -49,22 +49,19 @@ impl EventPublisher for Namespace {
     }
 }
 
+#[cfg(feature = "integration")]
 #[cfg(test)]
 mod tests {
     use super::*;
 
     // todo - make a distributed runtime fixture
     // todo - two options - fully mocked or integration test
-    #[cfg(feature = "integration")]
     #[tokio::test]
     async fn test_publish() {
-        // todo - use rtest - make fixtures
-        let dtr = DistributedRuntime::from_settings(Runtime::single_threaded().unwrap())
-            .await
-            .unwrap();
-
+        let rt = Runtime::from_current().unwrap();
+        let dtr = DistributedRuntime::from_settings(rt.clone()).await.unwrap();
         let ns = dtr.namespace("test".to_string()).unwrap();
-
         ns.publish("test", &"test".to_string()).await.unwrap();
+        rt.shutdown();
     }
 }

runtime/rust/src/component/service.rs

@@ -61,7 +61,7 @@ impl ServiceConfigBuilder {
         }
 
         // create service on the secondary runtime
-        let secondary = component.drt.runtime.secondary.clone();
+        let secondary = component.drt.runtime.secondary();
         let builder = component.drt.nats_client.client().service_builder();
         let service = secondary
             .spawn(async move {

runtime/rust/src/lib.rs

@@ -61,7 +61,7 @@ enum RuntimeType {
 pub struct Runtime {
     id: Arc<String>,
     primary: RuntimeType,
-    secondary: Arc<tokio::runtime::Runtime>,
+    secondary: RuntimeType,
     cancellation_token: CancellationToken,
 }
 

runtime/rust/src/logging.rs

@@ -56,7 +56,7 @@ use tracing_subscriber::{filter::Directive, fmt};
 const FILTER_ENV: &str = "TRD_LOG";
 
 /// Default log level
-const DEFAULT_FILTER_LEVEL: &str = "debug";
+const DEFAULT_FILTER_LEVEL: &str = "error";
 
 /// ENV used to set the path to the logging configuration file
 const CONFIG_PATH_ENV: &str = "TRD_LOGGING_CONFIG_PATH";

runtime/rust/src/runtime.rs

@@ -36,7 +36,7 @@ use tokio::{signal, task::JoinHandle};
 pub use tokio_util::sync::CancellationToken;
 
 impl Runtime {
-    fn new(runtime: RuntimeType) -> Result<Runtime> {
+    fn new(runtime: RuntimeType, secondary: Option<RuntimeType>) -> Result<Runtime> {
         // worker id
         let id = Arc::new(uuid::Uuid::new_v4().to_string());
 
@@ -44,19 +44,31 @@ impl Runtime {
         let cancellation_token = CancellationToken::new();
 
         // secondary runtime for background ectd/nats tasks
-        let secondary = RuntimeConfig::single_threaded().create_runtime()?;
+        let secondary = match secondary {
+            Some(secondary) => secondary,
+            None => {
+                RuntimeType::Shared(Arc::new(RuntimeConfig::single_threaded().create_runtime()?))
+            }
+        };
 
         Ok(Runtime {
             id,
             primary: runtime,
-            secondary: Arc::new(secondary),
+            secondary,
             cancellation_token,
         })
     }
 
+    pub fn from_current() -> Result<Runtime> {
+        let handle = tokio::runtime::Handle::current();
+        let primary = RuntimeType::External(handle.clone());
+        let secondary = RuntimeType::External(handle);
+        Runtime::new(primary, Some(secondary))
+    }
+
     pub fn from_handle(handle: tokio::runtime::Handle) -> Result<Runtime> {
         let runtime = RuntimeType::External(handle);
-        Runtime::new(runtime)
+        Runtime::new(runtime, None)
     }
 
     /// Create a [`Runtime`] instance from the settings
@@ -64,14 +76,14 @@ impl Runtime {
     pub fn from_settings() -> Result<Runtime> {
         let config = config::RuntimeConfig::from_settings()?;
         let owned = RuntimeType::Shared(Arc::new(config.create_runtime()?));
-        Runtime::new(owned)
+        Runtime::new(owned, None)
     }
 
     /// Create a [`Runtime`] with a single-threaded primary async tokio runtime
     pub fn single_threaded() -> Result<Runtime> {
         let config = config::RuntimeConfig::single_threaded();
         let owned = RuntimeType::Shared(Arc::new(config.create_runtime()?));
-        Runtime::new(owned)
+        Runtime::new(owned, None)
     }
 
     /// Returns the unique identifier for the [`Runtime`]
@@ -85,8 +97,8 @@ impl Runtime {
     }
 
     /// Returns a [`tokio::runtime::Handle`] for the secondary/background thread pool
-    pub fn secondary(&self) -> &Arc<tokio::runtime::Runtime> {
-        &self.secondary
+    pub fn secondary(&self) -> tokio::runtime::Handle {
+        self.secondary.handle()
     }
 
     /// Access the primary [`CancellationToken`] for the [`Runtime`]

runtime/rust/src/worker.rs

@@ -102,8 +102,9 @@ impl Worker {
         Fut: Future<Output = Result<()>> + Send + 'static,
     {
         let runtime = self.runtime;
+        let local_runtime = runtime.clone();
         let primary = runtime.primary();
-        let secondary = runtime.secondary.clone();
+        let secondary = runtime.secondary();
 
         let timeout = std::env::var(TRD_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT)
             .ok()
@@ -171,7 +172,9 @@ impl Worker {
             .take()
             .expect("Application initialized; but another thread is awaiting it; Worker.execute() can only be called once");
 
-        secondary.block_on(task)?
+        secondary.block_on(task)??;
+        local_runtime.shutdown();
+        Ok(())
     }
 }