Modelling a "processing pipeline:

[disconsented]

Hi, it's me again 👋 this is largely a follow-up of the issue I opened a little while ago. I'm having a bit of trouble modelling behaviour here.

In classic CSP, this is all straightforward, just use channels to pass data around. But, par's a little more tricky.

I'm designing a “pipeline” for processing data, I want it to work something like:

pub fn peusdocode_pipeline(self){
   while let Some(message) = self.incoming.next(){
       let mut continue = true;
       let (message, shouldContinue, widget) = self.widget_a.send(message).recv();
       continue |= shouldContinue;
       self.widget_a = widget;
       let (message, shouldContinue, widget) = self.widget_b.send(message).recv();
       continue |= shouldContinue;
       self.widget_b = widget;
       // ... etc
      if continue{
         self.something_else(message);
      }
   }
}

pub enum Data {
    Data(Recv<(Message, Send<(bool, Message)>), Recv<Data>>),
}
pub fn start(mut self) -> Send<Data> {
    fork(|client: Recv<Data>| async move {
      // Do stuff forever.
    })
}

I'm finding two parts tricky:
Lifting out of Par isn't obvious, I've not yet discovered a way to do this without something like a channel. How should I do this?

And, how do I use Session::link to chain together things, when, I want to modify T before?

[faiface]

@disconsented Hey, thanks for reaching out! I'm not sure I understand your question.

Yes, it is true that with Par, some things have to be done differently, but I'm not sure which part you're finding trouble with.

To your first question, why do you want to lift out of Par? I don't mean you shouldn't do it, I'm just trying to understand your purpose, so I can help.

To your second question, what's the T in question? Any example for this?

[disconsented]

To your first question, why do you want to lift out of Par? I don't mean you shouldn't do it, I'm just trying to understand your purpose, so I can help.

Off the top of my head, two use cases:

1. An escape hatch for when I get stuck on something and want to return to it later.
2. At the end of the process, handing off the result to something else. In this case, it'd be served up to the front end.

To your second question, what's the T in question? Any example for this?

For an example, (bool, Message), based upon the bool, I may wish to do any number of things, including, branch, discard, or return/escape. In my specific use case, it's used to mark whether to forward the message to the front end or to discard it.

Alternatively, maybe to make use of typestates.

I guess, really, some kind of map operation like map<U, F>(self, f: F) -> U.

[faiface]

@disconsented

For such lifting out of Par, yes, a channel is a good idea. The futures create has a good implementation of async/await channels, so I'd recommend that.

I'm still not a 100% sure I understand your second question, but is it maybe that you want the communication to proceed differently depending on if you got true or false? If that's the case, you should use branching, and make the protocol something like this:

enum Message {
  True(Recv<PayloadForTrue, RestOfProtocolForTrue>),
  False(Recv<PayloadForFalse, RestOfProtocolForFalse>),
}

Of course, change True/False to whatever makes sense there.

To understand this pattern better, check out the branching section of the docs: https://docs.rs/par/latest/par/#branching

But if I got you wrong, please clarify :)