Struct Junction 

pub struct Junction { /* private fields */ }

Struct managing the creation of new channels and Join Patterns.

This struct is used to group channels, such as SendChannel, which can be used in conjunction to create new Join Patterns. As such, it offers methods to create new channels which are then directly linked to this Junction. It also offers methods to start off the creation of new Join Patterns that rely on the channels created by this struct and can, in fact, only consist of channels associated with this struct.

Implementations

impl Junction

pub fn new() -> Junction

Create a new Junction and start control thread in background.

Create a new Junction and spawn a control thread in the background that will handle all incoming Packets for this Junction. A JoinHandle to this control thread is stored alongside the Junction.

Examples found in repository

examples/taxi-cab.rs (line 6)

fn main() {
    // Create a new Junction.
    let j = Junction::new();

    // Create new channels on the Junction j.
    let name = j.send_channel::<String>();
    let value = j.send_channel::<i32>();

    // Declare a new Join Pattern on the Junction using the channels above.
    j.when(&name).and(&value).then_do(|n, v| { println!("{} {}", n, v); });

    // Send all the required messages for the Join Pattern above to fire.
    value.send(1729).unwrap();
    name.send(String::from("Taxi")).unwrap();
}

examples/storage-cell.rs (line 8)

fn main() {
    /* Start of the Join Pattern setup. */

    // Declare a new Junction to create new channels and construct new
    // Join Patterns based on them.
    let cell = Junction::new();

    // New channel to retrieve the value of the storage cell.
    let get = cell.recv_channel::<i32>();

    // New channel to update the value of the storage cell.
    let put = cell.send_channel::<i32>();

    // New channel to swap the value of the storage cell for a new one and
    // retrieve the value that was just replaced.
    let swap = cell.bidir_channel::<i32, i32>();

    // New channel that will actually carry the value so that at no point
    // any given thread will have possession over it so concurrency issues
    // are avoided by design.
    let val = cell.send_channel::<i32>();

    // Set up some clones of the above channels we can move over to the
    // thread in which the function body of the Join Pattern will run.
    //
    // Clones of channels work like clones of the std::sync::mpsc::Sender
    // clones - any message sent from the clone will be received as if
    // sent from the original.
    let get_val = val.clone();
    let put_val = val.clone();
    let swap_val = val.clone();
    let val_val = val.clone();

    // Declare a new Join Pattern to update the storage cell value. If
    // both the put and val channel have sent a message, meaning someone
    // requested a value update and there is a value to be updated, send
    // a new val message through one of val's clones that carries the
    // updated value.
    cell.when(&put).and(&val).then_do(move |new, _old| {
        println!(">> put-val pattern fired with new={}!", new);
        put_val.send(new).unwrap();
    });

    // Declare a new Join Pattern to retrieve the storage cell value. If
    // both the get and val channel have sent a message, meaning someone
    // requested the value and there is a value to be given, return that
    // value and resend it through one of val's clones so that the value
    // is still available in future and not just consumed once.
    cell.when(&val).and_recv(&get).then_do(move |v| {
        println!(">> val-get pattern fired with v={}!", v);

        get_val.send(v.clone()).unwrap();

        v
    });

    // Declare a new Join Pattern to swap the storage cell value with a
    // new one and retrieve the old. Essentially works like a combination
    // of the previous two Join Patterns, with one crucial distinction:
    // with this Join Pattern, the update of the value and the retrieval
    // of the old are atomic, meaning that it is guaranteed that even in
    // a multithreaded environment with many users accessing the storage
    // cell, the value retrieved is exactly the value that has been
    // updated.
    cell.when(&val).and_bidir(&swap).then_do(move |old, new| {
        println!(
            ">> val-swap pattern fired with old={} and new={}!",
            old, new
        );
        swap_val.send(new).unwrap();

        old
    });

    // Declare a new Join Pattern that mentions the same channel multiple
    // times, so if the val channel has sent two messages they will be
    // combined into a single messages sent by a clone of val. This ensures
    // that eventually, the storage cell will only keep a single value
    // around.
    cell.when(&val).and(&val).then_do(move |a, b| {
        println!(">> val-val pattern fired with a={} and b={}!", a, b);
        val_val.send(a + b).unwrap();
    });

    /* End of the Join Pattern setup. */

    // Initialise the storage cell by sending an initial value that
    // can be picked up in future executions of the above Join Patterns.
    val.send(1729).unwrap();

    // Request a value update if one is available.
    put.send(42).unwrap();

    // Send another value that will eventually get combined with the
    // existing one.
    val.send(1).unwrap();

    // Request another value update.
    put.send(22).unwrap();

    // Request the current value of the storage cell and print it.
    println!("get.recv()={}", get.recv().unwrap());

    // Request a swap of the current value of the storage cell with a new
    // one and print the old value that is retrieved as a result.
    println!("swap.send_recv()={}", swap.send_recv(16).unwrap());

    // Request the current value of the storage cell again and print it.
    println!("get.recv()={}", get.recv().unwrap());
}

pub fn controller_handle(&mut self) -> Option<ControllerHandle>

Return handle to internal Controller if available.

Each Junction has an associated control thread with a Controller running to handle incoming Packets. This Controller is gracefully stopped and its thread joined as soon as the Junction goes out of scope. However, as this is sometimes undesired behavior, the user can retrieve the handle to the Junction’s Controller and its thread with this function and stop the Controller at a time of their choosing. Once this function has executed, the Junction will no long automatically stop its Controller and join the control thread upon going out of scope.

Note that this handle can only be retrieved once.

pub fn send_channel<T>(&self) -> SendChannel<T>

where T: Any + Send,

Create and return a new SendChannel on this Junction.

The generic parameter T is used to determine the type of values that can be sent on this channel.

Panics

Panics if it received an error while trying to receive a new channel ID from the control thread.

Examples found in repository

examples/taxi-cab.rs (line 9)

fn main() {
    // Create a new Junction.
    let j = Junction::new();

    // Create new channels on the Junction j.
    let name = j.send_channel::<String>();
    let value = j.send_channel::<i32>();

    // Declare a new Join Pattern on the Junction using the channels above.
    j.when(&name).and(&value).then_do(|n, v| { println!("{} {}", n, v); });

    // Send all the required messages for the Join Pattern above to fire.
    value.send(1729).unwrap();
    name.send(String::from("Taxi")).unwrap();
}

examples/storage-cell.rs (line 14)

fn main() {
    /* Start of the Join Pattern setup. */

    // Declare a new Junction to create new channels and construct new
    // Join Patterns based on them.
    let cell = Junction::new();

    // New channel to retrieve the value of the storage cell.
    let get = cell.recv_channel::<i32>();

    // New channel to update the value of the storage cell.
    let put = cell.send_channel::<i32>();

    // New channel to swap the value of the storage cell for a new one and
    // retrieve the value that was just replaced.
    let swap = cell.bidir_channel::<i32, i32>();

    // New channel that will actually carry the value so that at no point
    // any given thread will have possession over it so concurrency issues
    // are avoided by design.
    let val = cell.send_channel::<i32>();

    // Set up some clones of the above channels we can move over to the
    // thread in which the function body of the Join Pattern will run.
    //
    // Clones of channels work like clones of the std::sync::mpsc::Sender
    // clones - any message sent from the clone will be received as if
    // sent from the original.
    let get_val = val.clone();
    let put_val = val.clone();
    let swap_val = val.clone();
    let val_val = val.clone();

    // Declare a new Join Pattern to update the storage cell value. If
    // both the put and val channel have sent a message, meaning someone
    // requested a value update and there is a value to be updated, send
    // a new val message through one of val's clones that carries the
    // updated value.
    cell.when(&put).and(&val).then_do(move |new, _old| {
        println!(">> put-val pattern fired with new={}!", new);
        put_val.send(new).unwrap();
    });

    // Declare a new Join Pattern to retrieve the storage cell value. If
    // both the get and val channel have sent a message, meaning someone
    // requested the value and there is a value to be given, return that
    // value and resend it through one of val's clones so that the value
    // is still available in future and not just consumed once.
    cell.when(&val).and_recv(&get).then_do(move |v| {
        println!(">> val-get pattern fired with v={}!", v);

        get_val.send(v.clone()).unwrap();

        v
    });

    // Declare a new Join Pattern to swap the storage cell value with a
    // new one and retrieve the old. Essentially works like a combination
    // of the previous two Join Patterns, with one crucial distinction:
    // with this Join Pattern, the update of the value and the retrieval
    // of the old are atomic, meaning that it is guaranteed that even in
    // a multithreaded environment with many users accessing the storage
    // cell, the value retrieved is exactly the value that has been
    // updated.
    cell.when(&val).and_bidir(&swap).then_do(move |old, new| {
        println!(
            ">> val-swap pattern fired with old={} and new={}!",
            old, new
        );
        swap_val.send(new).unwrap();

        old
    });

    // Declare a new Join Pattern that mentions the same channel multiple
    // times, so if the val channel has sent two messages they will be
    // combined into a single messages sent by a clone of val. This ensures
    // that eventually, the storage cell will only keep a single value
    // around.
    cell.when(&val).and(&val).then_do(move |a, b| {
        println!(">> val-val pattern fired with a={} and b={}!", a, b);
        val_val.send(a + b).unwrap();
    });

    /* End of the Join Pattern setup. */

    // Initialise the storage cell by sending an initial value that
    // can be picked up in future executions of the above Join Patterns.
    val.send(1729).unwrap();

    // Request a value update if one is available.
    put.send(42).unwrap();

    // Send another value that will eventually get combined with the
    // existing one.
    val.send(1).unwrap();

    // Request another value update.
    put.send(22).unwrap();

    // Request the current value of the storage cell and print it.
    println!("get.recv()={}", get.recv().unwrap());

    // Request a swap of the current value of the storage cell with a new
    // one and print the old value that is retrieved as a result.
    println!("swap.send_recv()={}", swap.send_recv(16).unwrap());

    // Request the current value of the storage cell again and print it.
    println!("get.recv()={}", get.recv().unwrap());
}

pub fn recv_channel<R>(&self) -> RecvChannel<R>

where R: Any + Send,

Create and return a new RecvChannel on this Junction.

The generic parameter R is used to determine the type of values that can be received on this channel.

Panics

Panics if it received an error while trying to receive a new channel ID from the control thread.

Examples found in repository

examples/storage-cell.rs (line 11)

fn main() {
    /* Start of the Join Pattern setup. */

    // Declare a new Junction to create new channels and construct new
    // Join Patterns based on them.
    let cell = Junction::new();

    // New channel to retrieve the value of the storage cell.
    let get = cell.recv_channel::<i32>();

    // New channel to update the value of the storage cell.
    let put = cell.send_channel::<i32>();

    // New channel to swap the value of the storage cell for a new one and
    // retrieve the value that was just replaced.
    let swap = cell.bidir_channel::<i32, i32>();

    // New channel that will actually carry the value so that at no point
    // any given thread will have possession over it so concurrency issues
    // are avoided by design.
    let val = cell.send_channel::<i32>();

    // Set up some clones of the above channels we can move over to the
    // thread in which the function body of the Join Pattern will run.
    //
    // Clones of channels work like clones of the std::sync::mpsc::Sender
    // clones - any message sent from the clone will be received as if
    // sent from the original.
    let get_val = val.clone();
    let put_val = val.clone();
    let swap_val = val.clone();
    let val_val = val.clone();

    // Declare a new Join Pattern to update the storage cell value. If
    // both the put and val channel have sent a message, meaning someone
    // requested a value update and there is a value to be updated, send
    // a new val message through one of val's clones that carries the
    // updated value.
    cell.when(&put).and(&val).then_do(move |new, _old| {
        println!(">> put-val pattern fired with new={}!", new);
        put_val.send(new).unwrap();
    });

    // Declare a new Join Pattern to retrieve the storage cell value. If
    // both the get and val channel have sent a message, meaning someone
    // requested the value and there is a value to be given, return that
    // value and resend it through one of val's clones so that the value
    // is still available in future and not just consumed once.
    cell.when(&val).and_recv(&get).then_do(move |v| {
        println!(">> val-get pattern fired with v={}!", v);

        get_val.send(v.clone()).unwrap();

        v
    });

    // Declare a new Join Pattern to swap the storage cell value with a
    // new one and retrieve the old. Essentially works like a combination
    // of the previous two Join Patterns, with one crucial distinction:
    // with this Join Pattern, the update of the value and the retrieval
    // of the old are atomic, meaning that it is guaranteed that even in
    // a multithreaded environment with many users accessing the storage
    // cell, the value retrieved is exactly the value that has been
    // updated.
    cell.when(&val).and_bidir(&swap).then_do(move |old, new| {
        println!(
            ">> val-swap pattern fired with old={} and new={}!",
            old, new
        );
        swap_val.send(new).unwrap();

        old
    });

    // Declare a new Join Pattern that mentions the same channel multiple
    // times, so if the val channel has sent two messages they will be
    // combined into a single messages sent by a clone of val. This ensures
    // that eventually, the storage cell will only keep a single value
    // around.
    cell.when(&val).and(&val).then_do(move |a, b| {
        println!(">> val-val pattern fired with a={} and b={}!", a, b);
        val_val.send(a + b).unwrap();
    });

    /* End of the Join Pattern setup. */

    // Initialise the storage cell by sending an initial value that
    // can be picked up in future executions of the above Join Patterns.
    val.send(1729).unwrap();

    // Request a value update if one is available.
    put.send(42).unwrap();

    // Send another value that will eventually get combined with the
    // existing one.
    val.send(1).unwrap();

    // Request another value update.
    put.send(22).unwrap();

    // Request the current value of the storage cell and print it.
    println!("get.recv()={}", get.recv().unwrap());

    // Request a swap of the current value of the storage cell with a new
    // one and print the old value that is retrieved as a result.
    println!("swap.send_recv()={}", swap.send_recv(16).unwrap());

    // Request the current value of the storage cell again and print it.
    println!("get.recv()={}", get.recv().unwrap());
}

pub fn bidir_channel<T, R>(&self) -> BidirChannel<T, R>

where T: Any + Send, R: Any + Send,

Create and return a new BidirChannel on this Junction.

The generic parameter T is used to determine the type of values that can be sent on this channel while R is used to determine the type of values that can be received on this channel.

Panics

Panics if it received an error while trying to receive the new channel IDs from the control thread.

Examples found in repository

examples/storage-cell.rs (line 18)

fn main() {
    /* Start of the Join Pattern setup. */

    // Declare a new Junction to create new channels and construct new
    // Join Patterns based on them.
    let cell = Junction::new();

    // New channel to retrieve the value of the storage cell.
    let get = cell.recv_channel::<i32>();

    // New channel to update the value of the storage cell.
    let put = cell.send_channel::<i32>();

    // New channel to swap the value of the storage cell for a new one and
    // retrieve the value that was just replaced.
    let swap = cell.bidir_channel::<i32, i32>();

    // New channel that will actually carry the value so that at no point
    // any given thread will have possession over it so concurrency issues
    // are avoided by design.
    let val = cell.send_channel::<i32>();

    // Set up some clones of the above channels we can move over to the
    // thread in which the function body of the Join Pattern will run.
    //
    // Clones of channels work like clones of the std::sync::mpsc::Sender
    // clones - any message sent from the clone will be received as if
    // sent from the original.
    let get_val = val.clone();
    let put_val = val.clone();
    let swap_val = val.clone();
    let val_val = val.clone();

    // Declare a new Join Pattern to update the storage cell value. If
    // both the put and val channel have sent a message, meaning someone
    // requested a value update and there is a value to be updated, send
    // a new val message through one of val's clones that carries the
    // updated value.
    cell.when(&put).and(&val).then_do(move |new, _old| {
        println!(">> put-val pattern fired with new={}!", new);
        put_val.send(new).unwrap();
    });

    // Declare a new Join Pattern to retrieve the storage cell value. If
    // both the get and val channel have sent a message, meaning someone
    // requested the value and there is a value to be given, return that
    // value and resend it through one of val's clones so that the value
    // is still available in future and not just consumed once.
    cell.when(&val).and_recv(&get).then_do(move |v| {
        println!(">> val-get pattern fired with v={}!", v);

        get_val.send(v.clone()).unwrap();

        v
    });

    // Declare a new Join Pattern to swap the storage cell value with a
    // new one and retrieve the old. Essentially works like a combination
    // of the previous two Join Patterns, with one crucial distinction:
    // with this Join Pattern, the update of the value and the retrieval
    // of the old are atomic, meaning that it is guaranteed that even in
    // a multithreaded environment with many users accessing the storage
    // cell, the value retrieved is exactly the value that has been
    // updated.
    cell.when(&val).and_bidir(&swap).then_do(move |old, new| {
        println!(
            ">> val-swap pattern fired with old={} and new={}!",
            old, new
        );
        swap_val.send(new).unwrap();

        old
    });

    // Declare a new Join Pattern that mentions the same channel multiple
    // times, so if the val channel has sent two messages they will be
    // combined into a single messages sent by a clone of val. This ensures
    // that eventually, the storage cell will only keep a single value
    // around.
    cell.when(&val).and(&val).then_do(move |a, b| {
        println!(">> val-val pattern fired with a={} and b={}!", a, b);
        val_val.send(a + b).unwrap();
    });

    /* End of the Join Pattern setup. */

    // Initialise the storage cell by sending an initial value that
    // can be picked up in future executions of the above Join Patterns.
    val.send(1729).unwrap();

    // Request a value update if one is available.
    put.send(42).unwrap();

    // Send another value that will eventually get combined with the
    // existing one.
    val.send(1).unwrap();

    // Request another value update.
    put.send(22).unwrap();

    // Request the current value of the storage cell and print it.
    println!("get.recv()={}", get.recv().unwrap());

    // Request a swap of the current value of the storage cell with a new
    // one and print the old value that is retrieved as a result.
    println!("swap.send_recv()={}", swap.send_recv(16).unwrap());

    // Request the current value of the storage cell again and print it.
    println!("get.recv()={}", get.recv().unwrap());
}

pub fn when<T>(&self, send_channel: &SendChannel<T>) -> SendPartialPattern<T>

where T: Any + Send,

Create new partial Join Pattern starting with a SendChannel.

Panics

Panics if the supplied SendChannel does not carry the same JunctionID as this Junction, i.e. has not been created by and is associated with this Junction.

Examples found in repository

examples/taxi-cab.rs (line 13)

fn main() {
    // Create a new Junction.
    let j = Junction::new();

    // Create new channels on the Junction j.
    let name = j.send_channel::<String>();
    let value = j.send_channel::<i32>();

    // Declare a new Join Pattern on the Junction using the channels above.
    j.when(&name).and(&value).then_do(|n, v| { println!("{} {}", n, v); });

    // Send all the required messages for the Join Pattern above to fire.
    value.send(1729).unwrap();
    name.send(String::from("Taxi")).unwrap();
}

examples/storage-cell.rs (line 41)

fn main() {
    /* Start of the Join Pattern setup. */

    // Declare a new Junction to create new channels and construct new
    // Join Patterns based on them.
    let cell = Junction::new();

    // New channel to retrieve the value of the storage cell.
    let get = cell.recv_channel::<i32>();

    // New channel to update the value of the storage cell.
    let put = cell.send_channel::<i32>();

    // New channel to swap the value of the storage cell for a new one and
    // retrieve the value that was just replaced.
    let swap = cell.bidir_channel::<i32, i32>();

    // New channel that will actually carry the value so that at no point
    // any given thread will have possession over it so concurrency issues
    // are avoided by design.
    let val = cell.send_channel::<i32>();

    // Set up some clones of the above channels we can move over to the
    // thread in which the function body of the Join Pattern will run.
    //
    // Clones of channels work like clones of the std::sync::mpsc::Sender
    // clones - any message sent from the clone will be received as if
    // sent from the original.
    let get_val = val.clone();
    let put_val = val.clone();
    let swap_val = val.clone();
    let val_val = val.clone();

    // Declare a new Join Pattern to update the storage cell value. If
    // both the put and val channel have sent a message, meaning someone
    // requested a value update and there is a value to be updated, send
    // a new val message through one of val's clones that carries the
    // updated value.
    cell.when(&put).and(&val).then_do(move |new, _old| {
        println!(">> put-val pattern fired with new={}!", new);
        put_val.send(new).unwrap();
    });

    // Declare a new Join Pattern to retrieve the storage cell value. If
    // both the get and val channel have sent a message, meaning someone
    // requested the value and there is a value to be given, return that
    // value and resend it through one of val's clones so that the value
    // is still available in future and not just consumed once.
    cell.when(&val).and_recv(&get).then_do(move |v| {
        println!(">> val-get pattern fired with v={}!", v);

        get_val.send(v.clone()).unwrap();

        v
    });

    // Declare a new Join Pattern to swap the storage cell value with a
    // new one and retrieve the old. Essentially works like a combination
    // of the previous two Join Patterns, with one crucial distinction:
    // with this Join Pattern, the update of the value and the retrieval
    // of the old are atomic, meaning that it is guaranteed that even in
    // a multithreaded environment with many users accessing the storage
    // cell, the value retrieved is exactly the value that has been
    // updated.
    cell.when(&val).and_bidir(&swap).then_do(move |old, new| {
        println!(
            ">> val-swap pattern fired with old={} and new={}!",
            old, new
        );
        swap_val.send(new).unwrap();

        old
    });

    // Declare a new Join Pattern that mentions the same channel multiple
    // times, so if the val channel has sent two messages they will be
    // combined into a single messages sent by a clone of val. This ensures
    // that eventually, the storage cell will only keep a single value
    // around.
    cell.when(&val).and(&val).then_do(move |a, b| {
        println!(">> val-val pattern fired with a={} and b={}!", a, b);
        val_val.send(a + b).unwrap();
    });

    /* End of the Join Pattern setup. */

    // Initialise the storage cell by sending an initial value that
    // can be picked up in future executions of the above Join Patterns.
    val.send(1729).unwrap();

    // Request a value update if one is available.
    put.send(42).unwrap();

    // Send another value that will eventually get combined with the
    // existing one.
    val.send(1).unwrap();

    // Request another value update.
    put.send(22).unwrap();

    // Request the current value of the storage cell and print it.
    println!("get.recv()={}", get.recv().unwrap());

    // Request a swap of the current value of the storage cell with a new
    // one and print the old value that is retrieved as a result.
    println!("swap.send_recv()={}", swap.send_recv(16).unwrap());

    // Request the current value of the storage cell again and print it.
    println!("get.recv()={}", get.recv().unwrap());
}

pub fn when_recv<R>( &self, recv_channel: &RecvChannel<R>, ) -> RecvPartialPattern<R>

where R: Any + Send,

Create new partial Join Pattern starting with a RecvChannel.

Panics

Panics if the supplied RecvChannel does not carry the same JunctionID as this Junction, i.e. has not been created by and is associated with this Junction.

pub fn when_bidir<T, R>( &self, bidir_channel: &BidirChannel<T, R>, ) -> BidirPartialPattern<T, R>

where T: Any + Send, R: Any + Send,

Create a new partial Join Pattern starting with a BidirChannel.

Panics

Panics if the supplied BidirChannel does not carry the same JunctionID as this Junction, i.e. has not been created by and is associated with this Junction.

Trait Implementations

impl Drop for Junction

fn drop(&mut self)

Drop the Junction and free its resources.

If there is a ControllerHandle still available, use it to stop the associated Controller and join the control thread. Otherwise, no action is needed.