ensure consumers get a unique ephemeral name

Author: ayazabbas

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-stream"
-version = "0.1.6"
+version = "0.1.7"
 edition = "2021"
 
 [lib]

Cargo.toml

@@ -268,10 +268,10 @@ struct PriceInfo {
 
 async fn handle_nats_messages(jetstream: jetstream::Context, clients: Clients) -> Result<()> {
     let stream_name = "PYTH_PRICE_UPDATES";
-    let consumer_name = "websocket_server";
 
     let consumer_config = consumer::pull::Config {
-        durable_name: Some(consumer_name.to_string()),
+        deliver_policy: consumer::DeliverPolicy::All,
+        ack_policy: consumer::AckPolicy::None,
         ..Default::default()
     };
 
@@ -280,7 +280,7 @@ async fn handle_nats_messages(jetstream: jetstream::Context, clients: Clients) -
         .await
         .context("Failed to create NATS consumer")?;
 
-    info!(stream = %stream_name, consumer = %consumer_name, "Started handling NATS messages");
+    info!(stream = %stream_name, "Started handling NATS messages");
 
     loop {
         let mut messages = consumer
@@ -336,13 +336,6 @@ async fn handle_nats_messages(jetstream: jetstream::Context, clients: Clients) -
                             );
                         }
                     }
-
-                    // Spawn a new task for acknowledgment
-                    tokio::spawn(async move {
-                        if let Err(e) = msg.ack().await {
-                            warn!(error = %e, "Failed to acknowledge NATS message");
-                        }
-                    });
                 }
                 Err(e) => {
                     error!(error = %e, "Error receiving message from NATS");

src/bin/websocket_server.rs

@@ -9,7 +9,7 @@ pub async fn setup_jetstream(nats_client: &async_nats::Client) -> Result<jetstre
     let stream_config = stream::Config {
         name: "PYTH_PRICE_UPDATES".to_string(),
         subjects: vec!["pyth.price.updates".to_string()],
-        max_bytes: 1024 * 1024 * 4000,
+        max_bytes: 1024 * 1024 * 1000,
         duplicate_window: Duration::from_secs(60),
         discard: stream::DiscardPolicy::Old,
         allow_direct: true,