Implement basic combinators map, filter and filter_map for irpc channels. #68
Conversation
…and oneshot::Sender Implement mao, filter_map and filter for mpsc::Receiver oneshot::Receiver is a future, so you can just do this yourself using FutureExt.
this makes irpc::channel::mpsc::* a generic channel that can also be used for non-serializable msgs. Especially when mapping you might not want intermediate map results ot be serializable!
…erializzable msgs.
it was not exported anyway, so the resulting future was anonymous all th etime.
…an also be mapped.
|
Note: an in memory stream can return the following errors: pub enum SendError {
ReceiverClosed,
...
}An io based stream can in addition produce the following errors: ...
MaxMessageSizeExceeded,
Io(io::Error),
}We could model this by having a type parameter for streams that tracks if the stream uses io or not, but I am not sure if it is worth it. In any case I think it can wait for another PR. |
|
Actually I think having recv return a Result<Option, RecvError> is wrong. We got the case RecvError::SenderClosed in 2 cases now - as error and as returning None. We should modify this so that RecvError for mpsc has just the true error cases, since stream closed is already modeled as returning None. |
src/lib.rs
Outdated
| /// Applies a filter before sending. | ||
| /// | ||
| /// Messages that don't pass the filter are dropped. | ||
| pub fn with_filter(self, f: impl Fn(&T) -> bool + Send + Sync + 'static) -> Sender<T> |
There was a problem hiding this comment.
not super convinced that these are useful on the oneshot sender? but maybe they are :)
There was a problem hiding this comment.
🤷 just did it for completeness.
src/lib.rs
Outdated
| where | ||
| F: Fn(U) -> Option<T> + Send + Sync + 'static, | ||
| U: Send + Sync + 'static, | ||
| T: Send + Sync + 'static, |
There was a problem hiding this comment.
We could move those bounds to the top level (i.e. have T: Send + Sync + 'static on the top level impl block). But don't mind much either way.
There was a problem hiding this comment.
We could do another "trait alias" trick where you have
trait SendSyncStatic: Send + Sync + 'static {}
impl<T: Send + Sync + 'static> SendSyncStatic for T;
But these bounds are fine even for wasm, so probably best to live with the repetition I think.
src/lib.rs
Outdated
| pub fn map<U, F>(self, f: F) -> Receiver<U> | ||
| where | ||
| F: Fn(T) -> U + Send + Sync + 'static, | ||
| T: Send + Sync + 'static, |
There was a problem hiding this comment.
can drop Send + 'static bounds, are already on top level impl block
| fn recv( | ||
| &mut self, | ||
| ) -> Pin< | ||
| Box< |
There was a problem hiding this comment.
nit: that's a long type, maybe we add a BoxedSyncFuture type alias and use in all DynReceiver uses? But this is unrelated to this PR so can also come sometime later.
There was a problem hiding this comment.
We could use n0_future::boxed::BoxFuture if it weren't for the + '_. So maybe n0_future could contain such a type alias.
Frando
left a comment
Frando
left a comment
There was a problem hiding this comment.
Looks good! Some nits in comments.
I find these useful quite a few times. They already exist on Sink and Stream of course, but often you don't want to convert your senders and receivers into Sink / Stream and lose the extra fns. Also Sink API is just weird!
In the fns that filter (with_filter, with_filter_map, filter, filter_map) the item is just dropped if it does not match the filter. I think for these channels this is uncontroversial, besides SinkExt does the same. It does not produce an error if you use e.g. with_flat_map to map to an empty stream.
An additional change is that the bounds for crate::channel::mpsc::Sender/Receiver are reduced so you can now use a crate::channel::mpsc::Sender/Receiver as the channel to an actor even for messages that are not serializable.