1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

use crate::{
    pipe::{HandshakeInit, PipeBehavior, PipeReceiver},
    protocols::pipe::internal::{Handshake, InternalCmd},
    Context,
};
use ockam_core::compat::boxed::Box;
use ockam_core::{Address, Result, Routed, Worker};

/// Listen for pipe handshakes and creates PipeReceive workers
pub struct PipeListener {
    hooks: PipeBehavior,
}

#[crate::worker]
impl Worker for PipeListener {
    type Context = Context;
    type Message = InternalCmd;

    async fn initialize(&mut self, ctx: &mut Context) -> Result<()> {
        ctx.set_cluster(crate::pipe::CLUSTER_NAME).await
    }

    async fn handle_message(&mut self, ctx: &mut Context, msg: Routed<InternalCmd>) -> Result<()> {
        let route_to_sender = msg.return_route();
        match msg.body() {
            InternalCmd::InitHandshake => {
                info!("Creating new PipeReceiver for incoming handshake");

                // Create a new pipe receiver with a modified behavioral stack
                let recv_addr = Address::random_local();
                let int_addr = Address::random_local();
                let hooks = self.hooks.clone().attach(HandshakeInit::default());
                PipeReceiver::create(ctx, recv_addr.clone(), int_addr.clone(), hooks).await?;

                // Then send it the handshake message
                ctx.send(
                    vec![int_addr],
                    InternalCmd::Handshake(Handshake { route_to_sender }),
                )
                .await?;
            }
            cmd => debug!("Ignoring invalid cmd: {:?}", cmd),
        }

        Ok(())
    }
}

impl PipeListener {
    /// Create pipe creation listener with explicit behavior hooks
    pub async fn create_with_behavior(
        ctx: &mut Context,
        addr: Address,
        hooks: PipeBehavior,
    ) -> Result<()> {
        ctx.start_worker(addr, PipeListener { hooks }).await?;
        Ok(())
    }

    /// Create pipe creation listener with empty behavior hooks
    pub async fn create(ctx: &mut Context, addr: Address) -> Result<()> {
        Self::create_with_behavior(ctx, addr, PipeBehavior::empty()).await
    }
}