-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
rewrite events handling to remove watermill (#130)
As discussed in the community call, watermill doesn't give us the necessary features we'd like to utilize with the underlying nats message. We decided to switch to using nats directly but still wanted some support for possibly changing this out later. This rewrites events to use our own interfaces to allow for the possibility of a different event driver later. Additionally this switches to using pull subscriptions instead of push, supports Ack, Nak and Term as well as Request/Reply semantics. Due to the Request/Reply semantics, no longer are there separate Publisher and Subscriber configurations as the driver needs to be able to handle both. --------- Signed-off-by: Mike Mason <[email protected]>
- Loading branch information
Showing
22 changed files
with
1,584 additions
and
688 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| package events | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "go.uber.org/multierr" | ||
| ) | ||
|
|
||
| // Connection defines a connection handler. | ||
| type Connection interface { | ||
| // Gracefully close the connection. | ||
| Shutdown(ctx context.Context) error | ||
|
|
||
| // Source gives you the raw underlying connection object. | ||
| Source() any | ||
|
|
||
| Subscriber | ||
| Publisher | ||
|
|
||
| AuthRelationshipSubscriber | ||
| AuthRelationshipPublisher | ||
| } | ||
|
|
||
| // Subscriber specifies subscriber methods. | ||
| type Subscriber interface { | ||
| // SubscribeChanges subscribes to the provided topic responding with an ChangeMessage message. | ||
| SubscribeChanges(ctx context.Context, topic string) (<-chan Message[ChangeMessage], error) | ||
| // SubscribeEvents subscribes to the provided topic responding with an EventMessage message. | ||
| SubscribeEvents(ctx context.Context, topic string) (<-chan Message[EventMessage], error) | ||
| } | ||
|
|
||
| // Publisher specifies publisher methods. | ||
| type Publisher interface { | ||
| // PublishChange publishes to the specified topic with the message given. | ||
| PublishChange(ctx context.Context, topic string, message ChangeMessage) (Message[ChangeMessage], error) | ||
| // PublishEvent publishes to the specified topic with the message given. | ||
| PublishEvent(ctx context.Context, topic string, message EventMessage) (Message[EventMessage], error) | ||
| } | ||
|
|
||
| // AuthRelationshipSubscriber specifies the auth relationship subscriber methods. | ||
| type AuthRelationshipSubscriber interface { | ||
| // SubscribeAuthRelationshipRequests subscribes to the provided topic responding with an AuthRelationshipRequest message. | ||
| SubscribeAuthRelationshipRequests(ctx context.Context, topic string) (<-chan Request[AuthRelationshipRequest, AuthRelationshipResponse], error) | ||
| } | ||
|
|
||
| // AuthRelationshipPublisher specifies the auth relationship publisher methods. | ||
| type AuthRelationshipPublisher interface { | ||
| // PublishAuthRelationshipRequest publishes to the specified topic with the message given. | ||
| PublishAuthRelationshipRequest(ctx context.Context, topic string, message AuthRelationshipRequest) (Message[AuthRelationshipResponse], error) | ||
| } | ||
|
|
||
| // NewConnection creates a new Connection from the provided config. | ||
| func NewConnection(config Config, options ...Option) (Connection, error) { | ||
| var err error | ||
|
|
||
| for _, opt := range options { | ||
| err = multierr.Append(err, opt(&config)) | ||
| } | ||
|
|
||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| if config.NATS.Configured() { | ||
| return NewNATSConnection(config.NATS) | ||
| } | ||
|
|
||
| return nil, ErrProviderNotConfigured | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| package events | ||
|
|
||
| import "errors" | ||
|
|
||
| var ( | ||
| // ErrProviderNotConfigured is an error packages should return if no events provider is configured. | ||
| ErrProviderNotConfigured = errors.New("events provider not configured") | ||
|
|
||
| // ErrMissingChangeMessageEventType is returned when the event message has the incorrect field EventType value. | ||
| ErrMissingChangeMessageEventType = errors.New("change message EventType field required") | ||
| // ErrMissingChangeMessageSubjectID is returned when the event message has the incorrect field SubjectID value. | ||
| ErrMissingChangeMessageSubjectID = errors.New("change message SubjectID field required") | ||
|
|
||
| // ErrMissingEventMessageEventType is returned when the event message has the incorrect field EventType value. | ||
| ErrMissingEventMessageEventType = errors.New("event message EventType field required") | ||
| // ErrMissingEventMessageSubjectID is returned when the event message has the incorrect field SubjectID value. | ||
| ErrMissingEventMessageSubjectID = errors.New("event message SubjectID field required") | ||
|
|
||
| // ErrInvalidAuthRelationshipRequestAction is returned when the event message has the incorrect field Action value. | ||
| ErrInvalidAuthRelationshipRequestAction = errors.New("auth relationship request message Action field must be write or delete") | ||
| // ErrMissingAuthRelationshipRequestObjectID is returned when the event message has the incorrect field ObjectID value. | ||
| ErrMissingAuthRelationshipRequestObjectID = errors.New("auth relationship request message ObjectID field required") | ||
| // ErrMissingAuthRelationshipRequestRelation is returned when the event message has no relations defined. | ||
| ErrMissingAuthRelationshipRequestRelation = errors.New("auth relationship request message Relations field required") | ||
| // ErrMissingAuthRelationshipRequestRelationRelation is returned when the event message Relations has the incorrect field for Relation value. | ||
| ErrMissingAuthRelationshipRequestRelationRelation = errors.New("auth relationship request message Relations Relation field required") | ||
| // ErrMissingAuthRelationshipRequestRelationSubjectID is returned when the event message Relations has the incorrect field SubjectID value. | ||
| ErrMissingAuthRelationshipRequestRelationSubjectID = errors.New("auth relationship request message Relations SubjectID field required") | ||
|
|
||
| // ErrRequestNoResponders is returned when a request is attempted but no responder is listening. | ||
| ErrRequestNoResponders = errors.New("no responders for request") | ||
| ) |
Oops, something went wrong.