nats

README

NATS.IO

Install

go get github.com/aacfactory/fns-contrib/message-queues/nats

Usage

app.Deploy(nats.Service())

Config

nats:
  uri: "nats://"
  producers:
    yourSubjectName:
      size: 8
  consumers:
    bar:
      handler: default
      queue: ""
      subject: ""

Use user consumer handler

consumers:
  bar:
    handler: user_consumer
    handlerOptions:
      userId: "userId"

As proxy

published, publishErr := nats.Publish(ctx, nats.PublishArgument{
	Subject: "subject name", 
	Body: json.RawMessage([]byte("{}"))
})

Documentation

Index

• func Publish(ctx context.Context, argument PublishArgument) (ok bool, err errors.CodeError)
• func RegisterConsumerHandler(name string, builder ConsumerHandlerBuilder)
• func Service() service.Service
• type ClientTLSConfig
• type Config
• type Consumer
  • func (consumer *Consumer) Close() (err error)
  • func (consumer *Consumer) Consume(ctx context.Context) (err error)
• type ConsumerConfig
• type ConsumerHandler
• type ConsumerHandlerBuilder
• type ConsumerHandlerOptions
• type ConsumerMessage
• type DefaultConsumerMessage
  • func (msg *DefaultConsumerMessage) Ack() (err error)
  • func (msg *DefaultConsumerMessage) Body() (body []byte)
  • func (msg *DefaultConsumerMessage) Raw() (raw *nats.Msg)
  • func (msg *DefaultConsumerMessage) Reject() (err error)
• type Message
• type OptionsConfig
• type Producer
  • func (producer *Producer) Close() (err error)
  • func (producer *Producer) Publish(_ context.Context, msg []byte) (ok bool, err errors.CodeError)
• type ProducerConfig
• type PublishArgument
• type PublishResult

Functions

func Publish

func Publish(ctx context.Context, argument PublishArgument) (ok bool, err errors.CodeError)

func RegisterConsumerHandler

func RegisterConsumerHandler(name string, builder ConsumerHandlerBuilder)

func Service

func Service() service.Service

Types

type ClientTLSConfig

type ClientTLSConfig struct {
	CA   string `json:"ca"`
	Cert string `json:"cert"`
	Key  string `json:"key"`
}

type Config

type Config struct {
	URI       string                     `json:"uri"`
	Options   *OptionsConfig             `json:"options"`
	Producers map[string]*ProducerConfig `json:"producers"`
	Consumers map[string]*ConsumerConfig `json:"consumers"`
}

type Consumer

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

func (*Consumer) Close

func (consumer *Consumer) Close() (err error)

func (*Consumer) Consume

func (consumer *Consumer) Consume(ctx context.Context) (err error)

type ConsumerConfig

type ConsumerConfig struct {
	Handler        string          `json:"handler"`
	HandlerOptions json.RawMessage `json:"handlerOptions"`
	Subject        string          `json:"subject"`
	Queue          string          `json:"queue"`
}

type ConsumerHandler

type ConsumerHandler interface {
	Handle(ctx context.Context, message ConsumerMessage)
}

type ConsumerHandlerBuilder

type ConsumerHandlerBuilder func(options ConsumerHandlerOptions) (handler ConsumerHandler, err error)

type ConsumerHandlerOptions

type ConsumerHandlerOptions struct {
	Log    logs.Logger
	Config configures.Config
}

type ConsumerMessage

type ConsumerMessage interface {
	Body() (body []byte)
	Ack() (err error)
	Reject() (err error)
	Raw() (raw *nats.Msg)
}

type DefaultConsumerMessage

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

func (*DefaultConsumerMessage) Ack

func (msg *DefaultConsumerMessage) Ack() (err error)

func (*DefaultConsumerMessage) Body

func (msg *DefaultConsumerMessage) Body() (body []byte)

func (*DefaultConsumerMessage) Raw

func (msg *DefaultConsumerMessage) Raw() (raw *nats.Msg)

func (*DefaultConsumerMessage) Reject

func (msg *DefaultConsumerMessage) Reject() (err error)

type Message

type Message struct {
	Service  string          `json:"service"`
	Fn       string          `json:"fn"`
	Argument json.RawMessage `json:"argument"`
}

type OptionsConfig

type OptionsConfig struct {
	Name              string           `json:"name"`
	User              string           `json:"user"`
	Password          string           `json:"password"`
	Token             string           `json:"token"`
	EnableCompression bool             `json:"enableCompression"`
	TimeoutSeconds    int              `json:"timeoutSeconds"`
	ClientTLS         *ClientTLSConfig `json:"clientTLS"`
}

type Producer

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

func (*Producer) Close

func (producer *Producer) Close() (err error)

func (*Producer) Publish

func (producer *Producer) Publish(_ context.Context, msg []byte) (ok bool, err errors.CodeError)

type ProducerConfig

type ProducerConfig struct {
	Size int `json:"size"`
}

type PublishArgument

type PublishArgument struct {
	Subject string          `json:"subject"`
	Body    json.RawMessage `json:"body"`
}

type PublishResult

type PublishResult struct {
	Succeed bool `json:"succeed"`
}