Skip to content

Commit a9787b9

Browse files
authored
Merge pull request #33 from smol-rs/send-recv-types
Add `Send` and `Recv` futures
2 parents fba5d5b + 5fea266 commit a9787b9

File tree

1 file changed

+114
-31
lines changed

1 file changed

+114
-31
lines changed

src/lib.rs

Lines changed: 114 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -526,37 +526,10 @@ impl<T> Receiver<T> {
526526
/// assert_eq!(r.recv().await, Err(RecvError));
527527
/// # });
528528
/// ```
529-
pub async fn recv(&self) -> Result<T, RecvError> {
530-
let mut listener = None;
531-
532-
loop {
533-
// Attempt to receive a message.
534-
match self.try_recv() {
535-
Ok(msg) => {
536-
// If the capacity is larger than 1, notify another blocked receive operation.
537-
// There is no need to notify stream operations because all of them get
538-
// notified every time a message is sent into the channel.
539-
match self.channel.queue.capacity() {
540-
Some(1) => {}
541-
Some(_) | None => self.channel.recv_ops.notify(1),
542-
}
543-
return Ok(msg);
544-
}
545-
Err(TryRecvError::Closed) => return Err(RecvError),
546-
Err(TryRecvError::Empty) => {}
547-
}
548-
549-
// Receiving failed - now start listening for notifications or wait for one.
550-
match listener.take() {
551-
None => {
552-
// Start listening and then try receiving again.
553-
listener = Some(self.channel.recv_ops.listen());
554-
}
555-
Some(l) => {
556-
// Wait for a notification.
557-
l.await;
558-
}
559-
}
529+
pub fn recv(&self) -> Recv<'_, T> {
530+
Recv {
531+
receiver: self,
532+
listener: None,
560533
}
561534
}
562535

@@ -934,3 +907,113 @@ impl fmt::Display for TryRecvError {
934907
}
935908
}
936909
}
910+
911+
/// A future returned by [`Sender::send()`].
912+
#[derive(Debug)]
913+
#[must_use = "futures do nothing unless .awaited"]
914+
pub struct Send<'a, T> {
915+
sender: &'a Sender<T>,
916+
listener: Option<EventListener>,
917+
msg: Option<T>,
918+
}
919+
920+
impl<'a, T> Unpin for Send<'a, T> {}
921+
922+
impl<'a, T> Future for Send<'a, T> {
923+
type Output = Result<(), SendError<T>>;
924+
925+
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
926+
let mut this = Pin::new(self);
927+
928+
loop {
929+
let msg = this.msg.take().unwrap();
930+
// Attempt to send a message.
931+
match this.sender.try_send(msg) {
932+
Ok(()) => {
933+
// If the capacity is larger than 1, notify another blocked send operation.
934+
match this.sender.channel.queue.capacity() {
935+
Some(1) => {}
936+
Some(_) | None => this.sender.channel.send_ops.notify(1),
937+
}
938+
return Poll::Ready(Ok(()));
939+
}
940+
Err(TrySendError::Closed(msg)) => return Poll::Ready(Err(SendError(msg))),
941+
Err(TrySendError::Full(m)) => this.msg = Some(m),
942+
}
943+
944+
// Sending failed - now start listening for notifications or wait for one.
945+
match &mut this.listener {
946+
None => {
947+
// Start listening and then try receiving again.
948+
this.listener = Some(this.sender.channel.send_ops.listen());
949+
}
950+
Some(l) => {
951+
// Wait for a notification.
952+
match Pin::new(l).poll(cx) {
953+
Poll::Ready(_) => {
954+
this.listener = None;
955+
continue;
956+
}
957+
958+
Poll::Pending => return Poll::Pending,
959+
}
960+
}
961+
}
962+
}
963+
}
964+
}
965+
966+
/// A future returned by [`Receiver::recv()`].
967+
#[derive(Debug)]
968+
#[must_use = "futures do nothing unless .awaited"]
969+
pub struct Recv<'a, T> {
970+
receiver: &'a Receiver<T>,
971+
listener: Option<EventListener>,
972+
}
973+
974+
impl<'a, T> Unpin for Recv<'a, T> {}
975+
976+
impl<'a, T> Future for Recv<'a, T> {
977+
type Output = Result<T, RecvError>;
978+
979+
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
980+
let mut this = Pin::new(self);
981+
982+
loop {
983+
// Attempt to receive a message.
984+
match this.receiver.try_recv() {
985+
Ok(msg) => {
986+
// If the capacity is larger than 1, notify another blocked receive operation.
987+
// There is no need to notify stream operations because all of them get
988+
// notified every time a message is sent into the channel.
989+
match this.receiver.channel.queue.capacity() {
990+
Some(1) => {}
991+
Some(_) | None => this.receiver.channel.recv_ops.notify(1),
992+
}
993+
return Poll::Ready(Ok(msg));
994+
}
995+
Err(TryRecvError::Closed) => return Poll::Ready(Err(RecvError)),
996+
Err(TryRecvError::Empty) => {}
997+
}
998+
999+
// Receiving failed - now start listening for notifications or wait for one.
1000+
match &mut this.listener {
1001+
None => {
1002+
// Start listening and then try receiving again.
1003+
this.listener = Some(this.receiver.channel.recv_ops.listen());
1004+
}
1005+
Some(l) => {
1006+
// Wait for a notification.
1007+
match Pin::new(l).poll(cx) {
1008+
Poll::Ready(_) => {
1009+
this.listener = None;
1010+
continue;
1011+
}
1012+
1013+
Poll::Pending => return Poll::Pending,
1014+
}
1015+
}
1016+
}
1017+
}
1018+
}
1019+
}

0 commit comments

Comments
 (0)