jetstream

Documentation

Index

• func New(ctx context.Context, connection string, tlsInfo tls.Config) (server.Backend, error)
• type Config
• type JSValue
• type JetStream
  • func (j *JetStream) Count(ctx context.Context, prefix string) (revRet int64, count int64, err error)
  • func (j *JetStream) Create(ctx context.Context, key string, value []byte, lease int64) (revRet int64, errRet error)
  • func (j *JetStream) DbSize(context.Context) (int64, error)
  • func (j *JetStream) Delete(ctx context.Context, key string, revision int64) (revRet int64, kvRet *server.KeyValue, deletedRet bool, errRet error)
  • func (j *JetStream) Get(ctx context.Context, key, rangeEnd string, limit, revision int64) (revRet int64, kvRet *server.KeyValue, errRet error)
  • func (j *JetStream) List(ctx context.Context, prefix, startKey string, limit, revision int64) (revRet int64, kvRet []*server.KeyValue, errRet error)
  • func (j *JetStream) Start(ctx context.Context) error
  • func (j *JetStream) Update(ctx context.Context, key string, value []byte, revision, lease int64) (revRet int64, kvRet *server.KeyValue, updateRet bool, errRet error)
  • func (j *JetStream) Watch(ctx context.Context, prefix string, revision int64) <-chan []*server.Event

Functions

func New

func New(ctx context.Context, connection string, tlsInfo tls.Config) (server.Backend, error)

New get the JetStream Backend, establish connection to NATS JetStream. At the moment nats.go does not have connection string support so kine will use:

nats://(token|username:password)hostname:port?bucket=bucketName&contextFile=nats-context&slowMethod=<duration>&revHistory=<revCount>`.

If contextFile is provided then do not provide a hostname:port in the endpoint URL, instead use the context file to provide the NATS server url(s).

bucket: specifies the bucket on the nats server for the k8s key/values for this cluster (optional)
contextFile: specifies the nats context file to load e.g. /etc/nats/context.json
revHistory: controls the rev history for JetStream defaults to 10 must be > 2 and <= 64
slowMethod: used to log methods slower than provided duration default 500ms

Multiple urls can be passed in a comma separated format - only the first in the list will be evaluated for query parameters. While auth is valid in the url, the preferred way to pass auth is through a context file. If user/pass or token are provided in the url only the first one will be used for all urls. / If no bucket query parameter is provided it will default to kine

https://docs.nats.io/using-nats/nats-tools/nats_cli#configuration-contexts

example nats-context.json:

{
  "description": "optional context description",
  "url": "nats://127.0.0.1:4222",
  "token": "",
  "user": "",
  "password": "",
  "creds": "",
  "nkey": "",
  "cert": "",
  "key": "",
  "ca": "",
  "nsc": "",
  "jetstream_domain": "",
  "jetstream_api_prefix": "",
  "jetstream_event_prefix": ""
}

Types

type Config

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

type JSValue

type JSValue struct {
	KV           *server.KeyValue
	PrevRevision int64
	Create       bool
	Delete       bool
}

type JetStream

type JetStream struct {
	server.Backend
	// contains filtered or unexported fields
}

func (*JetStream) Count

func (j *JetStream) Count(ctx context.Context, prefix string) (revRet int64, count int64, err error)

Count returns an exact count of the number of matching keys and the current revision of the database

func (*JetStream) Create

func (j *JetStream) Create(ctx context.Context, key string, value []byte, lease int64) (revRet int64, errRet error)

Create

func (*JetStream) DbSize

func (j *JetStream) DbSize(context.Context) (int64, error)

DbSize get the kineBucket size from JetStream.

func (*JetStream) Delete

func (j *JetStream) Delete(ctx context.Context, key string, revision int64) (revRet int64, kvRet *server.KeyValue, deletedRet bool, errRet error)

func (*JetStream) Get

func (j *JetStream) Get(ctx context.Context, key, rangeEnd string, limit, revision int64) (revRet int64, kvRet *server.KeyValue, errRet error)

Get returns the associated server.KeyValue

func (*JetStream) List

func (j *JetStream) List(ctx context.Context, prefix, startKey string, limit, revision int64) (revRet int64, kvRet []*server.KeyValue, errRet error)

func (*JetStream) Start

func (j *JetStream) Start(ctx context.Context) error

func (*JetStream) Update

func (j *JetStream) Update(ctx context.Context, key string, value []byte, revision, lease int64) (revRet int64, kvRet *server.KeyValue, updateRet bool, errRet error)

func (*JetStream) Watch

func (j *JetStream) Watch(ctx context.Context, prefix string, revision int64) <-chan []*server.Event