A go-cloud pubsub plugin for nats io supporting jetstream as well.

Package natspubsub provides a pubsub implementation for NATS.io and Jetstream. Use OpenTopic to construct a *pubsub.Topic, and/or OpenSubscription to construct a *pubsub.Subscription. This package uses gob to encode and decode driver.Message to []byte.

    // opening a topic looks like this
	package main 
    import (
        "context"
    
        "gocloud.dev/pubsub"
        _ "github.com/pitabwire/natspubsub"
    )
	...
	ctx := context.Background()
    // pubsub.OpenTopic creates a *pubsub.Topic from a URL.
    // This URL will Dial the NATS server at the URL in the environment variable
    // NATS_SERVER_URL and send messages with subject "example.mysubject".
    topic, err := pubsub.OpenTopic(ctx, "nats://localhost:4222?jetstream=true&stream_name=example&subject=example.mysubject")
    if err != nil {
        return err
    }
	defer topic.Shutdown(ctx)
	
    err := topic.Send(ctx, &pubsub.Message{
        Body: []byte("Hello, World!\n"),
        Metadata: map[string]string{
            "language":   "en",
            "importance": "high",
        },
    })
    if err != nil {
    return err
    }
    
    

package main
import (
    "context"

    "gocloud.dev/pubsub"
    _ "github.com/pitabwire/natspubsub"
)
...
ctx := context.Background()
subs, err := pubsub.OpenSubscription(ctx, "nats://localhost:4222?jetstream=true&stream_name=example&subject=example.mysubject")
if err != nil {
    return fmt.Errorf("could not open topic subscription: %v", err)
}

// Loop on received messages.
for {
    msg, err := subs.Receive(ctx)
    if err != nil {
        // Errors from Receive indicate that Receive will no longer succeed.
        log.Printf("Receiving message: %v", err)
        break
    }
    // Do work based on the message, for example:
    fmt.Printf("Got message: %q\n", msg.Body)
    // Messages must always be acknowledged with Ack.
    msg.Ack()
}

defer subs.Shutdown(ctx)

For pubsub.OpenTopic and pubsub.OpenSubscription, natspubsub registers for the scheme "nats".

The default URL opener will connect to a default server based on the environment variable "NATS_SERVER_URL".

This implementation supports all versions of NATS Servers, we however recommend version 2.2.0 or later. Messages can be encoded using native NATS message headers, and native message content providing full support for non-Go clients. Operating in this manner is more efficient than using gob.Encoder. Using older versions of the server will result in errors and untested scenarios.

In the event the user cannot upgrade the nats server, we recommend the use of the in tree implementation instead: https://github.com/google/go-cloud/tree/master/pubsub/natspubsub

To customize the URL opener, or for more details on the URL format, see URLOpener. See https://gocloud.dev/concepts/urls/ for background information.

Example URLs:

Topic open - nats://127.0.0.1:4222?jetstream=true&stream_name=testq&stream_retention=workqueue&stream_storage=memory&stream_subjects=testq.*&subject=testq

Topic subscription - nats://127.0.0.1:4222?consumer_ack_policy=explicit&consumer_ack_wait=10s&consumer_deliver_policy=all&consumer_durable_name=durabletest&consumer_filter_subject=testq.*&jetstream=true&stream_name=testq&stream_retention=workqueue&stream_storage=memory&stream_subjects=testq.*

This driver also has a hidden super power to make dynamic publishes from the header content supplied. One just needs to set the header that should be checked to extend the existing topic subject during publishing:

• header_to_extended_subject

NATS supports : See https://godoc.org/gocloud.dev/pubsub#hdr-At_most_once_and_At_least_once_Delivery for more background.

1. at-most-once-semantics; applications need not call Message.Ack, applications must also not call Message.Nack.

2. at-least-once-semantics; this mode is possible with jetstream enabled. applications are supposed to call Message.Ack to remove processed messages from the stream. Enabling this the jetstream mode requires adding the jetstream parameter in the url query.

All the subscription options are passed in as query parameters to the nats url supplied.

Comprehensive definitions of the options can be found below. These are based on the official NATS JetStream configuration from:

• NATS JetStream Stream Configuration
• NATS JetStream Consumer Configuration

Stream config options are prefixed by "stream_" in URL query parameters.

Consumer config options are prefixed by "consumer_" in URL query parameters.

Batch receive options control how messages are batched when receiving.

Batch ack options control how message acknowledgments are batched.