peanats

README

peanats

NATS based server with API similar to net/http package.

Documentation

Index

• Constants
• func PublishError(pub Publisher, err error) error
• type AccessLogMiddlewareOption
  • func AccessLogMiddlewareWithLogger(log AccessLogger) AccessLogMiddlewareOption
  • func AccessLogMiddlewareWithWriter(w io.Writer) AccessLogMiddlewareOption
• type AccessLogger
• type ChanSubscriber
• type Codec
• type Connection
  • func NATS(nc *nats.Conn) Connection
• type Error
  • func (e Error) Error() string
  • func (e Error) Unwrap() error
• type Handler
  • func ChainMiddleware(h Handler, mw ...Middleware) Handler
  • func ErrorHandlerMiddleware(next Handler) Handler
  • func RequestUIDMiddleware(next Handler) Handler
  • func Typed[ArgT, ResT any](c Codec, f TypedHandler[ArgT, ResT]) Handler
• type HandlerFunc
  • func (f HandlerFunc) Serve(pub Publisher, req Request) error
• type JsonCodec
  • func (JsonCodec) Decode(data []byte, vPtr any) error
  • func (JsonCodec) Encode(v any) ([]byte, error)
• type Matcher
  • func NewMatcher(pat string) Matcher
• type Middleware
  • func MakeAccessLogMiddleware(opts ...AccessLogMiddlewareOption) Middleware
  • func MakePublishSubjectMiddleware(subject string) Middleware
• type ProtoCodec
  • func (ProtoCodec) Decode(data []byte, vPtr any) error
  • func (ProtoCodec) Encode(v any) ([]byte, error)
• type ProtojsonCodec
  • func (ProtojsonCodec) Decode(data []byte, vPtr any) error
  • func (ProtojsonCodec) Encode(v any) ([]byte, error)
• type PrototextCodec
  • func (PrototextCodec) Decode(data []byte, vPtr any) error
  • func (PrototextCodec) Encode(v any) ([]byte, error)
• type Publisher
  • func NewPublisher(p PublisherMsg) Publisher
  • func NewPublisherWithSubject(p PublisherMsg, subject string) Publisher
• type PublisherMsg
• type Request
• type RequesterMsg
• type ServeMux
  • func (m *ServeMux) Handle(f Handler, subjects ...string) error
  • func (m *ServeMux) HandleFunc(f func(Publisher, Request) error, subjects ...string) error
  • func (m *ServeMux) Serve(pub Publisher, req Request) error
• type Server
  • func (s *Server) Shutdown()
  • func (s *Server) Start() error
  • func (s *Server) Wait()
• type Subscriber
• type Subscription
• type TypedHandler
• type TypedHandlerFunc
  • func (f TypedHandlerFunc[ArgT, ResT]) Serve(pub TypedPublisher[ResT], req TypedRequest[ArgT]) error
• type TypedPublisher
• type TypedRequest
• type UIDGenerator
• type Unsubscriber

Constants

const (
	HeaderErrorCode    = "error-code"
	HeaderErrorMessage = "error-message"
)

const HeaderRequestUID = "Request-UID"

Functions

func PublishError

func PublishError(pub Publisher, err error) error

Types

type AccessLogMiddlewareOption

type AccessLogMiddlewareOption func(*loggingPublisher)

func AccessLogMiddlewareWithLogger

func AccessLogMiddlewareWithLogger(log AccessLogger) AccessLogMiddlewareOption

func AccessLogMiddlewareWithWriter

func AccessLogMiddlewareWithWriter(w io.Writer) AccessLogMiddlewareOption

type AccessLogger

type AccessLogger interface {
	Println(v ...any)
}

type ChanSubscriber

type ChanSubscriber interface {
	ChanSubscribe(subj string, ch chan *nats.Msg) (Unsubscriber, error)
	ChanQueueSubscribe(subj, queue string, ch chan *nats.Msg) (Unsubscriber, error)
}

type Codec

type Codec interface {
	Encode(v any) ([]byte, error)
	Decode(data []byte, vPtr any) error
}

type Connection

type Connection interface {
	PublisherMsg
	RequesterMsg
	Subscriber
	ChanSubscriber
	Drain() error
}

func NATS

func NATS(nc *nats.Conn) Connection

type Error

type Error struct {
	Code    int
	Message string
	Cause   error
}

func (Error) Error

func (e Error) Error() string

func (Error) Unwrap

func (e Error) Unwrap() error

type Handler

type Handler interface {
	Serve(Publisher, Request) error
}

func ChainMiddleware

func ChainMiddleware(h Handler, mw ...Middleware) Handler

func ErrorHandlerMiddleware

func ErrorHandlerMiddleware(next Handler) Handler

func RequestUIDMiddleware

func RequestUIDMiddleware(next Handler) Handler

func Typed

func Typed[ArgT, ResT any](c Codec, f TypedHandler[ArgT, ResT]) Handler

type HandlerFunc

type HandlerFunc func(Publisher, Request) error

func (HandlerFunc) Serve

func (f HandlerFunc) Serve(pub Publisher, req Request) error

type JsonCodec

type JsonCodec struct{}

func (JsonCodec) Decode

func (JsonCodec) Decode(data []byte, vPtr any) error

func (JsonCodec) Encode

func (JsonCodec) Encode(v any) ([]byte, error)

type Matcher

type Matcher interface {
	Match(string) bool
}

func NewMatcher

func NewMatcher(pat string) Matcher

type Middleware

type Middleware func(next Handler) Handler

func MakeAccessLogMiddleware

func MakeAccessLogMiddleware(opts ...AccessLogMiddlewareOption) Middleware

MakeAccessLogMiddleware returns a middleware that logs the request and response of a handler.

WARNING: the logger is inefficient and should not be used in production. TODO: replace with efficient implementation.

func MakePublishSubjectMiddleware

func MakePublishSubjectMiddleware(subject string) Middleware

type ProtoCodec

type ProtoCodec struct{}

func (ProtoCodec) Decode

func (ProtoCodec) Decode(data []byte, vPtr any) error

func (ProtoCodec) Encode

func (ProtoCodec) Encode(v any) ([]byte, error)

type ProtojsonCodec

type ProtojsonCodec struct{}

func (ProtojsonCodec) Decode

func (ProtojsonCodec) Decode(data []byte, vPtr any) error

func (ProtojsonCodec) Encode

func (ProtojsonCodec) Encode(v any) ([]byte, error)

type PrototextCodec

type PrototextCodec struct{}

func (PrototextCodec) Decode

func (PrototextCodec) Decode(data []byte, vPtr any) error

func (PrototextCodec) Encode

func (PrototextCodec) Encode(v any) ([]byte, error)

type Publisher

type Publisher interface {
	PublisherMsg
	Subject() string
	WithSubject(subject string) Publisher
	Header() *nats.Header
	Publish(data []byte) error
}

func NewPublisher

func NewPublisher(p PublisherMsg) Publisher

func NewPublisherWithSubject

func NewPublisherWithSubject(p PublisherMsg, subject string) Publisher

type PublisherMsg

type PublisherMsg interface {
	PublishMsg(msg *nats.Msg) error
}

type Request

type Request interface {
	Context() context.Context
	Subject() string
	Reply() string
	Header() nats.Header
	Data() []byte
}

type RequesterMsg

type RequesterMsg interface {
	RequestMsg(ctx context.Context, msg *nats.Msg) (*nats.Msg, error)
}

type ServeMux

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

ServeMux is a handler multiplexer

func (*ServeMux) Handle

func (m *ServeMux) Handle(f Handler, subjects ...string) error

func (*ServeMux) HandleFunc

func (m *ServeMux) HandleFunc(f func(Publisher, Request) error, subjects ...string) error

func (*ServeMux) Serve

func (m *ServeMux) Serve(pub Publisher, req Request) error

type Server

type Server struct {
	Conn           Connection
	Handler        Handler
	BaseContext    context.Context
	Concurrency    int
	BufferSize     int
	QueueName      string
	ListenSubjects []string
	// contains filtered or unexported fields
}

func (*Server) Shutdown

func (s *Server) Shutdown()

func (*Server) Start

func (s *Server) Start() error

func (*Server) Wait

func (s *Server) Wait()

type Subscriber

type Subscriber interface {
	Subscribe(subj string) (Subscription, error)
	QueueSubscribe(subj, queue string) (Subscription, error)
}

type Subscription

type Subscription interface {
	Unsubscriber
	NextMsg(ctx context.Context) (*nats.Msg, error)
}

type TypedHandler

type TypedHandler[ArgT, ResT any] interface {
	Serve(TypedPublisher[ResT], TypedRequest[ArgT]) error
}

type TypedHandlerFunc

type TypedHandlerFunc[ArgT, ResT any] func(TypedPublisher[ResT], TypedRequest[ArgT]) error

func (TypedHandlerFunc[ArgT, ResT]) Serve

func (f TypedHandlerFunc[ArgT, ResT]) Serve(pub TypedPublisher[ResT], req TypedRequest[ArgT]) error

type TypedPublisher

type TypedPublisher[ResT any] interface {
	Header() *nats.Header
	Publish(*ResT) error
	Subject() string
	WithSubject(string) TypedPublisher[ResT]
}

type TypedRequest

type TypedRequest[T any] interface {
	Context() context.Context
	Header() nats.Header
	Payload() *T
}

type UIDGenerator

type UIDGenerator interface {
	Next() string
}

type Unsubscriber

type Unsubscriber interface {
	Unsubscribe() error
}