Struct Bus 

pub struct Bus { /* private fields */ }

A bus for passing messages across threads.

Nysa buses are fully thread-safe, and thus, can be stored in static variables. The library provides a “default” bus in the module crate::global.

Implementations

impl Bus

pub fn new() -> Self

Creates a new bus.

Examples found in repository

examples/usage.rs (line 22)

fn local_bus() {
   // The bus needs to be behind an `Arc`, such that we can share its instance between threads.
   // Access to buses is inherently gated behind a mutex, so all we need to do is share the data.
   let bus = Arc::new(Bus::new());

   let adder = {
      // Now we can spawn the adder thread. We need to clone the Arc such that we can send it into
      // the thread, while still preserving our original instance.
      let bus = Arc::clone(&bus);
      std::thread::spawn(move || loop {
         // The thread will wait indefinitely until it receives an Add message. It's good practice
         // to specify which message you want to receive explicitly, as type inference can hide
         // that information away.
         match bus.wait_for::<Add>().consume() {
            Add::Two(a, b) => bus.push(AdditionResult(a + b)),
            Add::Quit => break,
         }
      })
   };

   // Now that the adder thread is spinned up and ready to go, we'll send it some requests to
   // add numbers. The requests will be delayed by a second, to demonstrate that the thread will
   // indeed block until new messages arrive.
   bus.push(Add::Two(1, 2));
   std::thread::sleep(Duration::from_secs(1));
   bus.push(Add::Two(4, 5));
   std::thread::sleep(Duration::from_secs(1));
   // After all these requests, we'll send a final shutdown request, and wait until the thread
   // finishes execution.
   bus.push(Add::Quit);
   adder.join().unwrap();

   // Now we can take a look at our results and print them all out. Note that while `retrieve_all`
   // is performing its job of draining the message queue, the bus for the given type of messages
   // is locked and new messages will not be pushed until the loop finishes.
   for message in &bus.retrieve_all::<AdditionResult>() {
      let AdditionResult(x) = message.consume();
      println!("{}", x);
   }
}

pub fn push<T>(&self, message_data: T)

where T: 'static + Send,

Pushes a message with the given data onto the bus.

Examples found in repository

examples/usage.rs (line 33)

fn local_bus() {
   // The bus needs to be behind an `Arc`, such that we can share its instance between threads.
   // Access to buses is inherently gated behind a mutex, so all we need to do is share the data.
   let bus = Arc::new(Bus::new());

   let adder = {
      // Now we can spawn the adder thread. We need to clone the Arc such that we can send it into
      // the thread, while still preserving our original instance.
      let bus = Arc::clone(&bus);
      std::thread::spawn(move || loop {
         // The thread will wait indefinitely until it receives an Add message. It's good practice
         // to specify which message you want to receive explicitly, as type inference can hide
         // that information away.
         match bus.wait_for::<Add>().consume() {
            Add::Two(a, b) => bus.push(AdditionResult(a + b)),
            Add::Quit => break,
         }
      })
   };

   // Now that the adder thread is spinned up and ready to go, we'll send it some requests to
   // add numbers. The requests will be delayed by a second, to demonstrate that the thread will
   // indeed block until new messages arrive.
   bus.push(Add::Two(1, 2));
   std::thread::sleep(Duration::from_secs(1));
   bus.push(Add::Two(4, 5));
   std::thread::sleep(Duration::from_secs(1));
   // After all these requests, we'll send a final shutdown request, and wait until the thread
   // finishes execution.
   bus.push(Add::Quit);
   adder.join().unwrap();

   // Now we can take a look at our results and print them all out. Note that while `retrieve_all`
   // is performing its job of draining the message queue, the bus for the given type of messages
   // is locked and new messages will not be pushed until the loop finishes.
   for message in &bus.retrieve_all::<AdditionResult>() {
      let AdditionResult(x) = message.consume();
      println!("{}", x);
   }
}

pub fn retrieve_all<'bus, T>(&'bus self) -> RetrieveAllRef<'bus, T>

where T: 'static + Send,

Retrieves all messages of the given type from the bus.

Note that the bus for the given message type is locked for the entire duration of this loop.

See also

• Bus::wait_for
• Bus::wait_for_timeout

Examples found in repository

examples/usage.rs (line 54)

fn local_bus() {
   // The bus needs to be behind an `Arc`, such that we can share its instance between threads.
   // Access to buses is inherently gated behind a mutex, so all we need to do is share the data.
   let bus = Arc::new(Bus::new());

   let adder = {
      // Now we can spawn the adder thread. We need to clone the Arc such that we can send it into
      // the thread, while still preserving our original instance.
      let bus = Arc::clone(&bus);
      std::thread::spawn(move || loop {
         // The thread will wait indefinitely until it receives an Add message. It's good practice
         // to specify which message you want to receive explicitly, as type inference can hide
         // that information away.
         match bus.wait_for::<Add>().consume() {
            Add::Two(a, b) => bus.push(AdditionResult(a + b)),
            Add::Quit => break,
         }
      })
   };

   // Now that the adder thread is spinned up and ready to go, we'll send it some requests to
   // add numbers. The requests will be delayed by a second, to demonstrate that the thread will
   // indeed block until new messages arrive.
   bus.push(Add::Two(1, 2));
   std::thread::sleep(Duration::from_secs(1));
   bus.push(Add::Two(4, 5));
   std::thread::sleep(Duration::from_secs(1));
   // After all these requests, we'll send a final shutdown request, and wait until the thread
   // finishes execution.
   bus.push(Add::Quit);
   adder.join().unwrap();

   // Now we can take a look at our results and print them all out. Note that while `retrieve_all`
   // is performing its job of draining the message queue, the bus for the given type of messages
   // is locked and new messages will not be pushed until the loop finishes.
   for message in &bus.retrieve_all::<AdditionResult>() {
      let AdditionResult(x) = message.consume();
      println!("{}", x);
   }
}

pub fn wait_for<'bus, T>(&'bus self) -> Message<'bus, T>

where T: 'static + Send,

Blocks execution in the current thread indefinitely until a message of the provided type arrives on the bus.

Although the Rust style guidelines say otherwise, to prevent bugs and aid readability it’s best to always use the turbofish syntax over implicit type inference with this function, such that it’s immediately visible what type of message is being waited for.

See also

• Bus::retrieve_all
• Bus::wait_for_timeout

Examples found in repository

examples/usage.rs (line 32)

fn local_bus() {
   // The bus needs to be behind an `Arc`, such that we can share its instance between threads.
   // Access to buses is inherently gated behind a mutex, so all we need to do is share the data.
   let bus = Arc::new(Bus::new());

   let adder = {
      // Now we can spawn the adder thread. We need to clone the Arc such that we can send it into
      // the thread, while still preserving our original instance.
      let bus = Arc::clone(&bus);
      std::thread::spawn(move || loop {
         // The thread will wait indefinitely until it receives an Add message. It's good practice
         // to specify which message you want to receive explicitly, as type inference can hide
         // that information away.
         match bus.wait_for::<Add>().consume() {
            Add::Two(a, b) => bus.push(AdditionResult(a + b)),
            Add::Quit => break,
         }
      })
   };

   // Now that the adder thread is spinned up and ready to go, we'll send it some requests to
   // add numbers. The requests will be delayed by a second, to demonstrate that the thread will
   // indeed block until new messages arrive.
   bus.push(Add::Two(1, 2));
   std::thread::sleep(Duration::from_secs(1));
   bus.push(Add::Two(4, 5));
   std::thread::sleep(Duration::from_secs(1));
   // After all these requests, we'll send a final shutdown request, and wait until the thread
   // finishes execution.
   bus.push(Add::Quit);
   adder.join().unwrap();

   // Now we can take a look at our results and print them all out. Note that while `retrieve_all`
   // is performing its job of draining the message queue, the bus for the given type of messages
   // is locked and new messages will not be pushed until the loop finishes.
   for message in &bus.retrieve_all::<AdditionResult>() {
      let AdditionResult(x) = message.consume();
      println!("{}", x);
   }
}

pub fn wait_for_timeout<'bus, T>( &'bus self, timeout: Duration, ) -> Option<Message<'bus, T>>

where T: 'static + Send,

Blocks execution in the current thread until a message of the provided type arrives on the bus, or the given timeout is reached.

This function will block for the specified amount of time at most and resume execution normally. Otherwise it will block indefinitely.

Returns Some(data) if data was successfully fetched, or None if the timeout was reached.

See also

• Bus::retrieve_all
• Bus::wait_for