fix: restore threads examples and add backward-compatible start() API

Author: ElFantasma

Cargo.lock

@@ -5,9 +5,12 @@ members = [
     "rt",
     "concurrency",
     "examples/bank",
+    "examples/bank_threads",
     "examples/name_server",
     "examples/ping_pong",
+    "examples/ping_pong_threads",
     "examples/updater",
+    "examples/updater_threads",
     "examples/blocking_genserver",
     "examples/busy_genserver_warning",
 ]

Cargo.toml

@@ -182,14 +182,19 @@ pub trait GenServer: Send + Sized {
     type OutMsg: Send + Sized;
     type Error: Debug + Send;
 
+    /// Start the GenServer with the default backend (Async).
+    fn start(self) -> GenServerHandle<Self> {
+        self.start_with_backend(Backend::default())
+    }
+
     /// Start the GenServer with the specified backend.
     ///
     /// # Arguments
     /// * `backend` - The execution backend to use:
     ///   - `Backend::Async` - Run on tokio async runtime (default, best for non-blocking workloads)
     ///   - `Backend::Blocking` - Run on tokio's blocking thread pool (for blocking operations)
     ///   - `Backend::Thread` - Run on a dedicated OS thread (for long-running blocking services)
-    fn start(self, backend: Backend) -> GenServerHandle<Self> {
+    fn start_with_backend(self, backend: Backend) -> GenServerHandle<Self> {
         match backend {
             Backend::Async => GenServerHandle::new(self),
             Backend::Blocking => GenServerHandle::new_blocking(self),
@@ -496,16 +501,15 @@ mod tests {
         }
     }
 
-    const ASYNC: Backend = Backend::Async;
     const BLOCKING: Backend = Backend::Blocking;
 
     #[test]
     pub fn badly_behaved_thread_non_blocking() {
         let runtime = rt::Runtime::new().unwrap();
         runtime.block_on(async move {
-            let mut badboy = BadlyBehavedTask.start(ASYNC);
+            let mut badboy = BadlyBehavedTask.start();
             let _ = badboy.cast(Unused).await;
-            let mut goodboy = WellBehavedTask { count: 0 }.start(ASYNC);
+            let mut goodboy = WellBehavedTask { count: 0 }.start();
             let _ = goodboy.cast(Unused).await;
             rt::sleep(Duration::from_secs(1)).await;
             let count = goodboy.call(InMessage::GetCount).await.unwrap();
@@ -523,9 +527,9 @@ mod tests {
     pub fn badly_behaved_thread() {
         let runtime = rt::Runtime::new().unwrap();
         runtime.block_on(async move {
-            let mut badboy = BadlyBehavedTask.start(BLOCKING);
+            let mut badboy = BadlyBehavedTask.start_with_backend(BLOCKING);
             let _ = badboy.cast(Unused).await;
-            let mut goodboy = WellBehavedTask { count: 0 }.start(ASYNC);
+            let mut goodboy = WellBehavedTask { count: 0 }.start();
             let _ = goodboy.cast(Unused).await;
             rt::sleep(Duration::from_secs(1)).await;
             let count = goodboy.call(InMessage::GetCount).await.unwrap();
@@ -580,7 +584,7 @@ mod tests {
     pub fn unresolving_task_times_out() {
         let runtime = rt::Runtime::new().unwrap();
         runtime.block_on(async move {
-            let mut unresolving_task = SomeTask.start(ASYNC);
+            let mut unresolving_task = SomeTask.start();
 
             let result = unresolving_task
                 .call_with_timeout(SomeTaskCallMsg::FastOperation, TIMEOUT_DURATION)
@@ -630,7 +634,7 @@ mod tests {
         runtime.block_on(async move {
             let (rx, tx) = mpsc::channel::<u8>();
             let sender_channel = Arc::new(Mutex::new(tx));
-            let _task = SomeTaskThatFailsOnInit::new(sender_channel).start(ASYNC);
+            let _task = SomeTaskThatFailsOnInit::new(sender_channel).start();
 
             // Wait a while to ensure the task has time to run and fail
             rt::sleep(Duration::from_secs(1)).await;

concurrency/src/tasks/gen_server.rs

@@ -1,6 +1,5 @@
 use crate::tasks::{
-    send_after, stream::spawn_listener, Backend, CallResponse, CastResponse, GenServer,
-    GenServerHandle,
+    send_after, stream::spawn_listener, CallResponse, CastResponse, GenServer, GenServerHandle,
 };
 use futures::{stream, StreamExt};
 use spawned_rt::tasks::{self as rt, BroadcastStream, ReceiverStream};
@@ -68,7 +67,7 @@ impl GenServer for Summatory {
 pub fn test_sum_numbers_from_stream() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
-        let mut summatory_handle = Summatory::new(0).start(Backend::Async);
+        let mut summatory_handle = Summatory::new(0).start();
         let stream = stream::iter(vec![1u16, 2, 3, 4, 5].into_iter().map(Ok::<u16, ()>));
 
         spawn_listener(
@@ -88,7 +87,7 @@ pub fn test_sum_numbers_from_stream() {
 pub fn test_sum_numbers_from_channel() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
-        let mut summatory_handle = Summatory::new(0).start(Backend::Async);
+        let mut summatory_handle = Summatory::new(0).start();
         let (tx, rx) = spawned_rt::tasks::mpsc::channel::<Result<u16, ()>>();
 
         // Spawn a task to send numbers to the channel
@@ -116,7 +115,7 @@ pub fn test_sum_numbers_from_channel() {
 pub fn test_sum_numbers_from_broadcast_channel() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
-        let mut summatory_handle = Summatory::new(0).start(Backend::Async);
+        let mut summatory_handle = Summatory::new(0).start();
         let (tx, rx) = tokio::sync::broadcast::channel::<u16>(5);
 
         // Spawn a task to send numbers to the channel
@@ -146,7 +145,7 @@ pub fn test_stream_cancellation() {
 
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
-        let mut summatory_handle = Summatory::new(0).start(Backend::Async);
+        let mut summatory_handle = Summatory::new(0).start();
         let (tx, rx) = spawned_rt::tasks::mpsc::channel::<Result<u16, ()>>();
 
         // Spawn a task to send numbers to the channel
@@ -193,7 +192,7 @@ pub fn test_stream_cancellation() {
 pub fn test_halting_on_stream_error() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
-        let mut summatory_handle = Summatory::new(0).start(Backend::Async);
+        let mut summatory_handle = Summatory::new(0).start();
         let stream = tokio_stream::iter(vec![Ok(1u16), Ok(2), Ok(3), Err(()), Ok(4), Ok(5)]);
         let msg_stream = stream.filter_map(|value| async move {
             match value {
@@ -217,7 +216,7 @@ pub fn test_halting_on_stream_error() {
 pub fn test_skipping_on_stream_error() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
-        let mut summatory_handle = Summatory::new(0).start(Backend::Async);
+        let mut summatory_handle = Summatory::new(0).start();
         let stream = tokio_stream::iter(vec![Ok(1u16), Ok(2), Ok(3), Err(()), Ok(4), Ok(5)]);
         let msg_stream = stream.filter_map(|value| async move {
             match value {

concurrency/src/tasks/stream_tests.rs

@@ -1,6 +1,6 @@
 use super::{
-    send_after, send_interval, Backend, CallResponse, CastResponse, GenServer, GenServerHandle,
-    InitResult, InitResult::Success,
+    send_after, send_interval, CallResponse, CastResponse, GenServer, GenServerHandle, InitResult,
+    InitResult::Success,
 };
 use spawned_rt::tasks::{self as rt, CancellationToken};
 use std::time::Duration;
@@ -102,7 +102,7 @@ pub fn test_send_interval_and_cancellation() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
         // Start a Repeater
-        let mut repeater = Repeater::new(0).start(Backend::Async);
+        let mut repeater = Repeater::new(0).start();
 
         // Wait for 1 second
         rt::sleep(Duration::from_secs(1)).await;
@@ -209,7 +209,7 @@ pub fn test_send_after_and_cancellation() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
         // Start a Delayed
-        let mut repeater = Delayed::new(0).start(Backend::Async);
+        let mut repeater = Delayed::new(0).start();
 
         // Set a just once timed message
         let _ = send_after(
@@ -253,7 +253,7 @@ pub fn test_send_after_gen_server_teardown() {
     let runtime = rt::Runtime::new().unwrap();
     runtime.block_on(async move {
         // Start a Delayed
-        let mut repeater = Delayed::new(0).start(Backend::Async);
+        let mut repeater = Delayed::new(0).start();
 
         // Set a just once timed message
         let _ = send_after(

concurrency/src/tasks/timer_tests.rs

@@ -24,13 +24,13 @@ mod server;
 
 use messages::{BankError, BankOutMessage};
 use server::Bank;
-use spawned_concurrency::tasks::{Backend, GenServer as _};
+use spawned_concurrency::tasks::GenServer as _;
 use spawned_rt::tasks as rt;
 
 fn main() {
     rt::run(async {
         // Starting the bank
-        let mut name_server = Bank::new().start(Backend::Async);
+        let mut name_server = Bank::new().start();
 
         // Testing initial balance for "main" account
         let result = Bank::withdraw(&mut name_server, "main".to_string(), 15).await;

examples/bank/src/main.rs

@@ -0,0 +1,13 @@
+[package]
+name = "bank_threads"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+spawned-rt = { workspace = true }
+spawned-concurrency = { workspace = true }
+tracing = { workspace = true }
+
+[[bin]]
+name = "bank_threads"
+path = "src/main.rs"

examples/bank_threads/Cargo.toml

@@ -0,0 +1,118 @@
+//! Simple example to test concurrency/Process abstraction.
+//!
+//! Based on Joe's Armstrong book: Programming Erlang, Second edition
+//! Section 22.1 - The Road to the Generic Server
+//!
+//! Erlang usage example:
+//! 1> my_bank:start().
+//! {ok,<0.33.0>}
+//! 2> my_bank:deposit("joe", 10).
+//! not_a_customer
+//! 3> my_bank:new_account("joe").
+//! {welcome,"joe"}
+//! 4> my_bank:deposit("joe", 10).
+//! {thanks,"joe",your_balance_is,10}
+//! 5> my_bank:deposit("joe", 30).
+//! {thanks,"joe",your_balance_is,40}
+//! 6> my_bank:withdraw("joe", 15).
+//! {thanks,"joe",your_balance_is,25}
+//! 7> my_bank:withdraw("joe", 45).
+//! {sorry,"joe",you_only_have,25,in_the_bank
+
+mod messages;
+mod server;
+
+use messages::{BankError, BankOutMessage};
+use server::Bank;
+use spawned_concurrency::threads::GenServer as _;
+use spawned_rt::threads as rt;
+
+fn main() {
+    rt::run(|| {
+        // Starting the bank
+        let mut name_server = Bank::new().start();
+
+        // Testing initial balance for "main" account
+        let result = Bank::withdraw(&mut name_server, "main".to_string(), 15);
+        tracing::info!("Withdraw result {result:?}");
+        assert_eq!(
+            result,
+            Ok(BankOutMessage::WidrawOk {
+                who: "main".to_string(),
+                amount: 985
+            })
+        );
+
+        let joe = "Joe".to_string();
+
+        // Error on deposit for an unexistent account
+        let result = Bank::deposit(&mut name_server, joe.clone(), 10);
+        tracing::info!("Deposit result {result:?}");
+        assert_eq!(result, Err(BankError::NotACustomer { who: joe.clone() }));
+
+        // Account creation
+        let result = Bank::new_account(&mut name_server, "Joe".to_string());
+        tracing::info!("New account result {result:?}");
+        assert_eq!(result, Ok(BankOutMessage::Welcome { who: joe.clone() }));
+
+        // Deposit
+        let result = Bank::deposit(&mut name_server, "Joe".to_string(), 10);
+        tracing::info!("Deposit result {result:?}");
+        assert_eq!(
+            result,
+            Ok(BankOutMessage::Balance {
+                who: joe.clone(),
+                amount: 10
+            })
+        );
+
+        // Deposit
+        let result = Bank::deposit(&mut name_server, "Joe".to_string(), 30);
+        tracing::info!("Deposit result {result:?}");
+        assert_eq!(
+            result,
+            Ok(BankOutMessage::Balance {
+                who: joe.clone(),
+                amount: 40
+            })
+        );
+
+        // Withdrawal
+        let result = Bank::withdraw(&mut name_server, "Joe".to_string(), 15);
+        tracing::info!("Withdraw result {result:?}");
+        assert_eq!(
+            result,
+            Ok(BankOutMessage::WidrawOk {
+                who: joe.clone(),
+                amount: 25
+            })
+        );
+
+        // Withdrawal with not enough balance
+        let result = Bank::withdraw(&mut name_server, "Joe".to_string(), 45);
+        tracing::info!("Withdraw result {result:?}");
+        assert_eq!(
+            result,
+            Err(BankError::InsufficientBalance {
+                who: joe.clone(),
+                amount: 25
+            })
+        );
+
+        // Full withdrawal
+        let result = Bank::withdraw(&mut name_server, "Joe".to_string(), 25);
+        tracing::info!("Withdraw result {result:?}");
+        assert_eq!(
+            result,
+            Ok(BankOutMessage::WidrawOk {
+                who: joe,
+                amount: 0
+            })
+        );
+
+        // Stopping the bank
+        let result = Bank::stop(&mut name_server);
+        tracing::info!("Stop result {result:?}");
+        assert_eq!(result, Ok(BankOutMessage::Stopped));
+    })
+}