ncl

Dec 16, 2024

Changes in this version

+ const BacErrorMessageType

+ const DefaultResponderProcessingTimeout

+ const KeyErrorCode

+ const KeyStatusCode

+ var ErrHandlerExists = errors.New("handler already exists for message type")

+ var ErrNoHandler = errors.New("no handler found for message type")

+ func BacErrorToEnvelope(err bacerrors.Error) *envelope.Message

type MockOrderedPublisher

+ func (m *MockOrderedPublisher) Request(ctx context.Context, request PublishRequest) (*envelope.Message, error)

type MockOrderedPublisherMockRecorder

+ func (mr *MockOrderedPublisherMockRecorder) Request(ctx, request interface{}) *gomock.Call

+ type MockProcessingNotifier struct

+ func NewMockProcessingNotifier(ctrl *gomock.Controller) *MockProcessingNotifier

+ func (m *MockProcessingNotifier) EXPECT() *MockProcessingNotifierMockRecorder

+ func (m *MockProcessingNotifier) OnProcessed(ctx context.Context, message *envelope.Message)

+ type MockProcessingNotifierMockRecorder struct

+ func (mr *MockProcessingNotifierMockRecorder) OnProcessed(ctx, message interface{}) *gomock.Call

type MockPublisher

+ func (m *MockPublisher) Request(ctx context.Context, request PublishRequest) (*envelope.Message, error)

type MockPublisherMockRecorder

+ func (mr *MockPublisherMockRecorder) Request(ctx, request interface{}) *gomock.Call

+ type MockRequestHandler struct

+ func NewMockRequestHandler(ctrl *gomock.Controller) *MockRequestHandler

+ func (m *MockRequestHandler) EXPECT() *MockRequestHandlerMockRecorder

+ func (m *MockRequestHandler) HandleRequest(ctx context.Context, message *envelope.Message) (*envelope.Message, error)

+ type MockRequestHandlerMockRecorder struct

+ func (mr *MockRequestHandlerMockRecorder) HandleRequest(ctx, message interface{}) *gomock.Call

+ type MockResponder struct

+ func NewMockResponder(ctrl *gomock.Controller) *MockResponder

+ func (m *MockResponder) Close(ctx context.Context) error

+ func (m *MockResponder) EXPECT() *MockResponderMockRecorder

+ func (m *MockResponder) Listen(ctx context.Context, messageType string, handler RequestHandler) error

+ type MockResponderMockRecorder struct

+ func (mr *MockResponderMockRecorder) Close(ctx interface{}) *gomock.Call

+ func (mr *MockResponderMockRecorder) Listen(ctx, messageType, handler interface{}) *gomock.Call

+ type NoopNotifier struct

+ func (n *NoopNotifier) OnProcessed(ctx context.Context, message *envelope.Message)

+ type ProcessingNotifier interface

+ OnProcessed func(ctx context.Context, message *envelope.Message)

type Publisher

+ Request func(ctx context.Context, request PublishRequest) (*envelope.Message, error)

+ type RequestHandler interface

+ HandleRequest func(ctx context.Context, message *envelope.Message) (*envelope.Message, error)

+ type RequestHandlerFunc func(ctx context.Context, message *envelope.Message) (*envelope.Message, error)

+ func (f RequestHandlerFunc) HandleRequest(ctx context.Context, message *envelope.Message) (*envelope.Message, error)

+ type Responder interface

+ Close func(ctx context.Context) error

+ Listen func(ctx context.Context, messageType string, handler RequestHandler) error

+ func NewResponder(nc *nats.Conn, config ResponderConfig) (Responder, error)

+ type ResponderConfig struct

+ MessageRegistry *envelope.Registry

+ MessageSerializer envelope.MessageSerializer

+ Name string

+ ProcessingTimeout time.Duration

+ Subject string

+ func DefaultResponderConfig() ResponderConfig

+ func (c *ResponderConfig) Validate() error

type SubscriberConfig

+ ProcessingNotifier ProcessingNotifier

Nov 29, 2024

Changes in this version

+ const DefaultBackoffInitialDelay

+ const DefaultBackoffMaximumDelay

+ const DefaultProcessingTimeout

+ const KeyMessageUUID

+ const KeySubject

+ func Ack(m *nats.Msg) error

+ func Nack(m *nats.Msg, err error) error

+ func NackWithDelay(m *nats.Msg, err error, delay time.Duration) error

+ type AckMode int

+ const ExplicitAck

+ const NoAck

+ type MockCheckpointer struct

+ func NewMockCheckpointer(ctrl *gomock.Controller) *MockCheckpointer

+ func (m *MockCheckpointer) Checkpoint(ctx context.Context, message *envelope.Message) error

+ func (m *MockCheckpointer) EXPECT() *MockCheckpointerMockRecorder

+ func (m *MockCheckpointer) GetLastCheckpoint() (int64, error)

+ type MockCheckpointerMockRecorder struct

+ func (mr *MockCheckpointerMockRecorder) Checkpoint(ctx, message interface{}) *gomock.Call

+ func (mr *MockCheckpointerMockRecorder) GetLastCheckpoint() *gomock.Call

+ type MockMessageFilter struct

+ func NewMockMessageFilter(ctrl *gomock.Controller) *MockMessageFilter

+ func (m *MockMessageFilter) EXPECT() *MockMessageFilterMockRecorder

+ func (m *MockMessageFilter) ShouldFilter(metadata *envelope.Metadata) bool

+ type MockMessageFilterMockRecorder struct

+ func (mr *MockMessageFilterMockRecorder) ShouldFilter(metadata interface{}) *gomock.Call

+ type MockMessageHandler struct

+ func NewMockMessageHandler(ctrl *gomock.Controller) *MockMessageHandler

+ func (m *MockMessageHandler) EXPECT() *MockMessageHandlerMockRecorder

+ func (m *MockMessageHandler) HandleMessage(ctx context.Context, message *envelope.Message) error

+ func (m *MockMessageHandler) ShouldProcess(ctx context.Context, message *envelope.Message) bool

+ type MockMessageHandlerMockRecorder struct

+ func (mr *MockMessageHandlerMockRecorder) HandleMessage(ctx, message interface{}) *gomock.Call

+ func (mr *MockMessageHandlerMockRecorder) ShouldProcess(ctx, message interface{}) *gomock.Call

+ type MockOrderedPublisher struct

+ func NewMockOrderedPublisher(ctrl *gomock.Controller) *MockOrderedPublisher

+ func (m *MockOrderedPublisher) Close(ctx context.Context) error

+ func (m *MockOrderedPublisher) EXPECT() *MockOrderedPublisherMockRecorder

+ func (m *MockOrderedPublisher) Publish(ctx context.Context, request PublishRequest) error

+ func (m *MockOrderedPublisher) PublishAsync(ctx context.Context, request PublishRequest) (PubFuture, error)

+ func (m *MockOrderedPublisher) Reset(ctx context.Context)

+ type MockOrderedPublisherMockRecorder struct

+ func (mr *MockOrderedPublisherMockRecorder) Close(ctx interface{}) *gomock.Call

+ func (mr *MockOrderedPublisherMockRecorder) Publish(ctx, request interface{}) *gomock.Call

+ func (mr *MockOrderedPublisherMockRecorder) PublishAsync(ctx, request interface{}) *gomock.Call

+ func (mr *MockOrderedPublisherMockRecorder) Reset(ctx interface{}) *gomock.Call

+ type MockPubFuture struct

+ func NewMockPubFuture(ctrl *gomock.Controller) *MockPubFuture

+ func (m *MockPubFuture) Done() <-chan struct{}

+ func (m *MockPubFuture) EXPECT() *MockPubFutureMockRecorder

+ func (m *MockPubFuture) Err() error

+ func (m *MockPubFuture) Msg() *nats.Msg

+ func (m *MockPubFuture) Result() *Result

+ func (m *MockPubFuture) Wait(ctx context.Context) error

+ type MockPubFutureMockRecorder struct

+ func (mr *MockPubFutureMockRecorder) Done() *gomock.Call

+ func (mr *MockPubFutureMockRecorder) Err() *gomock.Call

+ func (mr *MockPubFutureMockRecorder) Msg() *gomock.Call

+ func (mr *MockPubFutureMockRecorder) Result() *gomock.Call

+ func (mr *MockPubFutureMockRecorder) Wait(ctx interface{}) *gomock.Call

+ type MockPublisher struct

+ func NewMockPublisher(ctrl *gomock.Controller) *MockPublisher

+ func (m *MockPublisher) EXPECT() *MockPublisherMockRecorder

+ func (m *MockPublisher) Publish(ctx context.Context, request PublishRequest) error

+ type MockPublisherMockRecorder struct

+ func (mr *MockPublisherMockRecorder) Publish(ctx, request interface{}) *gomock.Call

+ type MockSubscriber struct

+ func NewMockSubscriber(ctrl *gomock.Controller) *MockSubscriber

+ func (m *MockSubscriber) Close(ctx context.Context) error

+ func (m *MockSubscriber) EXPECT() *MockSubscriberMockRecorder

+ func (m *MockSubscriber) Subscribe(ctx context.Context, subjects ...string) error

+ type MockSubscriberMockRecorder struct

+ func (mr *MockSubscriberMockRecorder) Close(ctx interface{}) *gomock.Call

+ func (mr *MockSubscriberMockRecorder) Subscribe(ctx interface{}, subjects ...interface{}) *gomock.Call

+ type OrderedPublisher interface

+ Close func(ctx context.Context) error

+ PublishAsync func(ctx context.Context, request PublishRequest) (PubFuture, error)

+ Reset func(ctx context.Context)

+ func NewOrderedPublisher(nc *nats.Conn, config OrderedPublisherConfig) (OrderedPublisher, error)

+ type OrderedPublisherConfig struct

+ AckMode AckMode

+ AckWait time.Duration

+ Destination string

+ DestinationPrefix string

+ MaxPending int

+ MessageRegistry *envelope.Registry

+ MessageSerializer envelope.MessageSerializer

+ Name string

+ RetryAttempts int

+ RetryWait time.Duration

+ func DefaultOrderedPublisherConfig() OrderedPublisherConfig

+ func (c *OrderedPublisherConfig) Validate() error

+ type PubFuture interface

+ Done func() <-chan struct{}

+ Err func() error

+ Msg func() *nats.Msg

+ Result func() *Result

+ Wait func(ctx context.Context) error

+ type PublishRequest struct

+ Message *envelope.Message

+ Subject string

+ SubjectPrefix string

+ func NewPublishRequest(message *envelope.Message) PublishRequest

+ func (r PublishRequest) WithSubject(subject string) PublishRequest

+ func (r PublishRequest) WithSubjectPrefix(prefix string) PublishRequest

+ type PublisherConfig struct

+ Destination string

+ DestinationPrefix string

+ MessageRegistry *envelope.Registry

+ MessageSerializer envelope.MessageSerializer

+ Name string

+ func DefaultPublisherConfig() PublisherConfig

+ func (c *PublisherConfig) Validate() error

+ type Result struct

+ Delay int64

+ Error string

+ func NewResult() *Result

+ func ParseResult(msg *nats.Msg) (*Result, error)

+ func (r *Result) DelayDuration() time.Duration

+ func (r *Result) Err() error

+ func (r *Result) WithDelay(delay time.Duration) *Result

+ func (r *Result) WithError(err error) *Result

+ type SubscriberConfig struct

+ Backoff backoff.Backoff

+ Checkpointer Checkpointer

+ MessageFilter MessageFilter

+ MessageHandler MessageHandler

+ MessageRegistry *envelope.Registry

+ MessageSerializer envelope.MessageSerializer

+ Name string

+ ProcessingTimeout time.Duration

+ func DefaultSubscriberConfig() SubscriberConfig

+ func (c *SubscriberConfig) Validate() error

Oct 7, 2024

Changes in this version

+ const CRCSize

+ const DefaultPayloadEncoding

+ const ErrEmptyData

+ const ErrMsgCRCFailed

+ const ErrMsgTooShort

+ const ErrNilMessage

+ const HeaderSize

+ const JSONPayloadSerDeType

+ const KeyEventTime

+ const KeyMessageID

+ const KeyMessageType

+ const KeyPayloadEncoding

+ const KeySource

+ const VersionSize

+ func NewErrDeserializationFailed(encoding string, err error) error

+ func NewErrSerializationFailed(encoding string, err error) error

+ func NewErrUnexpectedPayloadType(expected, actual string) error

+ type Checkpointer interface

+ Checkpoint func(message *Message) error

+ GetLastCheckpoint func() (int64, error)

+ type EnvelopedRawMessageSerDe struct

+ func NewEnvelopedRawMessageSerDe() *EnvelopedRawMessageSerDe

+ func (v *EnvelopedRawMessageSerDe) Deserialize(data []byte) (*RawMessage, error)

+ func (v *EnvelopedRawMessageSerDe) Serialize(msg *RawMessage) ([]byte, error)

+ func (v *EnvelopedRawMessageSerDe) WithSerializationVersion(version SchemaVersion) *EnvelopedRawMessageSerDe

+ type ErrBadMessage struct

+ Reason string

+ func NewErrBadMessage(reason string) *ErrBadMessage

+ func (e *ErrBadMessage) Error() string

+ type ErrBadPayload struct

+ Reason string

+ func NewErrBadPayload(reason string) *ErrBadPayload

+ func (e *ErrBadPayload) Error() string

+ type ErrDeserializationFailed struct

+ Encoding string

+ Err error

+ func (e *ErrDeserializationFailed) Error() string

+ func (e *ErrDeserializationFailed) Unwrap() error

+ type ErrSerializationFailed struct

+ Encoding string

+ Err error

+ func (e *ErrSerializationFailed) Error() string

+ func (e *ErrSerializationFailed) Unwrap() error

+ type ErrUnexpectedPayloadType struct

+ Actual string

+ Expected string

+ func (e *ErrUnexpectedPayloadType) Error() string

+ type ErrUnsupportedEncoding struct

+ Encoding string

+ func NewErrUnsupportedEncoding(encoding string) *ErrUnsupportedEncoding

+ func (e *ErrUnsupportedEncoding) Error() string

+ type ErrUnsupportedMessageType struct

+ Type string

+ func NewErrUnsupportedMessageType(messageType string) *ErrUnsupportedMessageType

+ func (e *ErrUnsupportedMessageType) Error() string

+ type JSONMessageSerDe struct

+ func (j *JSONMessageSerDe) Deserialize(rMsg *RawMessage, payloadType reflect.Type) (*Message, error)

+ func (j *JSONMessageSerDe) Serialize(message *Message) (*RawMessage, error)

+ type JSONRawMessageSerializer struct

+ func (j *JSONRawMessageSerializer) Deserialize(data []byte) (*RawMessage, error)

+ func (j *JSONRawMessageSerializer) Serialize(msg *RawMessage) ([]byte, error)

+ type Message struct

+ Metadata *Metadata

+ Payload any

+ func NewMessage(payload any) *Message

+ func (m *Message) GetPayload(t interface{}) (interface{}, bool)

+ func (m *Message) IsType(t interface{}) bool

+ func (m *Message) WithMetadata(metadata *Metadata) *Message

+ func (m *Message) WithMetadataValue(key, value string) *Message

+ type MessageFilter interface

+ ShouldFilter func(metadata *Metadata) bool

+ type MessageFilterFunc func(metadata *Metadata) bool

+ func (f MessageFilterFunc) ShouldFilter(metadata *Metadata) bool

+ type MessageHandler interface

+ HandleMessage func(ctx context.Context, message *Message) error

+ ShouldProcess func(ctx context.Context, message *Message) bool

+ type MessageHandlerFunc func(ctx context.Context, message *Message) error

+ func (f MessageHandlerFunc) HandleMessage(ctx context.Context, message *Message) error

+ func (f MessageHandlerFunc) ShouldProcess(ctx context.Context, message *Message) bool

+ type MessageSerDe interface

+ Deserialize func(rawMessage *RawMessage, payloadType reflect.Type) (*Message, error)

+ Serialize func(message *Message) (*RawMessage, error)

+ type MessageSerDeRegistry struct

+ func NewMessageSerDeRegistry() *MessageSerDeRegistry

+ func (r *MessageSerDeRegistry) Deserialize(rawMessage *RawMessage) (*Message, error)

+ func (r *MessageSerDeRegistry) Register(name string, payload any) error

+ func (r *MessageSerDeRegistry) Serialize(message *Message) (*RawMessage, error)

+ type Metadata map[string]string

+ func NewMetadataFromMap(m map[string]string) *Metadata

+ func NewMetadataFromMapCopy(m map[string]string) *Metadata

+ func (m Metadata) Get(key string) string

+ func (m Metadata) GetInt(key string) int

+ func (m Metadata) GetInt64(key string) int64

+ func (m Metadata) GetTime(key string) time.Time

+ func (m Metadata) Has(key string) bool

+ func (m Metadata) Set(key, value string)

+ func (m Metadata) SetInt(key string, value int)

+ func (m Metadata) SetInt64(key string, value int64)

+ func (m Metadata) SetTime(key string, value time.Time)

+ func (m Metadata) ToMap() map[string]string

+ type NoopCheckpointer struct

+ func (n *NoopCheckpointer) Checkpoint(_ *Message) error

+ func (n *NoopCheckpointer) GetLastCheckpoint() (int64, error)

+ type NoopMessageFilter struct

+ func (n NoopMessageFilter) ShouldFilter(_ *Metadata) bool

+ type ProtoSerializer struct

+ func (p *ProtoSerializer) Deserialize(data []byte) (*RawMessage, error)

+ func (p *ProtoSerializer) Serialize(msg *RawMessage) ([]byte, error)

+ type Publisher interface

+ Publish func(ctx context.Context, message *Message) error

+ func NewPublisher(nc *nats.Conn, opts ...PublisherOption) (Publisher, error)

+ type PublisherOption func(*publisher)

+ func WithPublisherDestination(destination string) PublisherOption

+ func WithPublisherDestinationPrefix(prefix string) PublisherOption

+ func WithPublisherMessageSerDeRegistry(registry *MessageSerDeRegistry) PublisherOption

+ func WithPublisherMessageSerializer(serializer RawMessageSerDe) PublisherOption

+ func WithPublisherName(name string) PublisherOption

+ type RawMessage struct

+ Metadata *Metadata

+ Payload []byte

+ type RawMessageSerDe interface

+ Deserialize func([]byte) (*RawMessage, error)

+ Serialize func(*RawMessage) ([]byte, error)

+ type SchemaVersion byte

+ const DefaultSchemaVersion

+ const SchemaVersionJSONV1

+ const SchemaVersionProtobufV1

+ func (v SchemaVersion) String() string

+ type Subscriber interface

+ Close func(ctx context.Context) error

+ Subscribe func(subjects ...string) error

+ func NewSubscriber(nc *nats.Conn, opts ...SubscriberOption) (Subscriber, error)

+ type SubscriberOption func(*subscriber)

+ func WithSubscriberCheckpointer(checkpointer Checkpointer) SubscriberOption

+ func WithSubscriberMessageDeserializer(deserializer RawMessageSerDe) SubscriberOption

+ func WithSubscriberMessageFilter(filter MessageFilter) SubscriberOption

+ func WithSubscriberMessageHandlers(handlers ...MessageHandler) SubscriberOption

+ func WithSubscriberMessageSerDeRegistry(registry *MessageSerDeRegistry) SubscriberOption