nats-io / nats.rs Goto Github PK

I'm able to create a connection, subscribe to my topic (with a subscription handler) and see data flowing but when I go to close the app and clean up I'm getting an error.

First thing I'm doing is calling unsubscribe on my handler like this:

    match nsh.unsubscribe() {
        Ok(_ok) => {
            println!("NATS unsubscribe() successful!");
        }
        Err(e) => {
            println!("NATS unsubscribe() failed!!! {}", e);
        }
    };

I see the NATS unscribe() successful! message like I'm expecting.

Immediately after I unsubscribe, I close the connection like this:

    match nc.close() {
        Ok(_ok) => {
            println!("NATS connection close() successful!");
        }
        Err(e) => {
            println!("NATS connection close() failed!!! {}", e);
        }
    }

Before I see the NATS connection close() successful! message, I get the parsing error. Here is what that output looks like:

NATS unsubscribe() successful!
error encountered in the parser read_loop: Custom { kind: InvalidInput, error: "parsing error" }
NATS connection close() successful!

Is there something that I'm not doing correctly when cleaning up my connection and subscription?

In unit tests, we're using demo.nats.io, which is good for demo purposes, but I'm not sure about CI. The demo server has been flaky a couple of times before. How would you feel about perhaps using a NATS server running on localhost instead?

Hi,
I am very new to this nats client. I am trying to read a simple message via pub/sub. While compiling I get the error
method not found in nats::Channel

My code looks like,
pub fn subscribe_nats(_tn: &String, _nc: &Client,) -> Result<(), Box<dyn std::error::Error>> { let sub = _nc.subscribe(_tn, None)?; while let Some(msg) = sub.next() { let s = String::from_utf8(msg.data)?; println!("sub: {}", s); } Ok(()) }

How do I get the messages from the Channel - is that the right way to do - your examples do not show me any other method
any help is appreciated
thanks
Candy

Make sure that these boxes are checked before submitting your issue -- thank you!

• [ x ] Included below version and environment information
• [ x ] Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

0.8.1

rustc version (rustc --version - we support Rust 1.37 and up)

1.47

OS/Container environment:

macOS (Rust bin) / Docker (NATS)

After upgrading to 0.8.1, although Connection::publish returns Ok, not all messages make it to the NATS server. If I send several thousand messages, around 20 are missing every time, each time varying slightly. Even with sending 3 messages total, the server receives only 1.

I'm using JetStream, so to debug this I went into the JetStream container and ran nats --tlsca=/certs/ca.pem --tlscert=/certs/nats.pem --tlskey=/certs/nats-key.pem str info MYSTREAM:

Messages: 1
               Bytes: 429 B
            FirstSeq: 1 @ 2020-10-21T03:19:52 UTC
             LastSeq: 1 @ 2020-10-21T03:19:52 UTC
    Active Consumers: 1

This was after publishing 3 messages. I matched on the result of nc.publish and logged whether it was Ok or Err, and saw 3 Ok logs.

I then downgraded nats.rs to 0.7.4 and went through the same tests, and it doesn't have the same problem; all the numbers match up every time.

I couldn't test on 0.8.0 because every attempt to create a Connection just hangs forever.

As someone who has lost a cumulative total of 37 hours (yes, I've been keeping track) every time I've been stalled by a rust build chain that required the native-tls crate, which in turn requires the abomination known as openssl-sys, I would love for my use of the NATS client to not give its consumers this kind of headache.

Every time you build the native-tls crate in any kind of cross-platform scenario, you have to do so on a machine that has the open SSL source code/header libraries installed for the right target platform and OS and trust me when I say that even having this installed doesn't mean it'll work.

tl;dr - cross-platform/multi-platform build pipelines become insufferably difficult in the presence of anything that relies on openssl. The rustls crate does not have this problem.

Make sure that these boxes are checked before submitting your issue -- thank you!

• Included below version and environment information
• Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

natsio version (grep 'name = "natsio"' Cargo.lock -A 1)

0.4.0

rustc version (rustc --version - we support Rust 1.37 and up)

1.43.1

OS/Container environment:

Ubuntu

Steps or code to reproduce the issue:

If I try and store the connection options builder in a variable, for example, to start building and then optionally add authentication if it's available, I can't use the authenticated and unauthenticated builders as the same variable, nor can I use them as branch arms.

This code works, but requires me to repeat the connect call inside the branch arm:

fn get_connection() -> nats::Connection {
    if let Some(creds) = get_credsfile() {
        nats::ConnectionOptions::new()
            .with_credentials(creds)
            .connect(&get_env(LATTICE_HOST_KEY, DEFAULT_LATTICE_HOST))
    } else {
        nats::ConnectionOptions::new().connect(&get_env(LATTICE_HOST_KEY, DEFAULT_LATTICE_HOST))
    }
    .unwrap()
}

what I'd like to be able to do is this (not currently possible)

fn get_connection() -> nats::Connection {
 let o = nats::ConnectionOptions::new();
  if let Some(creds) = get_credsfile() {
     o = o.with_credentials(creds);
  }
  o.connect(...).unwrap()
}

There is an item in TODO about this and I think it is one of the most important ones in terms of library usability.
So I decided to create a tracking issue so that people can subscribe to and get notified when it is implemented.

What is the priority on it BTW?

Need to decide on which tech to use and whether or not we make this a feature flag.

Contenders are:

1. native-tls
2. openssl
3. rustls

Any thoughts and comments welcome.

Running on macOS, everything works, but on Windows we are getting a huge panic that starts with:

thread '<unnamed>' panicked at 'could not load system certs: (Some(RootCertStore { roots: [OwnedTrustAnchor { subject: [49, 30, 48, 28, 6,

and ends with:

Custom { kind: InvalidData, error: BadDER })4, 33, 98, 102, 67, 112, 214, 213, 192, 7, 225, 2, 3, 1, 0, 1], name_constraints: None }] }), Custom { kind: InvalidData, error: BadDER })4, 33, 98, 102, 67, 112, 214, 213, 192, 7, 225, 2, 3, 1, 0, 1], name_constraints: None }] }), Custom { kind: InvalidData, error: BadDER })', ', ', C:\Users\username\.cargo\registry\src\github.com-1ecc6299db9ec823\nats-0.8.3\src\asynk\connector.rsC:\Users\username\.cargo\registry\src\github.com-1ecc6299db9ec823\nats-0.8.3\src\asynk\connector.rsC:\Users\username\.cargo\registry\src\github.com-1ecc6299db9ec823\nats-0.8.3\src\asynk\connector.rs:::404040:::141414

Everything in between looks like binary data.

As a test, I set nats = "=0.8.2" in Cargo.toml and everything ran with no panics.

When nats client crash, and then it reconnect to nats server, and resubscribe topic.
My problem is, will the nats client lost message produced between crash time.
And how to use the durable feature for nats-streaming server?

Is there a way to build a stream over nats client subscription ?
I tried to build a stream using functions supplied for subscription like example below but I find that is very wasteful for pc resources.

impl Stream for NatsStream
{
    type Item = Message;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {

        let mut guard=self.state.lock().unwrap();

        match &guard.state
        {
            State::Ready=> {
                guard.state = State::Waiting;
                return Poll::Ready(guard.data.clone());
            },
            State::Waiting=> {
                guard.waker = Some(cx.waker().clone());
                return Poll::Pending;

            },
            _=> {
                guard.state = State::Pending;
                return Poll::Ready(None);
            }


        }

    }
}

Make sure that these boxes are checked before submitting your issue -- thank you!

• Included below version and environment information
• Included a Minimal, Complete, and Verifiable example

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

nats.rs commit 357cfb6
NATS Server version 2.2.1
NATS CLI version 0.0.22

rustc version (rustc --version - we support Rust 1.41 and up)

rustc 1.51.0 (2fd73fabe 2021-03-23)

OS/Container environment:

Manjaro 21.0.1 Ornara

Steps or code to reproduce the issue:

1. Start the NATS server (localhost)
2. Clone the nats.go library (b8530c78)
3. Modify the nats-sub example in the nats.go library to print headers:

   diff --git a/examples/nats-sub/main.go b/examples/nats-sub/main.go
   index fc3b91e..cd3f906 100644
   --- a/examples/nats-sub/main.go
   +++ b/examples/nats-sub/main.go
   @@ -14,6 +14,7 @@
    package main
    
    import (
   +	"bytes"
    	"flag"
    	"log"
    	"os"
   @@ -38,7 +39,9 @@ func showUsageAndExit(exitcode int) {
    }
    
    func printMsg(m *nats.Msg, i int) {
   -	log.Printf("[#%d] Received on [%s]: '%s'", i, m.Subject, string(m.Data))
   +	var hbuf bytes.Buffer
   +	m.Header.Write(&hbuf)
   +	log.Printf("[#%d] Received on [%s]: '%s', headers: %s", i, m.Subject, string(m.Data), hbuf.String())
    }
    
    func main() {
   

4. Run the nats.go nats-sub example and subscribe to a topic that you'll use (I'm using test):

   $ go run main.go -s localhost test
   Listening on [test]
   

5. Run the nats.rs nats-box example and subscribe to the same topic:

   $ cargo run --example nats-box -- --server localhost sub test
       Finished dev [unoptimized + debuginfo] target(s) in 0.08s
        Running `target/debug/examples/nats-box --server localhost sub test`
   Listening on 'test'
   

6. Run the following command:

   $ nats pub -H "Test: works" test "test"
   

   The nats-sub example from nats.go says:

   [#1] Received on [test]: 'test', headers: Test: works
   

   The nats-box example from nats.rs says:

   Received a Message { subject: "test", headers: Some(Headers { inner: {"Test": {"works"}} }), reply: None, length: 4 }
   

   Of course, that's expected behavior.
7. Now run the following command (reproduce the issue):

   $ nats pub -H "Test: works:test" test "test"
   

Expected result:

The nats-sub example from nats.go says:

[#2] Received on [test]: 'test', headers: Test: works:test

The expected result is that a header with a colon in its value is parsed by the nats.rs library.
Or, more specifically, this output was expected:

Received a Message { subject: "test", headers: Some(Headers { inner: {"Test": {"works:test"}} }), reply: None, length: 4 }

Actual result:

The nats-box example from nats.rs fails to parse the message and does not log having received the message.

Make sure that these boxes are checked before submitting your issue -- thank you!

• Included below version and environment information
• Included a Minimal, Complete, and Verifiable example

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

nats.rs commit 357cfb6
NATS Server version 2.2.1
NATS CLI version 0.0.22

rustc version (rustc --version - we support Rust 1.41 and up)

rustc 1.51.0 (2fd73fabe 2021-03-23)

OS/Container environment:

Manjaro 21.0.1 Ornara

Steps or code to reproduce the issue:

1. Start the NATS server (localhost)
2. Clone the nats.go library (b8530c78)
3. Modify the nats-sub example in the nats.go library to print headers:

   diff --git a/examples/nats-sub/main.go b/examples/nats-sub/main.go
   index fc3b91e..cd3f906 100644
   --- a/examples/nats-sub/main.go
   +++ b/examples/nats-sub/main.go
   @@ -14,6 +14,7 @@
    package main
    
    import (
   +	"bytes"
    	"flag"
    	"log"
    	"os"
   @@ -38,7 +39,9 @@ func showUsageAndExit(exitcode int) {
    }
    
    func printMsg(m *nats.Msg, i int) {
   -	log.Printf("[#%d] Received on [%s]: '%s'", i, m.Subject, string(m.Data))
   +	var hbuf bytes.Buffer
   +	m.Header.Write(&hbuf)
   +	log.Printf("[#%d] Received on [%s]: '%s', headers: %s", i, m.Subject, string(m.Data), hbuf.String())
    }
    
    func main() {
   

4. Run the nats.go nats-sub example and subscribe to a topic that you'll use (I'm using test):

   $ go run main.go -s localhost test
   Listening on [test]
   

5. Run the nats.rs nats-box example and subscribe to the same topic:

   $ cargo run --example nats-box -- --server localhost sub test
       Finished dev [unoptimized + debuginfo] target(s) in 0.08s
        Running `target/debug/examples/nats-box --server localhost sub test`
   Listening on 'test'
   

6. Run the following command:

   $ nats pub test "test"
   

   The nats-sub example from nats.go says:

   [#1] Received on [test]: 'test', headers: 
   

   The nats-box example from nats.rs says:

   Received a Message { subject: "test", headers: None, reply: None, length: 4 }
   

   Of course, that's expected behavior.
7. Run the following command:

   $ nats pub test ""
   

   The nats-sub example from nats.go says:

   [#2] Received on [test]: '', headers: 
   

   The nats-box example from nats.rs says:

   Received a Message { subject: "test", headers: None, reply: None, length: 0 }
   

   That's also expected behavior.
8. Run the following command:

   $ nats pub -H "Test: works" test "test"
   

   The nats-sub example from nats.go says:

   [#3] Received on [test]: 'test', headers: Test: works
   

   The nats-box example from nats.rs says:

   Received a Message { subject: "test", headers: Some(Headers { inner: {"Test": {"works"}} }), reply: None, length: 4 }
   

   That's also expected behavior.
9. Now run the following command (reproduce the issue):

   $ nats pub -H "Test: works" test ""
   

Expected result:

The nats-sub example from nats.go says:

[#4] Received on [test]: '', headers: Test: works

The expected result is that a message with headers and an empty body is parsed by the nats.rs library.
Or, more specifically, this output was expected:

Received a Message { subject: "test", headers: Some(Headers { inner: {"Test": {"works"}} }), reply: None, length: 0 }

Actual result:

The nats-box example from nats.rs fails to parse the message and does not log having received the message.

Use Case:

From the Headers RFC:

The server that is able to send and receive headers will specify so in it's INFO protocol message. The headers field if present, will have a boolean value.

One could use this information to find out whether the server supports headers, and send messages accordingly.

Proposed Change:

Add a headers boolean field to ServerInfo.

Who Benefits From The Change(s)?

Anyone looking to add support for headers in their codebase using this crate.

Related issue: #160

Alternative Approaches

N/A

I'm using smol and nats, and just migrated from rants to this official client since it now has direct async support.

With rants, I was creating a subscription stream and mapping messages to their deserialized data, then spawning a task to run through that stream in the background. After migrating, I'm getting this:

error[E0277]: `std::sync::MutexGuard<'_, std::collections::HashSet<u64>>` cannot be sent between threads safely
   --> detect/src/main.rs:72:5
    |
72  |     Task::spawn(async move {
    |     ^^^^^^^^^^^ `std::sync::MutexGuard<'_, std::collections::HashSet<u64>>` cannot be sent between threads safely
    |
   ::: /Users/andrewbanchich/.rustup/toolchains/stable-x86_64-apple-darwin/lib/rustlib/src/rust/src/libcore/future/mod.rs:55:43
    |
55  | pub const fn from_generator<T>(gen: T) -> impl Future<Output = T::Return>
    |                                           ------------------------------- within this `impl core::future::future::Future`
    |
   ::: /Users/andrewbanchich/.cargo/registry/src/github.com-1ecc6299db9ec823/async-executor-0.1.2/src/lib.rs:421:31
    |
421 |     pub fn spawn(future: impl Future<Output = T> + Send + 'static) -> Task<T>
    |                               ------------------ required by this bound in `async_executor::Task::<T>::spawn`

Make sure that these boxes are checked before submitting your issue -- thank you!

• [ x] Included below version and environment information
• [ x] Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

natsio version (grep 'name = "natsio"' Cargo.lock -A 1)

version 0.7

rustc version (rustc --version - we support Rust 1.37 and up)

1.45

OS/Container environment:

linux (fedora 31)

Steps or code to reproduce the issue:

use std::{thread, time::Duration};

fn read_loop(subscription: nats::Subscription) {
    println!("reading messages");
    for msg in subscription.messages() {
        println!("{:?}", msg);
    }
    println!("subscription closed");
}

fn main() {
    let nats_conn = nats::connect("127.0.0.1").unwrap();

    let mut subscription = nats_conn.subscribe("some-channel").unwrap();

    let s1 = subscription.clone();
    let thread_wait = thread::Builder::new().spawn(move || read_loop(s1)).unwrap();

    thread::sleep(Duration::from_millis(2000));
    println!("stopping subscription");
    subscription.drain().unwrap();

    println!("waiting for thread");
    thread_wait.join().unwrap();
    println!("done waiting");
}

Expected result:

Expected the program to shutdown within a few seconds.

Actual result:

Program hangs at subscription.drain() line. Can reproduce it in v0.6 and v0.7. Works in v0.5.

replacing subscription.drain() with subscription.close() also has the same issue.

We can port over the Go NATS client version of JS support.

• management logic implemented in #149
• consumers are still in-flux

Just noticed it's there for nats::Message but not nats::asynk::Message.

Make sure that these boxes are checked before submitting your issue -- thank you!

• Included below version and environment information
• Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

NATS version async-nats = "0.9"

rustc version rustc 1.49.0 (e1884a8e3 2020-12-29)

OS/Container environment: Ubuntu 20.04, x86_64

Steps or code to reproduce the issue:

use async_nats::Options;

#[tokio::main]
async fn main() {
	let connection = Options::new()
		.no_echo()
		.connect("nats://127.0.0.1:4222")
		.await
		.unwrap();	
	let subscription = connection.subscribe("test").await.unwrap();
	connection.publish("test", "no echo eh?").await.unwrap();
	assert_eq!(
		subscription.next().await.unwrap().data, 
		b"no echo eh?".to_vec()
	);
}

Expected result:

With a nats bus running at 127.0.0.1:4222 this code should hang forever as with no_echo set no messages should be received.

Actual result:

This code does not crash demonstrating a message with the same payload as the outgoing message is received despite the presence of no_echo.

We have some errors that are sync in nature, either from the NATS server or from threaded callbacks for message handlers.

Should we have the notion of last_error() for both NATS system and another for the message handlers that would hang off of SubscriptionHandler?

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

version = "0.8.1"

rustc version (rustc --version - we support Rust 1.37 and up)

rustc 1.46.0 (04488afe3 2020-08-24)

Steps or code to reproduce the issue:

#[tokio::test]
async fn test_drop_clone() {
    let nats1 = nats::asynk::connect("nats://stan:4222").await.unwrap();
    let nats2 = nats1.clone();
    nats1.publish("foo", b"bar").await.unwrap();
    nats2.publish("foo", b"bar").await.unwrap();
    drop(nats1);
    nats2.publish("foo", b"bar").await.unwrap();
}

Expected result:

Connection should not shutdown when a clone is dropped.

Actual result:

Because the Drop handler of nats1 is called shutdown is initiated but nats2 is still around.

Other network related crates I've seen, like reqwest, had a dedicated async module in the early days with everything else being sync, then switched it around so the default is async with a blocking module.

Any thoughts on making async the default for NATS?

Use Case:

As far as I understand Nats supports at least once delivery guarantee and allows to subscribe from a given offset sequence number. I can not find this capability in the rust client.

Proposed Change:

Add support for durable connections, at least once delivery guarantee and streaming from a given offset

Who Benefits From The Change(s)?

Rust applications

Alternative Approaches

Have not found, except switching to kafka

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

Crate version: 0.9.6

Server version: 2.2.0-beta.15 (it's the current synadia/jsm:latest image in docker hub, re-pulled it before trying to make sure)

rustc version (rustc --version - we support Rust 1.41 and up)

rustc 1.50.0 (cb75ad5db 2021-02-10)

OS/Container environment:

Windows 10

Steps or code to reproduce the issue:

Code (same as the other issue):
https://pastebin.com/1N0huWGd

Compile and execute it (twice, due to #155).

Open a shell to the container running NATS and execute:

~ # nats --server=localhost pub testing.testing "Test 1,2,3"
11:51:33 Published 10 bytes to "testing.testing"

Expected result:

The message is correctly received.

Actual result:

The program panics at the consumer.process line:

thread 'main' panicked at 'called `Option::unwrap()` on a `None` value', C:\Users\pod\.cargo\registry\src\github.com-1ecc6299db9ec823\nats-0.9.6\src\jetstream.rs:874:57
stack backtrace:
   0: std::panicking::begin_panic_handler
             at /rustc/cb75ad5db02783e8b0222fee363c5f63f7e2cf5b\/library\std\src\panicking.rs:493
   1: core::panicking::panic_fmt
             at /rustc/cb75ad5db02783e8b0222fee363c5f63f7e2cf5b\/library\core\src\panicking.rs:92
   2: core::panicking::panic
             at /rustc/cb75ad5db02783e8b0222fee363c5f63f7e2cf5b\/library\core\src\panicking.rs:50
   3: core::option::Option<nats::jetstream_types::JetStreamMessageInfo>::unwrap<nats::jetstream_types::JetStreamMessageInfo>
             at C:\Users\pod\.rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\core\src\option.rs:386
   4: nats::jetstream::Consumer::process<tuple<>,closure-0>
             at C:\Users\pod\.cargo\registry\src\github.com-1ecc6299db9ec823\nats-0.9.6\src\jetstream.rs:874
   5: jetstream_test::main
             at .\src\main.rs:52
   6: core::ops::function::FnOnce::call_once<fn(),tuple<>>
             at C:\Users\pod\.rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\core\src\ops\function.rs:227

The ed25519 crate along with signatory have made some very large, very breaking changes to their public APIs. Any crates that use this and also use NATS can no longer use NATS until the dependency is upgraded.

Make sure that these boxes are checked before submitting your issue -- thank you!

• Included below version and environment information
• Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

0.9.1

rustc version (rustc --version - we support Rust 1.41 and up)

1.48

OS/Container environment:

Linux (Pop Os 20.04)

Steps or code to reproduce the issue:

Run the code in this gist

Fix:

Removing upper case conversion here fixes the problem. Throughput is good after that.

Rename ConnectionOptions:

• Go and Java clients call it Options. I like that.
• Standard library has OpenOptions - note it's not named FileOptions because it's about the "open" action rather than the "file" result. In our case, the action is "connect" and the result of the action is "connection".
• Not a big deal, but I think ConnectOptions or Options would be a bit more idiomatic.

Don't re-export native-tls:

• Its API is not amazing to use and is not stable like Go's tls is.
• By re-exporting native-tls, we're bound to its versioning policy - if it bumps its major version number and we want to upgrade, then we have to bump our version number, too.
• Other crates typically provide a subset of its features.
• Check out how warp handles TLS with its TlsServer type. Here's an example.
• I suggest simply adding connect options like cert_path() and key_path() and we're done.

Don't export ServerInfo:

• It's just an internally used type.
• Other clients don't export it.

Make sure that these boxes are checked before submitting your issue -- thank you!

• Included below version and environment information
• Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

0.9.7

rustc version (rustc --version - we support Rust 1.41 and up)

1.5.0

OS/Container environment:

Docker 20.10.5

Steps or code to reproduce the issue:

• Run a nats server versioned below 2.2 (any version that doesn't support message headers).
• Subscribe to a subject.
• Use the publish_with_reply_or_headers() method to publish a message with headers.

Expected result:

An error is thrown explaining to the user that this version of the server doesn't support message headers.

Actual result:

The message is silently dropped (not published) without any error being thrown.

Use Case:

I want to track on client side, how many messages and/or bytes (but I mostly interested in message count) are pending. It's required for making a decision, when I need to add more consumers (some kind of manually written auto-scaling)

Proposed Change:

Add some kind of API for getting pending message count and pending bytes.

Who Benefits From The Change(s)?

At least me :) However, such feature will be usefil for any developer.

Alternative Approaches

I don't know.

I'm configuring NATS to use mTLS, and using add_root_certificate to pass in the ca.pem file that is just the public key of the CA.

The docs for add_root_certificate don't go into detail on what should be included in this file, and I don't see any equivalent to add client certs like the Go client has. I would have thought it would just be the root CA's public key, but the wording in the docs suggests it should include multiple keys.

Am I able to use mTLS with nats.rs?

I was playing with this library, with a subscriber and a publisher program adapted from the readme, and I noticed my program was exiting before the underlying messages were sent.

    let conn = nats::connect(&server)?;
    if let Some(matches) = matches.subcommand_matches("pub") {
        let topic = matches.value_of("topic").unwrap();
        let rest = matches.value_of("values").unwrap();
        conn.publish(topic, rest)?;
        conn.flush();   // <---- this fixed it
    };
    if let Some(matches) = matches.subcommand_matches("sub") {
        let topic = matches.value_of("topic").unwrap();
        let sub = conn.subscribe(topic)?;
        while let Some(ms) = sub.next() {
            let s = String::from_utf8(ms.data)?;
            println!("sub: {}", s);
        }
    };
// exit

Not sure the best way to present in the README, but it was a little paper cut for me, so I'm noting it here.

According to the changelog:

An experimental async Connection is now available to adventurous explorers by calling Options::connect_async.

Unfortunately, the asynk::Connection returned by connect_async is not usable because asynk is not public.

Should asynk be made public?

First of all, this (and similar code for close function):

The property .unsub is immediately flipped back.

Second, the only place where .unsub property was checked seems to be incorrect too:

I have a feeling that it should have been if !self.unsub, or better yet, don't use .unsub and name it .subscribed with the opposite meaning, then if self.subscribed would be easy and natural to read.

I would be able to make a PR, but I'm not 100% I understood it correctly.

P.S. I'd not mutate that property from the outside at all if possible, maybe consider moving different structs into different files, at least to a degree, to add more boundaries and make API cleaner.

Just suggestion for a small improvement in documentation:
Here:
https://docs.rs/nats/0.7.0/nats/struct.Options.html#method.max_reconnects

Set the maximum number of reconnect attempts. If no servers remain that are under this threshold, all servers will still be attempted.

There is no information what None mean. Infinite reconnect attempts or don't attempts at all (Some(0))?

Use Case:

At the moment the library takes care of reconnections, but the it can't connect to the server to begin with it just fails. Might be nice to have the library handle the initial connection retries by itself, depending on an option in Options.

Proposed Change:

Add an option to auto connect to Options and in the client run set use_backoff accordingly in

Who Benefits From The Change(s)?

No big deal, as we can always add a retry with timeout ourselves but it might avoid having to do it...

Alternative Approaches

User of the library adds it's own retry logic, should they wish to retry ofcourse.

It's really easy to make async NATS leak subscriptions. I could be mistaken, but I believe this is caused by an Arc reference cycle between the IO thread pool and user threads. When the user held Subscription goes out of scope, the Inner reference in the thread pool can remain there until the next message comes in. If the topic is inactive, it can remain there indefinately.

I'm not sure what would be the best way to fix this. I think internal references and user references should probably be separate things though, so when the last user reference is released, it can cancel the blocking operations and release the inner subscription.

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

0.9.16

rustc version (rustc --version - we support Rust 1.41 and up)

rustc 1.51.0

OS/Container environment:

Mac OS Catalina

Steps or code to reproduce the issue:

Here's a minimal example showcasing the issue:

use futures::StreamExt;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let conn = async_nats::connect("localhost").await.unwrap();
    let task_conn = conn.clone();

    let task = tokio::spawn(async move {
        let sub = task_conn.subscribe("test").await.unwrap();

        // create a stream of subscription messages
        let mut stream = futures::stream::unfold(sub, |sub| async move {
            sub.next().await.map(|result| (result, sub))
        })
        .boxed();

        // poll the stream once to activate it
        let _ = futures::poll!(stream.next());

        // the task has to yield at some point to reproduce the issue
        tokio::task::yield_now().await;

        // the stream holding the Subscription is dropped here
    });

    // wait for the task to finish
    task.await.unwrap();
    println!("Task exited.");

    // at this point the subscription remains active
    tokio::time::sleep(Duration::from_secs(600)).await;
    drop(conn);
}

It's possible to workaround this issue by wrapping the Subscription in a struct and explicitly calling unsubscribe on Drop. This unsubscribes from the topic regardless of how many subscription Inner references are still active.

Expected result:

You can verify the number of subscriptions in NATS metrics.

Number of subscriptions: 0

Actual result:

Number of subscriptions: 1

natsio version (grep 'name = "natsio"' Cargo.lock -A 1)

BTW, I think the above command needs to be updated to use "nats" instead of "natsio".

version = "0.6.0"

rustc version (rustc --version - we support Rust 1.37 and up)

rustc 1.44.1 (c7087fe00 2020-06-17)

OS/Container environment:

macOS Catalina 10.15.5

When I add nats to my project I get several errors:

error[E0252]: the name `AsyncWrite` is defined multiple times
 --> /Users/andrewbanchich/.cargo/registry/src/github.com-1ecc6299db9ec823/async-native-tls-0.3.3/src/runtime.rs:9:53
  |
2 | pub(crate) use async_std::io::{Read as AsyncRead, Write as AsyncWrite};
  |                                                   ------------------- previous import of the trait `AsyncWrite` here
...
9 | pub(crate) use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
  |                                                     ^^^^^^^^^^ `AsyncWrite` reimported here
  |
  = note: `AsyncWrite` must be defined only once in the type namespace of this module
help: you can use `as` to change the binding name of the import
  |
9 | pub(crate) use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite as OtherAsyncWrite};

error: only one of 'runtime-async-std' or 'runtime-tokio' features must be enabled
  --> /Users/andrewbanchich/.cargo/registry/src/github.com-1ecc6299db9ec823/async-native-tls-0.3.3/src/lib.rs:38:1
   |
38 | compile_error!("only one of 'runtime-async-std' or 'runtime-tokio' features must be enabled");

error[E0407]: method `poll_shutdown` is not a member of trait `AsyncWrite`
   --> /Users/andrewbanchich/.cargo/registry/src/github.com-1ecc6299db9ec823/async-native-tls-0.3.3/src/tls_stream.rs:116:5
    |
116 | /     fn poll_shutdown(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<io::Result<()>> {
117 | |         match self.with_context(ctx, |s| s.shutdown()) {
118 | |             Ok(()) => Poll::Ready(Ok(())),
119 | |             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Poll::Pending,
120 | |             Err(e) => Poll::Ready(Err(e)),
121 | |         }
122 | |     }
    | |_____^ not a member of trait `AsyncWrite`

This is only after adding nats as a dependency; I'm not using it in my code anywhere yet.

I'm currently using the smol runtime (which is awesome by the way).

Hey all;
I have recently updated Nats dependency from 0.7.4 to 0.8 and I am my system stopped working.

1. I am getting sys out output in the logs

POLL_FILL_BUF
[/home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-lite-1.11.0/src/io.rs:627] this.inner.as_mut().poll_read(cx, this.buf) = Ready(
    Ok(
        6,
    ),
)
>>> PING
 <<<
POLL_FILL_BUF

2. I am also getting error an error

thread '<unnamed>' panicked at 'called `Result::unwrap()` on an `Err` value: Utf8Error { valid_up_to: 93, error_len: Some(1) }', /home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-lite-1.11.0/src/io.rs:630:32
stack backtrace:
13:     0x561a6ef6e7f3 - core::option::expect_none_failed::hfcafefc70a7975bb
                               at src/libcore/option.rs:1272
  14:     0x561a6eb6784a - core::result::Result<T,E>::unwrap::h5f5b5aa469743cd0
                               at /home/dawid/.rustup/toolchains/stable-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libcore/result.rs:1005
  15:     0x561a6df3adb0 - <futures_lite::io::BufReader<R> as futures_io::if_std::AsyncBufRead>::poll_fill_buf::hdadca2e529b2a5bd
                               at /home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-lite-1.11.0/src/io.rs:630
  16:     0x561a6df3771e - <&mut T as futures_io::if_std::AsyncBufRead>::poll_fill_buf::he171da15e47565c9
                               at /home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-io-0.3.6/src/lib.rs:528
  17:     0x561a6df376ae - <&mut T as futures_io::if_std::AsyncBufRead>::poll_fill_buf::h72a7d46f2b2a5a47
                               at /home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-io-0.3.6/src/lib.rs:528
  18:     0x561a6df2ad5b - futures_lite::io::read_until_internal::hffcc16481c7243ee
                               at /home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-lite-1.11.0/src/io.rs:1578
  19:     0x561a6df3ce78 - <futures_lite::io::ReadUntilFuture<R> as core::future::future::Future>::poll::hc5a44bdc57b27932
                               at /home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-lite-1.11.0/src/io.rs:1565
  20:     0x561a6dee8088 - nats::asynk::proto::decode::{{closure}}::hb64363651553c3ca
                               at /home/dawid/.cargo/registry/src/github.com-1ecc6299db9ec823/nats-0.8.0/src/asynk/proto.rs:55

Hey, I'm trying to use your client but imagine my great disappointment after seeing:

pub fn request(&self, subject: &str, msg: impl AsRef<[u8]>) -> io::Result<Message> {
    future::block_on(self.0.request(subject, msg)).map(Message::from_async)
}

Could you add pub modifier to mod asynk?

It would be helpful for those of us using JetStream to have an example of how to use the existing NATS client for streaming messages.

I've been trying to benchmark NATS for my use case. Using the latency-tests tool I get something like this:

./latency -tr 1000 -tt 5s -sz 512
==============================
Pub Server RTT : 98µs
Sub Server RTT : 80µs
Message Payload: 512B
Target Duration: 5s
Target Msgs/Sec: 1000
Target Band/Sec: 1000K
==============================
HDR Percentiles:
10:       88µs
50:       108µs
75:       122µs
90:       137µs
99:       202µs
99.9:     253µs
99.99:    265µs
99.999:   982µs
99.9999:  982µs
99.99999: 982µs
100:      982µs
==============================

However, when using the Rust client and doing a very simple benchmark, my latency is ~2-5x of that. The following code is the simplest variation I have, but I tried various things: Reading and writing in a different thread, using tokio, etc, to replicate more closely what the Go benchmark does. I also tried putting explicit flush calls. The results are always the same. Am I doing something fundamentally wrong in the benchmark?

For reference, the rtt function gives the latency I expect, around 80 micros.

    let nc1 = nats::connect("localhost:4222").unwrap();
    let nc2 = nats::connect("localhost:4222").unwrap();
    let sub = nc2.subscribe("foo").unwrap();
    loop {
        let now = std::time::Instant::now();
        nc1.publish("foo", "").unwrap();
        sub.next().unwrap();
        println!("RTT: {:?}", now.elapsed());
        std::thread::sleep(std::time::Duration::from_millis(100));     
    }

RTT: 339.942µs
RTT: 337.414µs
RTT: 367.48µs
RTT: 357.508µs
RTT: 333.453µs
RTT: 392.305µs
RTT: 302.826µs
RTT: 495.053µs
RTT: 213.117µs
RTT: 187.48µs
RTT: 351.434µs
RTT: 419.223µs
RTT: 437.84µs
RTT: 376.603µs
RTT: 295.522µs
RTT: 303.573µs
RTT: 361.104µs
RTT: 306.155µs
RTT: 405.807µs
RTT: 313.317µs
RTT: 403.848µs
RTT: 315.56µs
RTT: 386.924µs
RTT: 379.003µs
RTT: 378.367µs
RTT: 331.52µs
RTT: 444.34µs
RTT: 354.675µs
RTT: 346.96µs
RTT: 482.608µs
RTT: 297.046µs
RTT: 214.371µs
RTT: 373.295µs
RTT: 422.137µs
RTT: 358.374µs
RTT: 374.481µs
RTT: 371.184µs

I have seen this and others reported as well. If you do not capture a subscriber from a call such as

nc.subscribe("help.request")?.with_handler(move |m| {
    m.respond("ans=42")?; Ok(())
});

The subscription will be dropped immediately and no messages will be received.

We could return an io::Result to have the compiler complain that the return result is not being handled, etc.

Or maybe look to have the subscription lifetime tied to the callback thread?

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

Crate version: 0.9.6

Server version: 2.2.0-beta.15 (it's the current synadia/jsm:latest image in docker hub, re-pulled it before trying to make sure)

rustc version (rustc --version - we support Rust 1.41 and up)

rustc 1.50.0 (cb75ad5db 2021-02-10)

OS/Container environment:

Windows 10

Steps or code to reproduce the issue:

Code:
https://pastebin.com/1N0huWGd

Run and execute it.

Expected result:

No panics

Actual result:

The first time the code runs it panics with:

thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Custom { kind: InvalidData, error: Error("data did not match any variant of untagged enum ApiResponse", line: 0, column: 0) }', src\main.rs:47:89
stack backtrace:
   0: std::panicking::begin_panic_handler
             at /rustc/cb75ad5db02783e8b0222fee363c5f63f7e2cf5b\/library\std\src\panicking.rs:493
   1: core::panicking::panic_fmt
             at /rustc/cb75ad5db02783e8b0222fee363c5f63f7e2cf5b\/library\core\src\panicking.rs:92
   2: core::option::expect_none_failed
             at /rustc/cb75ad5db02783e8b0222fee363c5f63f7e2cf5b\/library\core\src\option.rs:1268
   3: core::result::Result<nats::jetstream::Consumer, std::io::error::Error>::unwrap<nats::jetstream::Consumer,std::io::error::Error>
             at C:\Users\pod\.rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\core\src\result.rs:973
   4: jetstream_test::main
             at .\src\main.rs:47

The panic occurs at the line with the Consumer::create_or_open call.

However, I can see that the stream and the consumer are created

~ # nats --server=localhost stream ls
Streams:

        testing

~ # nats --server=localhost stream info testing
Information for Stream testing

Configuration:

             Subjects: testing.*
     Acknowledgements: true
            Retention: File - Limits
             Replicas: 1
       Discard Policy: Old
     Duplicate Window: 0s
     Maximum Messages: unlimited
        Maximum Bytes: unlimited
          Maximum Age: 0s
 Maximum Message Size: unlimited
    Maximum Consumers: unlimited

State:

            Messages: 0
               Bytes: 0 B
            FirstSeq: 0
             LastSeq: 0
    Active Consumers: 1

(Note that I expected 3600s, or at least 3,6s if it is interpreted in ms, for the Duplicate Window setting)

~ # nats --server=localhost consumer info
? Select a Stream testing
? Select a Consumer testing_1_durable
Information for Consumer testing > testing_1_durable

Configuration:

        Durable Name: testing_1_durable
           Pull Mode: true
      Filter Subject: testing.testing
         Deliver All: true
          Ack Policy: Explicit
            Ack Wait: 2µs
       Replay Policy: Instant

State:

  Last Delivered Message: Consumer sequence: 0 Stream sequence: 0
    Acknowledgment floor: Consumer sequence: 0 Stream sequence: 0
        Pending Messages: 0
    Redelivered Messages: 0

The second, and subsequent times I run the same code, it works. I can see the "Listening" printf.

I noticed not all features of nats are there for async-nats (request_timeout for example).

NATS version (grep 'name = "nats"' Cargo.lock -A 1)

name = "nats"
version = "0.8.6"

rustc version (rustc --version - we support Rust 1.37 and up)

rustc 1.49.0 (e1884a8e3 2020-12-29)

OS/Container environment:

Pop OS (Debian Linux)

Steps or code to reproduce the issue:

Create a static NATS Connection using once_cell::sync::Lazy:

pub static NATS: Lazy<nats::asynk::Connection> = Lazy::new(|| {
    let nats_host = std::env::var("NATS_HOST").unwrap();

    let nc = nats::Options::new()
        .add_root_certificate(format!("{}/ca.pem", *CERT_DIR))
        .connect_async(&nats_host);

    smol::block_on(nc).expect("Couldn't connect to NATS")
});

In a stream, publish messages using the static NATS Connection and flush:

\\ inside stream.for_each

if let Err(e) = NATS.publish(&sub, &serde_json::to_vec(&foo).unwrap()).await {
    error!("Couldn't publish message: {:#?}", e);
}

NATS.flush().await.unwrap();

Some messages are not published.

Switch to create new NATS Connection instead of using static:

\\ inside stream.for_each

let nc = nats::Options::new()
    .add_root_certificate(format!("{}/ca.pem", std::env::var("CERT_DIR").unwrap()))
    .connect_async(&std::env::var("NATS_HOST").unwrap())
    .await
    .expect("Couldn't connect to NATS");

if let Err(e) = nc.publish(&sub, &serde_json::to_vec(&foo).unwrap()).await {
    error!("Couldn't publish message: {:#?}", e);
}

nc.flush().await.unwrap();

All messages are now published.

Expected result:

I'd like to share a static NATS Connection across my program and have it send all messages.

I have a trouble while using the jetstream::Consumer::pull method.

The doc said this method is only usable with pull-based consumers, which means ConsumerConfig.deliver_subject must be None.

However, if I try to use it with pull-based consumers, it complains the deliver_subject should not be None.

Maybe a bug?

@stjepang Forgive me in advance :) but I was wondering if there is any easy way to remove Smol dependency such that I can only use one executor in my app?
My application relies heavily on Tokio and Tokio ecosystem so naturally, I would prefer to keep it that way.

The rationale for my request is that when I was trying to convert my app to use Async NATS, I have run into some locking problems when I was trying to pass messages from tasks running in Smal to tasks running in Tokio via channels.

I must admit, I quickly revert back to the blocking code, but it would be nice if everything was executed asynchronously in one executor.

I'm unable to connect to NATS v2.1.0 with this client. NATS v2.1.4 works fine. It just so happened that I had v2.1.0 installed on my production server, and v2.1.4 on my local machine.

The error messages I get is:

 missing field `server_name` at line 1 column 186

And indeed, telnet confirms that this field is missing. The dotnet client doesn't have this issue, so that makes me thing the client is being overly strict. But I'm open to calling it an already-fixed in the server. I will be definitely updating my server in either case.