Use idiomatic method for writing Option and accessing inner T

nats-io · Jul 17, 2023 · 326a845 · 326a845
1 parent fa6a885
commit 326a845
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 12 deletions.
diff --git a/async-nats/src/jetstream/consumer/pull.rs b/async-nats/src/jetstream/consumer/pull.rs
@@ -460,7 +460,7 @@ impl<'a> futures::Stream for Sequence<'a> {
                 let request = self.request.clone();
                 let pending_messages = self.pending_messages;
 
-                self.next = Some(Box::pin(async move {
+                let next = self.next.insert(Box::pin(async move {
                     let inbox = context.client.new_inbox();
                     let subscriber = context
                         .client
@@ -484,7 +484,7 @@ impl<'a> futures::Stream for Sequence<'a> {
                     })
                 }));
 
-                match self.next.as_mut().unwrap().as_mut().poll(cx) {
+                match next.as_mut().poll(cx) {
                     Poll::Ready(result) => {
                         self.next = None;
                         Poll::Ready(Some(result.map_err(|err| {

diff --git a/async-nats/src/jetstream/consumer/push.rs b/async-nats/src/jetstream/consumer/push.rs
@@ -578,16 +578,17 @@ impl<'a> futures::Stream for Ordered<'a> {
                         let sequence = self.stream_sequence.clone();
                         let config = self.consumer.config.clone();
                         let stream_name = self.consumer.info.stream_name.clone();
-                        self.subscriber_future = Some(Box::pin(async move {
-                            recreate_consumer_and_subscription(
-                                context,
-                                config,
-                                stream_name,
-                                sequence.load(Ordering::Relaxed),
-                            )
-                            .await
-                        }));
-                        match self.subscriber_future.as_mut().unwrap().as_mut().poll(cx) {
+                        let subscriber_future =
+                            self.subscriber_future.insert(Box::pin(async move {
+                                recreate_consumer_and_subscription(
+                                    context,
+                                    config,
+                                    stream_name,
+                                    sequence.load(Ordering::Relaxed),
+                                )
+                                .await
+                            }));
+                        match subscriber_future.as_mut().poll(cx) {
                             Poll::Ready(subscriber) => {
                                 self.subscriber_future = None;
                                 self.subscriber = Some(subscriber.map_err(|err| {