-
Notifications
You must be signed in to change notification settings - Fork 37
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #45 from weaveworks/28_events
feat: event bus implementation
- Loading branch information
Showing
10 changed files
with
486 additions
and
0 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,13 @@ | ||
| package events | ||
|
|
||
| import "fmt" | ||
|
|
||
| // ErrTopicNotFound is an error created when a topic with a specific name isn't found. | ||
| type ErrTopicNotFound struct { | ||
| Name string | ||
| } | ||
|
|
||
| // Error returns the error message. | ||
| func (e ErrTopicNotFound) Error() string { | ||
| return fmt.Sprintf("topic %s not found", e.Name) | ||
| } |
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,41 @@ | ||
| package events | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/google/uuid" | ||
| ) | ||
|
|
||
| // Handler represents an event handling function. | ||
| type Handler func(e *Envelope) | ||
|
|
||
| // ErrorHandler represents an error handling function. | ||
| type ErrorHandler func(err error) | ||
|
|
||
| // Handlers represents a pair of event/error handlers. | ||
| type Handlers struct { | ||
| // Event is the event handler function. | ||
| Event Handler | ||
| // Error is the error handler function. | ||
| Error ErrorHandler | ||
| } | ||
|
|
||
| // EventBus is the interface that an event bus must implement. | ||
| type EventBus interface { | ||
| // CreateTopic will create a named topic (a.k.a channel or queue) for events. | ||
| CreateTopic(ctx context.Context, topic string) error | ||
| // Publish will publish an event to a specific topic. | ||
| Publish(ctx context.Context, topic string, event interface{}) error | ||
| // Subscribe will subscribe to events on a named topic and will call the relevant handlers. | ||
| Subscribe(ctx context.Context, topic string, handlers Handlers) error | ||
| } | ||
|
|
||
| // Envelope represents an event envelope. | ||
| type Envelope struct { | ||
| // ID is the unique identifier for the event. | ||
| ID uuid.UUID `json:"id"` | ||
| // Topic is the name of the topic the event originated from. | ||
| Topic string `json:"topic"` | ||
| // Event is the actual event payload. | ||
| Event interface{} `json:"event"` | ||
| } |
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,3 @@ | ||
| package mock | ||
|
|
||
| //go:generate ../../../hack/tools/bin/mockgen -destination mock.go -package mock github.com/weaveworks/reignite/pkg/events EventBus |
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
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,9 @@ | ||
| package transport | ||
|
|
||
| import "errors" | ||
|
|
||
| var ( | ||
| errHandlerRequired = errors.New("event handler is required") | ||
| errErrorHandlerRequired = errors.New("error handler is required") | ||
| errTopicRequired = errors.New("topic name is required") | ||
| ) |
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,101 @@ | ||
| package transport | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/vmware/transport-go/bus" | ||
| "github.com/vmware/transport-go/model" | ||
|
|
||
| "github.com/weaveworks/reignite/pkg/events" | ||
| ) | ||
|
|
||
| // New creates a new event bus based on Transport (https://vmware.github.io/transport/). | ||
| func New() events.EventBus { | ||
| return &transportEvents{ | ||
| eventBus: bus.GetBus(), | ||
| } | ||
| } | ||
|
|
||
| type transportEvents struct { | ||
| eventBus bus.EventBus | ||
| } | ||
|
|
||
| // CreateTopic will create a named topic (a.k.a channel or queue) for events. | ||
| func (te *transportEvents) CreateTopic(ctx context.Context, topic string) error { | ||
| if topic == "" { | ||
| return errTopicRequired | ||
| } | ||
|
|
||
| manager := te.eventBus.GetChannelManager() | ||
|
|
||
| if !manager.CheckChannelExists(topic) { | ||
| manager.CreateChannel(topic) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // Publish will publish an event to a specific topic. | ||
| func (te *transportEvents) Publish(ctx context.Context, topic string, event interface{}) error { | ||
| if topic == "" { | ||
| return errTopicRequired | ||
| } | ||
|
|
||
| manager := te.eventBus.GetChannelManager() | ||
|
|
||
| if !manager.CheckChannelExists(topic) { | ||
| return events.ErrTopicNotFound{Name: topic} | ||
| } | ||
|
|
||
| if err := te.eventBus.SendRequestMessage(topic, event, nil); err != nil { | ||
| return fmt.Errorf("sending message to channel: %w", err) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // Subscribe will subscribe to events on a named topic and will call the relevant handler. | ||
| func (te *transportEvents) Subscribe(ctx context.Context, topic string, handlers events.Handlers) error { | ||
| if handlers.Event == nil { | ||
| return errHandlerRequired | ||
| } | ||
| if handlers.Error == nil { | ||
| return errErrorHandlerRequired | ||
| } | ||
| if topic == "" { | ||
| return errTopicRequired | ||
| } | ||
|
|
||
| manager := te.eventBus.GetChannelManager() | ||
|
|
||
| if !manager.CheckChannelExists(topic) { | ||
| return events.ErrTopicNotFound{Name: topic} | ||
| } | ||
|
|
||
| h, err := te.eventBus.ListenRequestStream(topic) | ||
| if err != nil { | ||
| return fmt.Errorf("listening for transport events: %w", err) | ||
| } | ||
| h.Handle(te.subsciberHandler(handlers.Event), te.subsciberErrorHandler(handlers.Error)) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (te *transportEvents) subsciberHandler(handler events.Handler) bus.MessageHandlerFunction { | ||
| return func(msg *model.Message) { | ||
| evt := &events.Envelope{ | ||
| ID: *msg.Id, | ||
| Topic: msg.Channel, | ||
| Event: msg.Payload, | ||
| } | ||
|
|
||
| handler(evt) | ||
| } | ||
| } | ||
|
|
||
| func (te *transportEvents) subsciberErrorHandler(errHandler events.ErrorHandler) bus.MessageErrorFunction { | ||
| return func(err error) { | ||
| errHandler(err) | ||
| } | ||
| } |
Oops, something went wrong.