nats

Documentation

Index

• func ClientRequest[REQUEST any, RESPONSE any](subject string, req REQUEST) (RESPONSE, error)
• func Connect(cfg Config) error
• func ConvertData[T any](data []byte) T
• func ToBytes(data interface{}) []byte
• type CommonResponse
• type Config
• type JSONEncoder
  • func (e *JSONEncoder) Publish(reply string, data interface{}) error
  • func (e *JSONEncoder) QueueSubscribe(subject string, queue string, cb nats.Handler) (*nats.Subscription, error)
  • func (e *JSONEncoder) Request(subject string, data interface{}, res interface{}, isUseNameSpace bool) error
  • func (e *JSONEncoder) RequestCommonNS(subject string, data interface{}, res interface{}) error
  • func (e *JSONEncoder) Subscribe(subject string, cb nats.Handler) (*nats.Subscription, error)
• type JetStream
  • func GetJetStream() JetStream
  • func (js JetStream) AddConsumer(stream, subject, name string) error
  • func (js JetStream) AddStream(name string, subjects []string) error
  • func (js JetStream) AddStreamSubjects(name string, subjects []string) error
  • func (js JetStream) DeleteStream(name string) error
  • func (js JetStream) GetConsumerInfo(stream, name string) (*nats.ConsumerInfo, error)
  • func (js JetStream) GetStreamInfo(name string) (*nats.StreamInfo, error)
  • func (js JetStream) Publish(stream, subject string, payload []byte) error
  • func (js JetStream) PullSubscribe(stream, subject, consumer string) (*nats.Subscription, error)
  • func (js JetStream) QueueSubscribe(stream, subject, queueName string, cb nats.MsgHandler) error
  • func (js JetStream) Subscribe(stream, subject string, cb nats.MsgHandler) (*nats.Subscription, error)
• type MethodHandle
• type Nats
  • func GetServer() *Nats
  • func (sv *Nats) NewJSONEncodedConn() (*JSONEncoder, error)
  • func (sv *Nats) Publish(sub string, payload []byte) error
  • func (sv *Nats) PublishRequest(sub, reply string, data []byte) error
  • func (sv *Nats) QueueSubscribe(subject, queue string, cb nats.MsgHandler) (*nats.Subscription, error)
  • func (sv *Nats) Reply(msg *nats.Msg, payload []byte) error
  • func (sv *Nats) Request(subject string, payload []byte) (*nats.Msg, error)
  • func (sv *Nats) Response(msg *nats.Msg, payload interface{}, message string) error
  • func (sv *Nats) Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error)
  • func (sv *Nats) SubscribeSync() (*nats.Subscription, error)
• type NatsResponse
• type QueueDesc
  • func (q QueueDesc) GetSubject(serverName string) string
• type ServiceDesc
  • func (s ServiceDesc) Register(sv *Nats, queue interface{}) error
• type TLSConfig

Functions

func ClientRequest

func ClientRequest[REQUEST any, RESPONSE any](subject string, req REQUEST) (RESPONSE, error)

ClientRequest ...

func Connect

func Connect(cfg Config) error

Connect ...

func ConvertData

func ConvertData[T any](data []byte) T

ConvertData ...

func ToBytes

func ToBytes(data interface{}) []byte

ToBytes ...

Types

type CommonResponse

type CommonResponse struct {
	Data  interface{} `json:"data"`
	Error string      `json:"error"`
}

type Config

type Config struct {
	// Connect url
	URL string

	// Auth user
	User string

	// Auth password
	Password string

	// TLS config
	TLS *TLSConfig

	// RequestTimeout
	RequestTimeout time.Duration

	Debug bool
}

Config ...

type JSONEncoder

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

JSONEncoder ...

func (*JSONEncoder) Publish

func (e *JSONEncoder) Publish(reply string, data interface{}) error

Publish ...

func (*JSONEncoder) QueueSubscribe

func (e *JSONEncoder) QueueSubscribe(subject string, queue string, cb nats.Handler) (*nats.Subscription, error)

QueueSubscribe ...

func (*JSONEncoder) Request

func (e *JSONEncoder) Request(subject string, data interface{}, res interface{}, isUseNameSpace bool) error

Request ...

func (*JSONEncoder) RequestCommonNS

func (e *JSONEncoder) RequestCommonNS(subject string, data interface{}, res interface{}) error

func (*JSONEncoder) Subscribe

func (e *JSONEncoder) Subscribe(subject string, cb nats.Handler) (*nats.Subscription, error)

Subscribe ...

type JetStream

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

JetStream ...

func GetJetStream

func GetJetStream() JetStream

GetJetStream ...

func (JetStream) AddConsumer

func (js JetStream) AddConsumer(stream, subject, name string) error

AddConsumer ...

func (JetStream) AddStream

func (js JetStream) AddStream(name string, subjects []string) error

AddStream add new stream, with default config Due to subject must have a unique name, subject name will be combined with stream name E.g: stream name is "DEMO", subject name is "Subject-1", so final name in NATS will be: DEMO.Subject-1

func (JetStream) AddStreamSubjects

func (js JetStream) AddStreamSubjects(name string, subjects []string) error

AddStreamSubjects ...

func (JetStream) DeleteStream

func (js JetStream) DeleteStream(name string) error

DeleteStream ...

func (JetStream) GetConsumerInfo

func (js JetStream) GetConsumerInfo(stream, name string) (*nats.ConsumerInfo, error)

GetConsumerInfo ...

func (JetStream) GetStreamInfo

func (js JetStream) GetStreamInfo(name string) (*nats.StreamInfo, error)

GetStreamInfo ...

func (JetStream) Publish

func (js JetStream) Publish(stream, subject string, payload []byte) error

Publish ...

func (JetStream) PullSubscribe

func (js JetStream) PullSubscribe(stream, subject, consumer string) (*nats.Subscription, error)

PullSubscribe ...

func (JetStream) QueueSubscribe

func (js JetStream) QueueSubscribe(stream, subject, queueName string, cb nats.MsgHandler) error

QueueSubscribe ...

func (JetStream) Subscribe

func (js JetStream) Subscribe(stream, subject string, cb nats.MsgHandler) (*nats.Subscription, error)

Subscribe ...

type MethodHandle

type MethodHandle func(queue interface{}) nats.MsgHandler

type Nats

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

Nats ...

func GetServer

func GetServer() *Nats

GetServer ...

func (*Nats) NewJSONEncodedConn

func (sv *Nats) NewJSONEncodedConn() (*JSONEncoder, error)

NewJSONEncodedConn ...

func (*Nats) Publish

func (sv *Nats) Publish(sub string, payload []byte) error

func (*Nats) PublishRequest

func (sv *Nats) PublishRequest(sub, reply string, data []byte) error

func (*Nats) QueueSubscribe

func (sv *Nats) QueueSubscribe(subject, queue string, cb nats.MsgHandler) (*nats.Subscription, error)

QueueSubscribe ...

func (*Nats) Reply

func (sv *Nats) Reply(msg *nats.Msg, payload []byte) error

Reply ...

func (*Nats) Request

func (sv *Nats) Request(subject string, payload []byte) (*nats.Msg, error)

Request ...

func (*Nats) Response

func (sv *Nats) Response(msg *nats.Msg, payload interface{}, message string) error

Response ...

func (*Nats) Subscribe

func (sv *Nats) Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error)

Subscribe ...

func (*Nats) SubscribeSync

func (sv *Nats) SubscribeSync() (*nats.Subscription, error)

type NatsResponse

type NatsResponse struct {
	Success bool   `json:"success"`
	Message string `json:"message"`
	Data    []byte `json:"data"`
}

NatsResponse ..

type QueueDesc

type QueueDesc struct {
	Subject string
	Worker  string
	Handle  MethodHandle
}

func (QueueDesc) GetSubject

func (q QueueDesc) GetSubject(serverName string) string

GetSubject get subject for nats request

type ServiceDesc

type ServiceDesc struct {
	Queues []QueueDesc
}

func (ServiceDesc) Register

func (s ServiceDesc) Register(sv *Nats, queue interface{}) error

Register callback handler

type TLSConfig

type TLSConfig struct {
	// Cert file
	CertFilePath string

	// Key file
	KeyFilePath string

	// Root CA
	RootCAFilePath string
}

TLSConfig ...