natsutil

README

natsutil.go

Status: EXPERIMENTAL

A collection of utilities for working with NATS.io.

Documentation

Full go doc style documentation for the project can be viewed online without installing this package by using the excellent GoDoc site here: http://godoc.org/github.com/41north/natsutil.go

You can also view the documentation locally once the package is installed with the godoc tool by running godoc -http=":6060" and pointing your browser to http://localhost:6060/pkg/github.com/41north/natsutil.go

Installation

$ go get -u github.com/41north/natsutil.go

Add this import line to the file you're working in:

import natsutil "github.com/41north/natsutil.go"

Quick Start

The following utilities will be of interest:

• Subject Builder
• Generic Key Value Store

Subject Builder

A builder much like strings.Builder which helps with constructing valid subject names and ensures invalid characters are not used:

sb := natsutil.SubjectBuilder{}

// 'foo.bar.baz.*'
sb.Push("foo", "bar", "baz")
sb.Star()
subject := sb.String()

// 'foo.bar'
sb.Pop(2)
subject = sb.String()

// 'foo.bar.hello.>'
sb.Push("hello")
sb.Chevron()
subject = sb.String()

Generic Key Value Store

A generic interface for interacting with JetStream Key-Value stores can be created with the following:

type testPayload struct {
	Value int `json:""`
}

// create a nats connection
nc, err := nats.Connect(s.ClientURL(), opts...)
...

// get a JetStream context
js, err := nc.JetStream(nats.MaxWait(10 * time.Second))
...

// get a reference to the nats KV interface
kv, err := js.KeyValue("my-bucket")
...

// create a generic KeyValue interface that uses JSON encoding
encoder := builtin.JsonEncoder{}
kvT = natsutil.NewKeyValue[testPayload](kv, &codec)

// put a testPayload object
kvT.Put("foo", testPayload{1})

// get a testPayload object
get, err := kvT.Get("foo")

// create a generic Watcher
watcher, err := kvT.WatchAll()
...

License

Go-async is licensed under the Apache 2.0 License

Contact

If you want to get in touch drop us an email at hello@41north.dev

Documentation

Index

• Constants
• type KeyValue
  • func NewKeyValue[T any](delegate nats.KeyValue, encoder nats.Encoder) KeyValue[T]
• type KeyValueEntry
• type KeyWatcher
  • func NewKeyWatcher[T any](watcher nats.KeyWatcher, encoder nats.Encoder) KeyWatcher[T]
• type SubjectBuilder
  • func (b *SubjectBuilder) Chevron()
  • func (b *SubjectBuilder) MustPop(count int)
  • func (b *SubjectBuilder) MustPush(elements ...string)
  • func (b *SubjectBuilder) Pop(count int) error
  • func (b *SubjectBuilder) Push(elements ...string) error
  • func (b *SubjectBuilder) Star()
  • func (b *SubjectBuilder) String() string

Constants

const (
	SubjectSeparator = "."
	SubjectStar      = "*"
	SubjectChevron   = ">"
)

const (
	ErrSubjectInvalidCharacters = errors.ConstError("subject component must only contain [a-zA-Z0-9_]")
	ErrPopInsufficientElements  = errors.ConstError("cannot pop more elements than have already been pushed")
)

Types

type KeyValue

type KeyValue[T any] interface {
	// Delegate returns the underlying nats.KeyValue instance.
	Delegate() nats.KeyValue
	// Encoder returns the codec used for marshalling to and from bytes.
	Encoder() nats.Encoder
	// Get returns the latest value for the key.
	Get(key string) (entry KeyValueEntry[T], err error)
	// GetRevision returns a specific revision value for the key.
	GetRevision(key string, revision uint64) (entry KeyValueEntry[T], err error)
	// Put will place the new value for the key into the store.
	Put(key string, value T) (revision uint64, err error)
	// Create will add the key/value pair iff it does not exist.
	Create(key string, value T) (revision uint64, err error)
	// Update will update the value iff the latest revision matches.
	Update(key string, value T, last uint64) (revision uint64, err error)
	// Watch for any updates to keys that match the keys argument which could include wildcards.
	// Watch will send a nil entry when it has received all initial values.
	Watch(keys string, opts ...nats.WatchOpt) (KeyWatcher[T], error)
	// WatchAll will invoke the callback for all updates.
	WatchAll(opts ...nats.WatchOpt) (KeyWatcher[T], error)
	// History will return all historical values for the key.
	History(key string, opts ...nats.WatchOpt) ([]KeyValueEntry[T], error)
	// Bucket returns the current bucket name.
	Bucket() string
}

KeyValue provides a generic interface for nats.KeyValue.

func NewKeyValue

func NewKeyValue[T any](delegate nats.KeyValue, encoder nats.Encoder) KeyValue[T]

type KeyValueEntry

type KeyValueEntry[T any] interface {
	nats.KeyValueEntry
	// UnmarshalValue decodes and returns the retrieve value.
	UnmarshalValue() (T, error)
}

KeyValueEntry provides a generic interface for nats.KeyValueEntry.

type KeyWatcher

type KeyWatcher[T any] interface {
	nats.KeyWatcher
	// UpdatesUnmarshalled provides a decoded view of the Updates() channel.
	UpdatesUnmarshalled() <-chan KeyValueEntry[T]
}

KeyWatcher provides a generic interface for nats.KeyWatcher.

func NewKeyWatcher

func NewKeyWatcher[T any](watcher nats.KeyWatcher, encoder nats.Encoder) KeyWatcher[T]

type SubjectBuilder

type SubjectBuilder struct {
	// contains filtered or unexported fields
}

SubjectBuilder helps with constructing valid nats subject names.

func (*SubjectBuilder) Chevron

func (b *SubjectBuilder) Chevron()

Chevron appends a '>' wildcard to the subject under construction, prepending a separator as needed.

func (*SubjectBuilder) MustPop

func (b *SubjectBuilder) MustPop(count int)

MustPop is a variant of Pop which panics if an error is returned.

func (*SubjectBuilder) MustPush

func (b *SubjectBuilder) MustPush(elements ...string)

MustPush is a variant of Push which panics if an error is returned.

func (*SubjectBuilder) Pop

func (b *SubjectBuilder) Pop(count int) error

Pop removes a number of elements from the end of the subject currently under construction. Pop returns ErrPopInsufficientElements if you attempt to pop more elements than have been pushed.

func (*SubjectBuilder) Push

func (b *SubjectBuilder) Push(elements ...string) error

Push takes the provided component and appends them to the subject under construction

func (*SubjectBuilder) Star

func (b *SubjectBuilder) Star()

Star appends a '*' wildcard to the subject under construction, prepending a separator as needed.

func (*SubjectBuilder) String

func (b *SubjectBuilder) String() string

String outputs the subject that has been constructed so far.