Jetstream refactor

[Jarema]

Problem diagnosis

In current version of client (0.16.0) JetStream is exposed in given state:

no JetStream publish method

That means that the only way to publish to JetStream is by standard NATS publish method.
That, however, does not allow to:

• use deduplication as published messages are missing Nats-Msg-Id header
• Receive ACK (PubAck)
• set other JetStream related options (Domain, API prefix, etc)

Consumer:: methods have both JetStream and Consumer config opts

Because of that:

• there is no clear distinction between JetStream and Consumer options
• Consumers JetStream config have to be set up for each Consumer

Such implementation made Rust Client to drift quite far from reference Go implementation increasing cost of implementing new features available in Go client.

Solution

Follow Go Client approach and have a JetStream Context.
Go JetStreamContext is an interface aggregating all JetStream interactions and enables them to reuse the same base setup.
That means, that both JetStream publish and all Consumers can reuse the same context, which is especially useful when user want to switch a domain.

*Example usage (early DRAFT):

use nats::jetstream;

let nc = nats::connet("localhost");

// simple config with defaults
let mut js = jetstream::Context::new(nc);
js.publish("ORDERS", "data");
let consumer = js.pull("subject", "stream")
let msg = consumer.next();

// having some setup
let mut js = nats::jetstream::Options::new()
    .domain("HUB")
    .build();
   
let mut consumer = js.pull("ORDERS");

But how to pass options?

Below snippet shows possible solution for passing options wihtout using struct instances with ..Default::default()

use nats::jetstream;
let nc = nats::connet("localhost")?;

// simple config with defaults
let mut js = jetstream::Context::new(nc);


// having some setup
let mut js = nats::jetstream::Options::new()
    .domain("HUB")
    .build();

// I used dependency injection for JetStream Context here, as rest of the Rust library follows that approach.

// default JetStream Publish
// here dependency injection looks ok
jetstream:::publish(js, "events.asdf", "hello")?;

// publish with options
// here dependency injection looks ok
jetstream::PubOptions::new()
    .expect_stream("EVENTS")
    .expect_sequence(1)
    .publish(js, "events.asdf", "hello")?;

// ephemeral consumer
// here dependency injection doesn't look that nice...
let mut consumer = jetstream::Consumer::new(js);
let msg_data_len: usize = consumer.process(|msg| {
    println!("got message {:?}", msg);
    Ok(msg.data.len())
})?;

// durable consumer
// here dependency injection doesn't look that nice again.
// I don't fill good here with the implicit consumer type
// as it leads to some options being valid while others not.
// Would be good to be able to determine those at compile time.
// I'll go even further: let's make that a requirement.
let mut consumer = jetstream::ConsumerOptions::new(js)
    .durable("MONITOR")
    .max_deliver(3)
    .build();


// We can push the changes further and mimic Go client closer.
let mut consumer = js.PullSubscribeOptions::new()
    .bind_stream("HUB")
    // Rust API guidlines for Builders suggest specific names instead of generic (.build())
    // (though knowing the final method for each builder seems cumbersome, I admit)
    .create();

It rises some questions though:

• How to ensure correctness of options at compile time while maintaining close resemblence to Go client?
• do we prefer to have JetStream methods as part of JetStreamContext, or to pass the Context? (ugly?)

NOTE: Work in progress.

[Jarema]

NOTE: @caspervonb and @Jarema agreed that exposing everything through JetStreamContext as factories and operations will be the best approach.

[caspervonb]

Yeah I'm thinking more along the lines of:

let jetstream = client.jetstream(nats::jetstream::JetStreamOptions::new());
jetstream.publish_with_options("foo", nats::jetstream::PublishOptions::new()
    .expect_stream("EVENTS")
    .expect_sequence(1)
);

[Jarema]

For consumers we can go this way:

        let nc = crate::connect("demo,nats.io")?;
        let js = client.jetstream(nats::jetstream::JetStreamOptions::new());
        let ephemeral = js.subscribe("EVENTS.>");
       // we don't need separate functions for pull_consumer and push_consumer. We ca have either approach.
        let push_consumer = js.push_consumer(PushOpts::new().bind("HUB"));
        let pull_consumer = js.pull_consumer(PullOpts::new().bind("HUB"));

Each consumer exposes subscription functions available for given type of consumer.

[derekcollison]

We are trying hard to hide or at least make transparent the notion of delivery subject for push based consumers, its a impl detail.

[Jarema]

Ah, often using NATS-CLI for testing, I had a habit of always setting delivery subject explicitly in Go client.
After checking details, I see that it uses newInbox() random string if delivery == "" and !isPullMode.

Adapting the API above. Let's do the same in Rust client.

[Jarema]

Refactor has been implemented and merged.
Locking discussion.