nats_messaging

README

nats-messaging-go

Module nats-messaging-go implements convenience routines around the NATS - Go Client. This package also provides support for OTel trace context propagation via NATs messages using NATS message headers. Its intended to be used by Go apps that want to connect & communicate over NATS pub/sub without knowing much about NATS - Go Client.

Installation

At a high level, this module depends on nats client and opentelemetry go libraries. Running the following go get command will fetch these and any future dependencies as referenced in the code:

go get github.com/mskcc/nats-messaging-go

Basic Usage

Code examples for connecting, publishing, and subscribing to a NATS messaging service.

import (
	nm "github.com/mskcc/nats-messaging-go"
)

// without TLS
m, err := nm.NewMessaging("localhost:4222")

// with TLS
m, err := nm.NewSecureMessaging("localhost:4222", "certPath", "keyPath", "userId", "pw")
if err != nil {
	// do something
}
defer m.Shutdown()

// publish a message
err = m.Publish("subject", []byte("Hello World"))
if err != nil {
	// do something	
}

// subscribe to a subject
// consumer id much match an authorized id setup in NATS/Jetstream configuration 
m.Subscribe("consumer id", " subject", func(m *nm.Msg) {
	log.Println("Subscriber received an message via NATS on subject:", m.Subject))
	log.Println("Subscriber received an message via NATS:", string(m.Data))
})

// publish a message and propagate otel trace context
ctx, span := tracer.Start(...)
...
err = m.PublishWithTraceContext(ctx, "subject", []byte("Hello World"))
if err != nil {
    // do something
}

// subscribe to a subject and receive a propagated otel trace context
m.SubscribeWithTraceContext("consumer id", " subject", func(ctx context.Context, m *nm.Msg) {
    _, span := tracer.Start(ctx, "Nats Subscriber")_
    defer span.End()
	log.Println("Subscriber received an message via NATS on subject:", m.Subject))
	log.Println("Subscriber received a message via NATS:", string(m.Data))
    ...
})

Documentation

Overview

nats_messaging implements convenience routines around the NATS - Go Client (https://github.com/nats-io/nats.go). provides support for OTel trace context propagation via NATs messages using NATS message headers. intended to be used by go apps that want to communicate via NATS pub/sub without having to know much about nats.go

Index

• type Messaging
  • func NewMessaging(url string) (*Messaging, error)
  • func NewSecureMessaging(url string, certPath, keyPath, user, pw string) (*Messaging, error)
  • func (m *Messaging) Publish(subj string, data []byte) error
  • func (m *Messaging) PublishWithTraceContext(ctx context.Context, subj string, data []byte) error
  • func (m *Messaging) Shutdown()
  • func (m *Messaging) Subscribe(con string, subj string, mh MsgHandler) error
  • func (m *Messaging) SubscribeWithTraceContext(con string, subj string, mh MsgHandlerWithTraceContext) error
• type Msg
• type MsgHandler
• type MsgHandlerWithTraceContext

Types

type Messaging

type Messaging struct {

	// Js available outside package for stream & consumer management
	Js nats.JetStreamContext
	// contains filtered or unexported fields
}

this type represents a reference to the nats messaging client. it is obtained via a call to NewMessaging() and is require for all subsequent calls to pub/sub routines: Messaging.Publish(...), Messaging.Subscribe(...), ...

func NewMessaging

func NewMessaging(url string) (*Messaging, error)

initializes a Messaging type. a call to this routine is a prerequisite to pub/sub operations

func NewSecureMessaging

func NewSecureMessaging(url string, certPath, keyPath, user, pw string) (*Messaging, error)

initializes a secure/tls Messaging type. a call to this routine is a prerequisite to pub/sub operations

func (*Messaging) Publish

func (m *Messaging) Publish(subj string, data []byte) error

routine used to Publish a message (represented by []byte) on the given subject

func (*Messaging) PublishWithTraceContext

func (m *Messaging) PublishWithTraceContext(ctx context.Context, subj string, data []byte) error

func (*Messaging) Shutdown

func (m *Messaging) Shutdown()

func (*Messaging) Subscribe

func (m *Messaging) Subscribe(con string, subj string, mh MsgHandler) error

func (*Messaging) SubscribeWithTraceContext

func (m *Messaging) SubscribeWithTraceContext(con string, subj string, mh MsgHandlerWithTraceContext) error

type Msg

type Msg struct {
	Subject     string
	Data        []byte
	ProviderMsg *nats.Msg
}

these types are here so clients do not have to depend on nats libs when subscribing

type MsgHandler

type MsgHandler func(msg *Msg)

type MsgHandlerWithTraceContext

type MsgHandlerWithTraceContext func(ctx context.Context, msg *Msg)