Documentation
¶
Overview ¶
Package natspubsub provides a pubsub implementation for NATS.io. 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.
URLs ¶
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". To customize the URL opener, or for more details on the URL format, see URLOpener. See https://github.com/eliben/gocdkx/concepts/urls/ for background information.
Message Delivery Semantics ¶
NATS supports at-most-semantics; applications need not call Message.Ack, and must not call Message.Nack. See https://godoc.org/github.com/eliben/gocdkx/pubsub#hdr-At_most_once_and_At_least_once_Delivery for more background.
As ¶
natspubsub exposes the following types for As:
- Topic: *nats.Conn
- Subscription: *nats.Subscription
- Message.BeforeSend: None.
- Message: *nats.Msg
Example (OpenSubscription) ¶
package main
import (
"context"
"log"
"github.com/eliben/gocdkx/pubsub"
)
func main() {
// This example is used in https://github.com/eliben/gocdkx/howto/pubsub/subscribe/#nats
// import _ "github.com/eliben/gocdkx/pubsub/natspubsub"
// Variables set up elsewhere:
ctx := context.Background()
// OpenSubscription creates a *pubsub.Subscription from a URL.
// This URL will Dial the NATS server at the URL in the environment variable
// NATS_SERVER_URL and receive messages with subject "example.mysubject".
subscription, err := pubsub.OpenSubscription(ctx,
"nats://example.mysubject?ackfunc=panic")
if err != nil {
log.Fatal(err)
}
defer subscription.Shutdown(ctx)
}
Output:
Example (OpenTopic) ¶
package main
import (
"context"
"log"
"github.com/eliben/gocdkx/pubsub"
)
func main() {
// This example is used in https://github.com/eliben/gocdkx/howto/pubsub/publish/#nats
// import _ "github.com/eliben/gocdkx/pubsub/natspubsub"
// Variables set up elsewhere:
ctx := context.Background()
// 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://example.mysubject")
if err != nil {
log.Fatal(err)
}
defer topic.Shutdown(ctx)
}
Output:
Index ¶
Examples ¶
Constants ¶
const AckWarning = "" /* 188-byte string literal not displayed */
AckWarning is a message that may be used in ackFuncs.
const Scheme = "nats"
Scheme is the URL scheme natspubsub registers its URLOpeners under on pubsub.DefaultMux.
Variables ¶
This section is empty.
Functions ¶
func OpenSubscription ¶
func OpenSubscription(nc *nats.Conn, subject string, ackFunc func(), _ *SubscriptionOptions) (*pubsub.Subscription, error)
OpenSubscription returns a *pubsub.Subscription representing a NATS subscription. The subject is the NATS Subject to subscribe to; for more info, see https://nats.io/documentation/writing_applications/subjects.
ackFunc will be called when the application calls pubsub.Topic.Ack on a received message; Ack is a meaningless no-op for NATS. You can provide an empty function to leave it a no-op, or panic/log a warning if you don't expect Ack to be called. See https://godoc.org/github.com/eliben/gocdkx/pubsub#hdr-At_most_once_and_At_least_once_Delivery for more background.
TODO(dlc) - Options for queue groups?
Example ¶
// This example is used in https://github.com/eliben/gocdkx/howto/pubsub/subscribe/#nats-ctor
// Variables set up elsewhere:
ctx := context.Background()
natsConn, err := nats.Connect("nats://nats.example.com")
if err != nil {
log.Fatal(err)
}
defer natsConn.Close()
subscription, err := natspubsub.OpenSubscription(
natsConn,
"example.mysubject",
func() { panic("nats does not have ack") },
nil)
if err != nil {
log.Fatal(err)
}
defer subscription.Shutdown(ctx)
Output:
func OpenTopic ¶
func OpenTopic(nc *nats.Conn, subject string, _ *TopicOptions) (*pubsub.Topic, error)
OpenTopic returns a *pubsub.Topic for use with NATS. The subject is the NATS Subject; for more info, see https://nats.io/documentation/writing_applications/subjects.
Example ¶
// This example is used in https://github.com/eliben/gocdkx/howto/pubsub/publish/#nats-ctor
// Variables set up elsewhere:
ctx := context.Background()
natsConn, err := nats.Connect("nats://nats.example.com")
if err != nil {
log.Fatal(err)
}
defer natsConn.Close()
topic, err := natspubsub.OpenTopic(natsConn, "example.mysubject", nil)
if err != nil {
log.Fatal(err)
}
defer topic.Shutdown(ctx)
Output:
Types ¶
type SubscriptionOptions ¶
type SubscriptionOptions struct{}
SubscriptionOptions sets options for constructing a *pubsub.Subscription backed by NATS.
type TopicOptions ¶
type TopicOptions struct{}
TopicOptions sets options for constructing a *pubsub.Topic backed by NATS.
type URLOpener ¶
type URLOpener struct {
// Connection to use for communication with the server.
Connection *nats.Conn
// TopicOptions specifies the options to pass to OpenTopic.
TopicOptions TopicOptions
// SubscriptionOptions specifies the options to pass to OpenSubscription.
SubscriptionOptions SubscriptionOptions
}
URLOpener opens NATS URLs like "nats://mysubject".
The URL host+path is used as the subject.
The following query parameters are supported:
- ackfunc: One of "log", "noop", "panic"; defaults to "panic". Determines the behavior if pubsub.Subscription.Ack (which is a meaningless no-op for NATS) is called: "log" means a log.Printf warning will be emitted; "noop" means nothing will happen; and "panic" means the application will panic. See https://godoc.org/github.com/eliben/gocdkx/pubsub#hdr-At_most_once_and_At_least_once_Delivery for more background.
No query parameters are supported.
func (*URLOpener) OpenSubscriptionURL ¶
func (o *URLOpener) OpenSubscriptionURL(ctx context.Context, u *url.URL) (*pubsub.Subscription, error)
OpenSubscriptionURL opens a pubsub.Subscription based on u.
Source Files