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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use std::{
  future::Future,
  time::Duration,
};

use ferrite_session::{
  either::*,
  prelude::*,
};
use tokio::time::sleep;

type Queue<A> = Rec<InternalChoice<Either<End, SendValue<A, Z>>>>;

type StringQueue = Queue<String>;

fn nil_queue<A>() -> Session<Queue<A>>
where
  A: Send + 'static,
{
  fix_session(offer_case(LeftLabel, terminate()))
}

fn append_queue<A, Func, Fut>(
  builder: Func,
  rest: Session<Queue<A>>,
) -> Session<Queue<A>>
where
  A: Send + 'static,
  Func: FnOnce() -> Fut + Send + 'static,
  Fut: Future<Output = A> + Send + 'static,
{
  fix_session(offer_case!(
    Right,
    step(async move { send_value(builder().await, rest) })
  ))
}

fn read_queue() -> Session<ReceiveChannel<StringQueue, End>>
{
  receive_channel(|queue| {
    unfix_session(
      queue,
      case! { queue ;
        Left => {
          wait ( queue, terminate () )
        }
        Right => {
          receive_value_from( queue,
            move |val| {
              println!("Receive value: {}", val);

              include_session (
                read_queue (),
                |next| {
                  send_channel_to (
                    next,
                    queue,
                    forward ( next )
                  ) })
            } )
        }
      },
    )
  })
}

pub fn queue_session() -> Session<End>
{
  let p11: Session<StringQueue> = nil_queue();

  let p12: Session<StringQueue> = append_queue(
    || async {
      println!("producing world..");

      sleep(Duration::from_secs(3)).await;

      "World".to_string()
    },
    p11,
  );

  let p13: Session<StringQueue> = append_queue(
    || async {
      println!("producing hello..");

      sleep(Duration::from_secs(2)).await;

      "Hello".to_string()
    },
    p12,
  );

  apply_channel(read_queue(), p13)
}

#[tokio::main]

pub async fn main()
{
  run_session(queue_session()).await;
}