How to attach an mpsc channel to a Subscription?

[madmaxio]

I think of subscription as long running process (stream of values), so I create it, then it is passed to the runtime, and I don't need to create another one, and when the next command happens (subscription method is called) - I return Subscription::none. But right after this previous subscriptions are dropped?

EDIT: I'm creating channel let (tx, mut rx) = futures::channel::mpsc::unbounded() and I need to convert every message from rx to application message (update function call) - where should rx be located ?

[13r0ck]

The subscriptions you return are telling iced to check that subscription each cycle.

Iced checks the hash value of the subscription. If it changed then it produces a message for you app If the hash is the same, then no message is produced.

The subscription function is run every cycle, and what you return is what is checked for changes that cycle. So the async task can keep running, just iced stops checking for changes when you return subscription::none.

If you want to keep checking for changes then keep returning the subscription. If you have multiple subscriptions you need to keep alive then look into subscription::batch

[madmaxio]

But at the moment I need to return new instance of subscription every time, I can't return hash only for example to notify the runtime hash didn't change. But, I passed some state on the first creation of subscription, it is not an option for me to create this state every time subscription method is called.

[madmaxio]

If you want to keep checking for changes then keep returning the subscription

That's an option for me, but how can I return same subscription instance instead of creating new one?

[yusdacra]

If you want to keep checking for changes then keep returning the subscription

That's an option for me, but how can I return same subscription instance instead of creating new one?

You need to make sure that the hash function will return the same hash. So use something that will uniquely identify it; an ID, for something like a download request a URL etc.

[madmaxio]

This method fn subscription(&self) -> Subscription<Self::Message requires to return new instance of subscription. Even if i pass same hash I need to fill other fields in my recipe struct.

[yusdacra]

This method fn subscription(&self) -> Subscription<Self::Message requires to return new instance of subscription. Even if i pass same hash I need to fill other fields in my recipe struct.

You should not create your state beforehand. You need to pass the inputs to create your state that you can pass in every subscription call. Alternatively, using a Command instead of a Subscription could work, with a Message variant that you can use to "process" your state in update, and passing the state around via Command.

[madmaxio]

I'm trying to map a stream of messages from channel to application message:

pub struct Subscription {
	pub id: String,	
	pub rx: UnboundedReceiver<Data>
}

impl<H, E> Recipe<H, E> for Subscription where H: std::hash::Hasher {
    type Output = Message;

    fn hash(&self, state: &mut H) {
        use std::hash::Hash;

        std::any::TypeId::of::<Self>().hash(state);
        self.id.hash(state);
    }

    fn stream(self: Box<Self>, _input: Pin<Box<dyn Stream<Item = E> + 'static>>) -> Pin<Box<dyn Stream<Item = Self::Output> + 'static>> {		
        Box::pin(self.rx.then(move |data| process(data)).map(|res| Message::Data(res)))
    }
}

As you can see stream of data is created from rx.
Does it mean subscription is not a good fit for this task in general (mapping channel output stream to application message)?

[tarkah]

It seems like you are receiving some data, processing it and returning it to the application. Where is the sending side of this channel being used?

In general you should just return a Command::perform, where the future takes the data, processes and returns it back to the application. Subscriptions should generally be used for long lived, background tasks. Seems like we need some more context for exactly what you're trying to accomplish to better help you.

[madmaxio]

It seems like you are receiving some data, processing it and returning it to the application. Where is the sending side of this channel being used?

In general you should just return a Command::perform, where the future takes the data, processes and returns it back to the application. Subscriptions should generally be used for long lived, background tasks. Seems like we need some more context for exactly what you're trying to accomplish to better help you.

tx part is used in the event handler of one of the html elements via web-sys. The problem with returning rx receive as a future for command::perfrom is due to lifetime issue.

[madmaxio]

Ok, here is the direct question:

I'm creating channel let (tx, mut rx) = futures::channel::mpsc::unbounded() and I need to convert every message from rx to application message (update function call) - where should rx be located ?

[hecrj]

It should be created in Recipe::stream and moved inside the returned Stream. If you want bi-directional communication, you can send tx to the Application as the first message of the subscription.

[madmaxio]

This solution worked, thx a lot.

[Ciantic]

For what it's worth, I'm struggling with same problem: I have crossbeam_channel::Receiver<...> and how to push messages from there to app. I already found a one way, but it involves cloning the receiver on subscription each time and receiving just single message at a time.

[whooie]

@Ciantic I think I'm trying to do the same thing as you. What was your solution?

[hecrj]

The channel should be created in Recipe::stream and moved inside the returned Stream. If you want bi-directional communication, you can send tx to the Application as the first message of the subscription.