lib: remove calls to .flush()

async-nats/src/client.rs

@@ -346,9 +346,6 @@ impl Client {
             }
             None => self.publish_with_reply(subject, inbox, payload).await?,
         }
-        self.flush()
-            .await
-            .map_err(|err| RequestError::with_source(RequestErrorKind::Other, err))?;
         let request = match timeout {
             Some(timeout) => {
                 tokio::time::timeout(timeout, sub.next())

async-nats/src/jetstream/consumer/pull.rs

@@ -153,11 +153,6 @@ impl Consumer<Config> {
             .publish_with_reply(subject, inbox, payload.into())
             .await
             .map_err(|err| BatchRequestError::with_source(BatchRequestErrorKind::Publish, err))?;
-        self.context
-            .client
-            .flush()
-            .await
-            .map_err(|err| BatchRequestError::with_source(BatchRequestErrorKind::Flush, err))?;
         debug!("batch request sent");
         Ok(())
     }
@@ -924,9 +919,6 @@ impl Stream {
                         .publish_with_reply(subject.clone(), inbox.clone(), request.clone())
                         .await
                         .map(|_| pending_reset);
-                    if let Err(err) = consumer.context.client.flush().await {
-                        debug!("flush failed: {err:?}");
-                    }
                     // TODO: add tracing instead of ignoring this.
                     request_result_tx
                         .send(result.map(|_| pending_reset).map_err(|err| {

async-nats/src/jetstream/consumer/push.rs

@@ -684,7 +684,6 @@ impl<'a> futures::Stream for Ordered<'a> {
                                                 .publish(subject, Bytes::from_static(b""))
                                                 .await
                                                 .ok();
-                                            client.flush().await.ok();
                                         });
                                     }
                                     continue;

async-nats/tests/kv_tests.rs

@@ -335,7 +335,7 @@ mod kv {
             .await
             .unwrap();
 
-        let context = async_nats::jetstream::new(client);
+        let context = async_nats::jetstream::new(client.clone());
 
         let kv = context
             .create_key_value(async_nats::jetstream::kv::Config {
@@ -352,6 +352,7 @@ mod kv {
         // check if we get only updated values. This should not pop up in watcher.
         kv.put("foo", 22.to_string().into()).await.unwrap();
         let mut watch = kv.watch("foo").await.unwrap().enumerate();
+        client.flush().await.unwrap();
 
         tokio::task::spawn({
             let kv = kv.clone();