golnats

README

golnats

facade for regular used NATS use-cases

extracted from an older pandora box of such packages at abhishekkr/gol

Public Functions

Simple Flow, not Jetstream

• (*GolNats) Connect() (err error)
• (*GolNats) Close()
• (*GolNats) Unsubscribe()
• (*GolNats) Request(reqData []byte) (result []byte, err error)
• (*GolNats) Reply(evalMsg func(*nats.Msg)) (err error)
• (*GolNats) Publish(msg []byte) (err error)
• (*GolNats) PublishMessage(reply string, data []byte) (err error)
• (*GolNats) QueueSubscriber(qName string, evalMsg func(*nats.Msg)) (err error)
• (*GolNats) SubscriberAsync(evalMsg func(*nats.Msg)) (err error)
• (*GolNats) SubscriberSync(evalMsg func(*nats.Msg), timeout time.Duration) (err error)
• (*GolNats) SubscriberChan(evalMsg func(*nats.Msg)) (err error)
• (*GolNats) Log()
• ConnectNats(natsURLs, subject, authToken string) GolNats
• Log(m *nats.Msg)

Jetstream Flow

• (*JSGolNats) Connect(ctx context.Context, natsURLs, authToken string) error
• (*JSGolNats) Close()
• (*JSGolNats) AddDurableConsumerJS(ctx context.Context, consumerID string) error
• (*JSGolNats) PublishJS(ctx context.Context, subject string, msg []byte) error
• (*JSGolNats) SubscriberJS(ctx context.Context, subject string, evalMsg func(*nats.Msg)) error
• (*JSGolNats) PullSubscriberJS(ctx context.Context, subject, qName string, evalMsg func(*nats.Msg), breakMsg func(*nats.Msg) bool) error
• (*JSGolNats) LogJS()

---

Documentation

Index

• Variables
• func Log(m *nats.Msg)
• type GolNats
  • func ConnectNats(natsURLs, subject, authToken string) GolNats
  • func (nc *GolNats) Close()
  • func (nc *GolNats) Connect() (err error)
  • func (nc *GolNats) Log()
  • func (nc *GolNats) Publish(msg []byte) (err error)
  • func (nc *GolNats) PublishMessage(reply string, data []byte) (err error)
  • func (nc *GolNats) QueueSubscriber(qName string, evalMsg func(*nats.Msg)) (err error)
  • func (nc *GolNats) Reply(evalMsg func(*nats.Msg)) (err error)
  • func (nc *GolNats) Request(reqData []byte) (result []byte, err error)
  • func (nc *GolNats) SubscriberAsync(evalMsg func(*nats.Msg)) (err error)
  • func (nc *GolNats) SubscriberChan(evalMsg func(*nats.Msg)) (err error)
  • func (nc *GolNats) SubscriberSync(evalMsg func(*nats.Msg), timeout time.Duration) (err error)
  • func (nc *GolNats) Unsubscribe()
• type JSGolNats
  • func (jnc *JSGolNats) AddDurableConsumerJS(ctx context.Context, consumerID string) error
  • func (jnc *JSGolNats) Close()
  • func (jnc *JSGolNats) Connect(ctx context.Context, natsURLs, authToken string) error
  • func (jnc *JSGolNats) LogJS()
  • func (jnc *JSGolNats) PublishJS(ctx context.Context, subject string, msg []byte) error
  • func (jnc *JSGolNats) PullSubscriberJS(ctx context.Context, subject, qName string, evalMsg func(*nats.Msg), ...) error
  • func (jnc *JSGolNats) SubscriberJS(ctx context.Context, subject string, evalMsg func(*nats.Msg)) error

Variables

var (
	DefaultTimeoutMillisec = 100 * time.Millisecond
	DrainTimeoutMillisec   = 1000 * time.Millisecond
	FlushTimeoutMillisec   = 1000 * time.Millisecond
)

var (
	PubAsyncMaxPending = 256
	//https://docs.nats.io/nats-concepts/core-nats/queue
	DefaultRetentionPolicy = nats.LimitsPolicy // nats.WorkQueuePolicy
	//https://docs.nats.io/nats-concepts/jetstream/consumers
	DefaultAckPolicy       = nats.AckExplicitPolicy
	DefaultAckWaitMilliSec = 60 * time.Second
	DefaultDeliverPolicy   = nats.DeliverAllPolicy
	DefaultMaxDeliver      = 5
)

Functions

func Log

func Log(m *nats.Msg)

Types

type GolNats

type GolNats struct {
	URL        string // this would support CSV for cluster as well
	ConnName   string // for monitoring stats
	ConnToken  string // for using user:password, make them part of URLs as "myname:password@127.0.0.1;4222"
	Connection *nats.Conn

	Subscription *nats.Subscription
	Channel      chan *nats.Msg
	Subject      string
	LastMessage  []byte
	Timeout      time.Duration
}

func ConnectNats

func ConnectNats(natsURLs, subject, authToken string) GolNats

func (*GolNats) Close

func (nc *GolNats) Close()

func (*GolNats) Connect

func (nc *GolNats) Connect() (err error)

func (*GolNats) Log

func (nc *GolNats) Log()

func (*GolNats) Publish

func (nc *GolNats) Publish(msg []byte) (err error)

func (*GolNats) PublishMessage

func (nc *GolNats) PublishMessage(reply string, data []byte) (err error)

func (*GolNats) QueueSubscriber

func (nc *GolNats) QueueSubscriber(qName string, evalMsg func(*nats.Msg)) (err error)

func (*GolNats) Reply

func (nc *GolNats) Reply(evalMsg func(*nats.Msg)) (err error)

func (*GolNats) Request

func (nc *GolNats) Request(reqData []byte) (result []byte, err error)

func (*GolNats) SubscriberAsync

func (nc *GolNats) SubscriberAsync(evalMsg func(*nats.Msg)) (err error)

func (*GolNats) SubscriberChan

func (nc *GolNats) SubscriberChan(evalMsg func(*nats.Msg)) (err error)

func (*GolNats) SubscriberSync

func (nc *GolNats) SubscriberSync(evalMsg func(*nats.Msg), timeout time.Duration) (err error)

func (*GolNats) Unsubscribe

func (nc *GolNats) Unsubscribe()

type JSGolNats

type JSGolNats struct {
	GolNats        GolNats
	JS             nats.JetStreamContext
	Stream         string
	Subjects       []string
	Retention      nats.RetentionPolicy
	AckPolicy      nats.AckPolicy
	AckWait        time.Duration
	DeliveryPolicy nats.DeliverPolicy
	MaxDeliver     int
}

func (*JSGolNats) AddDurableConsumerJS

func (jnc *JSGolNats) AddDurableConsumerJS(ctx context.Context, consumerID string) error

func (*JSGolNats) Close

func (jnc *JSGolNats) Close()

func (*JSGolNats) Connect

func (jnc *JSGolNats) Connect(ctx context.Context, natsURLs, authToken string) error

func (*JSGolNats) LogJS

func (jnc *JSGolNats) LogJS()

func (*JSGolNats) PublishJS

func (jnc *JSGolNats) PublishJS(ctx context.Context, subject string, msg []byte) error

func (*JSGolNats) PullSubscriberJS

func (jnc *JSGolNats) PullSubscriberJS(ctx context.Context, subject, qName string, evalMsg func(*nats.Msg), breakMsg func(*nats.Msg) bool) error

func (*JSGolNats) SubscriberJS

func (jnc *JSGolNats) SubscriberJS(ctx context.Context, subject string, evalMsg func(*nats.Msg)) error