events

Jun 14, 2024

Changes in this version

+ type Errors []error

+ func (e *Errors) UnmarshalJSON(b []byte) error

+ func (e Errors) Error() string

+ func (e Errors) MarshalJSON() ([]byte, error)

Apr 2, 2024

Changes in this version

type ChangeMessage

+ func (m ChangeMessage) GetAddSubjects() []gidx.PrefixedID

+ func (m ChangeMessage) GetEventType() string

+ func (m ChangeMessage) GetSubject() gidx.PrefixedID

type EventMessage

+ func (m EventMessage) GetAddSubjects() []gidx.PrefixedID

+ func (m EventMessage) GetEventType() string

+ func (m EventMessage) GetSubject() gidx.PrefixedID

Aug 8, 2023

Changes in this version

type NATSConfig

+ SubscriberStartTime time.Time

Aug 4, 2023

Changes in this version

+ var ErrInvalidAuthRelationshipRequestAction = errors.New("auth relationship request message Action field must be write or delete")

+ var ErrMissingAuthRelationshipRequestObjectID = errors.New("auth relationship request message ObjectID field required")

+ var ErrMissingAuthRelationshipRequestRelation = errors.New("auth relationship request message Relations field required")

+ var ErrMissingAuthRelationshipRequestRelationRelation = errors.New("auth relationship request message Relations Relation field required")

+ var ErrMissingAuthRelationshipRequestRelationSubjectID = errors.New("auth relationship request message Relations SubjectID field required")

+ var ErrMissingChangeMessageEventType = errors.New("change message EventType field required")

+ var ErrMissingChangeMessageSubjectID = errors.New("change message SubjectID field required")

+ var ErrMissingEventMessageEventType = errors.New("event message EventType field required")

+ var ErrMissingEventMessageSubjectID = errors.New("event message SubjectID field required")

+ var ErrNATSInvalidAuthConfiguration = errors.New("invalid nats confinguration, both token and creds file are specified")

+ var ErrNATSInvalidDeliveryPolicy = errors.New("invalid delivery policy")

+ var ErrNATSMessageNoReplySubject = errors.New("unable to reply to auth relationship request, no reply subject specified")

+ var ErrProviderNotConfigured = errors.New("events provider not configured")

+ var ErrRequestNoResponders = errors.New("no responders for request")

+ var NATSDefaultConnectTimeout = 5 * time.Second

+ var NATSDefaultShutdownTimeout = 5 * time.Second

+ var NATSDefaultSubscriberFetchBackoff = 5 * time.Second

+ var NATSDefaultSubscriberFetchBatchSize = 20

+ var NATSDefaultSubscriberFetchTimeout = 5 * time.Second

+ func MustViperFlags(v *viper.Viper, flags *pflag.FlagSet, appName string)

+ func MustViperFlagsForNATS(v *viper.Viper, flags *pflag.FlagSet, appName string)

+ func NATSConsumerDurableName(queueGroup, subject string) string

+ type AuthRelationshipPublisher interface

+ PublishAuthRelationshipRequest func(ctx context.Context, topic string, message AuthRelationshipRequest) (Message[AuthRelationshipResponse], error)

+ type AuthRelationshipRelation struct

+ Relation string

+ SubjectID gidx.PrefixedID

+ func (r AuthRelationshipRelation) Validate() error

type AuthRelationshipRequest

+ Relations []AuthRelationshipRelation

+ func (m AuthRelationshipRequest) GetTraceContext(ctx context.Context) context.Context

+ func (m AuthRelationshipRequest) Validate() error

type AuthRelationshipResponse

+ func (m AuthRelationshipResponse) GetTraceContext(ctx context.Context) context.Context

+ func (m AuthRelationshipResponse) Validate() error

+ type AuthRelationshipSubscriber interface

+ SubscribeAuthRelationshipRequests func(ctx context.Context, topic string) (<-chan Request[AuthRelationshipRequest, AuthRelationshipResponse], error)

type ChangeMessage

+ func (m ChangeMessage) GetTraceContext(ctx context.Context) context.Context

+ func (m ChangeMessage) Validate() error

+ type Config struct

+ NATS NATSConfig

+ type Connection interface

+ Shutdown func(ctx context.Context) error

+ Source func() any

+ func NewConnection(config Config, options ...Option) (Connection, error)

type EventMessage

+ func (m EventMessage) GetTraceContext(ctx context.Context) context.Context

+ func (m EventMessage) Validate() error

+ type Message interface

+ Ack func() error

+ Connection func() Connection

+ Deliveries func() uint64

+ Error func() error

+ ID func() string

+ Message func() T

+ Nak func(delay time.Duration) error

+ Source func() any

+ Term func() error

+ Timestamp func() time.Time

+ Topic func() string

+ type NATSAuthRelationshipRequest struct

+ func (r *NATSAuthRelationshipRequest) Reply(ctx context.Context, message AuthRelationshipResponse) (Message[AuthRelationshipResponse], error)

type NATSConfig

+ ConnectTimeout time.Duration

+ PublishPrefix string

+ QueueGroup string

+ ShutdownTimeout time.Duration

+ Source string

+ SubscribePrefix string

+ SubscriberDeliveryPolicy string

+ SubscriberFetchBackoff time.Duration

+ SubscriberFetchBatchSize int

+ SubscriberFetchTimeout time.Duration

+ SubscriberNoAckExplicit bool

+ SubscriberNoManualAck bool

+ SubscriberStartSequence uint64

+ URL string

+ func (c NATSConfig) Configured() bool

+ func (c NATSConfig) Validate() error

+ func (c NATSConfig) WithDefaults() NATSConfig

+ type NATSConnection struct

+ func NewNATSConnection(config NATSConfig, options ...NATSOption) (*NATSConnection, error)

+ func (c *NATSConnection) PublishAuthRelationshipRequest(ctx context.Context, topic string, message AuthRelationshipRequest) (Message[AuthRelationshipResponse], error)

+ func (c *NATSConnection) PublishChange(ctx context.Context, topic string, message ChangeMessage) (Message[ChangeMessage], error)

+ func (c *NATSConnection) PublishEvent(ctx context.Context, topic string, message EventMessage) (Message[EventMessage], error)

+ func (c *NATSConnection) Shutdown(ctx context.Context) error

+ func (c *NATSConnection) Source() any

+ func (c *NATSConnection) SubscribeAuthRelationshipRequests(ctx context.Context, topic string) (<-chan Request[AuthRelationshipRequest, AuthRelationshipResponse], error)

+ func (c *NATSConnection) SubscribeChanges(ctx context.Context, topic string) (<-chan Message[ChangeMessage], error)

+ func (c *NATSConnection) SubscribeEvents(ctx context.Context, topic string) (<-chan Message[EventMessage], error)

+ type NATSMessage struct

+ func (m *NATSMessage[T]) Ack() error

+ func (m *NATSMessage[T]) Connection() Connection

+ func (m *NATSMessage[T]) Deliveries() uint64

+ func (m *NATSMessage[T]) Error() error

+ func (m *NATSMessage[T]) ID() string

+ func (m *NATSMessage[T]) Message() T

+ func (m *NATSMessage[T]) Nak(delay time.Duration) error

+ func (m *NATSMessage[T]) Source() any

+ func (m *NATSMessage[T]) Term() error

+ func (m *NATSMessage[T]) Timestamp() time.Time

+ func (m *NATSMessage[T]) Topic() string

+ type NATSOption func(c *NATSConfig) error

+ func WithNATSConnectOptions(options ...nats.Option) NATSOption

+ func WithNATSJetStreamOptions(options ...nats.JSOpt) NATSOption

+ func WithNATSLogger(logger *zap.SugaredLogger) NATSOption

+ func WithNATSSubscribeOptions(options ...nats.SubOpt) NATSOption

+ type Option func(config *Config) error

+ func WithLogger(logger *zap.SugaredLogger) Option

+ func WithNATSOptions(options ...NATSOption) Option

+ type Request interface

+ Reply func(ctx context.Context, message TResponse) (Message[TResponse], error)

Jul 20, 2023

Changes in this version

+ var DeleteAuthRelationshipAction AuthRelationshipAction = "delete"

+ var WriteAuthRelationshipAction AuthRelationshipAction = "write"

+ func TraceContextFromChangeMessage(ctx context.Context, msg ChangeMessage) context.Context

+ func TraceContextFromEventMessage(ctx context.Context, msg EventMessage) context.Context

+ type AuthRelationshipAction string

type AuthRelationshipRequest

+ Action AuthRelationshipAction

+ TraceContext map[string]string

type AuthRelationshipResponse

+ TraceContext map[string]string

type ChangeMessage

+ TraceContext map[string]string

type EventMessage

+ TraceContext map[string]string

Jul 18, 2023

Changes in this version

+ type AuthRelationshipRequest struct

+ ConditionName string

+ ConditionValues map[string]interface{}

+ ObjectID gidx.PrefixedID

+ RelationshipName string

+ SpanID string

+ SubjectID gidx.PrefixedID

+ TraceID string

+ func UnmarshalAuthRelationshipRequest(b []byte) (AuthRelationshipRequest, error)

+ type AuthRelationshipResponse struct

+ Errors []error

+ SpanID string

+ TraceID string

+ func UnmarshalAuthRelationshipResponse(b []byte) (AuthRelationshipResponse, error)

Jun 19, 2023

Changes in this version

type Publisher

+ func (p *Publisher) Close() error

type Subscriber

+ func (s *Subscriber) Close() error

May 23, 2023

Changes in this version

type Publisher

+ func NewPublisherWithLogger(cfg PublisherConfig, logger *zap.SugaredLogger) (*Publisher, error)

type Subscriber

+ func NewSubscriberWithLogger(cfg SubscriberConfig, logger *zap.SugaredLogger) (*Subscriber, error)

May 19, 2023

Changes in this version

+ var ErrMissingEventType = errors.New("event type missing")

+ var ErrUnsupportedPubsub = errors.New("unsupported pubsub provider")

+ func MustViperFlagsForPublisher(v *viper.Viper, flags *pflag.FlagSet, appName string)

+ func MustViperFlagsForSubscriber(v *viper.Viper, flags *pflag.FlagSet)

+ type ChangeMessage struct

+ ActorID gidx.PrefixedID

+ AdditionalData map[string]interface{}

+ AdditionalSubjectIDs []gidx.PrefixedID

+ EventType string

+ FieldChanges []FieldChange

+ Source string

+ SpanID string

+ SubjectFields map[string]string

+ SubjectID gidx.PrefixedID

+ Timestamp time.Time

+ TraceID string

+ func UnmarshalChangeMessage(b []byte) (ChangeMessage, error)

+ type ChangeType string

+ var CreateChangeType ChangeType = "create"

+ var DeleteChangeType ChangeType = "delete"

+ var UpdateChangeType ChangeType = "update"

+ type EventMessage struct

+ AdditionalSubjectIDs []gidx.PrefixedID

+ Data map[string]interface{}

+ EventType string

+ Source string

+ SpanID string

+ SubjectID gidx.PrefixedID

+ Timestamp time.Time

+ TraceID string

+ func UnmarshalEventMessage(b []byte) (EventMessage, error)

+ type FieldChange struct

+ CurrentValue string

+ Field string

+ PreviousValue string

+ type NATSConfig struct

+ CredsFile string

+ Token string

+ type Publisher struct

+ func NewPublisher(cfg PublisherConfig) (*Publisher, error)

+ func (p *Publisher) PublishChange(ctx context.Context, subjectType string, change ChangeMessage) error

+ func (p *Publisher) PublishEvent(_ context.Context, event EventMessage) error

+ type PublisherConfig struct

+ NATSConfig NATSConfig

+ Prefix string

+ Source string

+ Timeout time.Duration

+ URL string

+ type Subscriber struct

+ func NewSubscriber(cfg SubscriberConfig) (*Subscriber, error)

+ func (s *Subscriber) SubscribeChanges(ctx context.Context, topic string) (<-chan *message.Message, error)

+ func (s *Subscriber) SubscribeEvents(ctx context.Context, topic string) (<-chan *message.Message, error)

+ type SubscriberConfig struct

+ NATSConfig NATSConfig

+ Prefix string

+ QueueGroup string

+ Timeout time.Duration

+ URL string