-
Notifications
You must be signed in to change notification settings - Fork 5
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 #63 from igmagollo/feature/add-outbox-storer
feat(outbox): add outbox storer with observability instrumented
- Loading branch information
Showing
36 changed files
with
5,164 additions
and
34 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,5 @@ | ||
| test: | ||
| test-example: | ||
| @go run github.com/onsi/ginkgo/v2/ginkgo -v ./example/testing | ||
|
|
||
| test: | ||
| @go run github.com/onsi/ginkgo/v2/ginkgo -v -p --race ./tests |
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,171 @@ | ||
| package outbox | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "reflect" | ||
|
|
||
| "github.com/TheRafaBonin/roxy" | ||
| "github.com/gothunder/thunder/internal/utils" | ||
| ) | ||
|
|
||
| const ( | ||
| SetTopic = "SetTopic" | ||
| SetHeaders = "SetHeaders" | ||
| SetPayload = "SetPayload" | ||
| Exec = "Exec" | ||
|
|
||
| Create = "Create" | ||
| CreateBulk = "CreateBulk" | ||
| ) | ||
|
|
||
| var ( | ||
| ErrNilClient = errors.New("nil OutboxMessageClient") | ||
| ) | ||
|
|
||
| type MessageClient interface { | ||
| Create() MessageCreator | ||
| CreateBulk(messageCreators ...MessageCreator) MessageBulkCreator | ||
| } | ||
|
|
||
| type MessageCreator interface { | ||
| SetTopic(topic string) MessageCreator | ||
| SetPayload(payload []byte) MessageCreator | ||
| SetHeaders(headers map[string]string) MessageCreator | ||
| Exec(ctx context.Context) error | ||
| Unwrap() interface{} | ||
| } | ||
|
|
||
| type MessageBulkCreator interface { | ||
| Exec(ctx context.Context) error | ||
| } | ||
|
|
||
| type outboxMessageCreateWrapper struct { | ||
| outboxMessageCreate interface{} | ||
| } | ||
|
|
||
| func (omcw *outboxMessageCreateWrapper) SetHeaders(headers map[string]string) MessageCreator { | ||
| _, err := utils.SafeCallMethod(omcw.outboxMessageCreate, SetHeaders, []reflect.Value{ | ||
| reflect.ValueOf(headers), | ||
| }) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| return omcw | ||
| } | ||
|
|
||
| func (omcw *outboxMessageCreateWrapper) SetPayload(payload []byte) MessageCreator { | ||
| _, err := utils.SafeCallMethod(omcw.outboxMessageCreate, SetPayload, []reflect.Value{ | ||
| reflect.ValueOf(payload), | ||
| }) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| return omcw | ||
| } | ||
|
|
||
| func (omcw *outboxMessageCreateWrapper) SetTopic(topic string) MessageCreator { | ||
| _, err := utils.SafeCallMethod(omcw.outboxMessageCreate, SetTopic, []reflect.Value{ | ||
| reflect.ValueOf(topic), | ||
| }) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| return omcw | ||
| } | ||
|
|
||
| func (omcw *outboxMessageCreateWrapper) Exec(ctx context.Context) error { | ||
| results, err := utils.SafeCallMethod(omcw.outboxMessageCreate, Exec, []reflect.Value{ | ||
| reflect.ValueOf(ctx), | ||
| }) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return results[0].Interface().(error) | ||
| } | ||
|
|
||
| func (omcw *outboxMessageCreateWrapper) Unwrap() interface{} { | ||
| return omcw.outboxMessageCreate | ||
| } | ||
|
|
||
| func WrapOutboxMessageCreate(omc interface{}) (MessageCreator, error) { | ||
| handleError := func(methodName string) (MessageCreator, error) { | ||
| return nil, roxy.Wrap(utils.ErrMethodNotFound, methodName) | ||
| } | ||
|
|
||
| methods := []string{SetTopic, SetHeaders, SetPayload, Exec} | ||
| for _, method := range methods { | ||
| if !utils.HasMethod(omc, method) { | ||
| return handleError(method) | ||
| } | ||
| } | ||
|
|
||
| return &outboxMessageCreateWrapper{outboxMessageCreate: omc}, nil | ||
| } | ||
|
|
||
| type MessageBuilderInterface[T any] interface { | ||
| SetHeaders(headers map[string]string) T | ||
| SetPayload(payload []byte) T | ||
| SetTopic(topic string) T | ||
| Exec(ctx context.Context) error | ||
| } | ||
|
|
||
| type outboxMessageClientWrapper struct { | ||
| OutboxMessageClient interface{} | ||
| } | ||
|
|
||
| func (c *outboxMessageClientWrapper) Create() MessageCreator { | ||
| results, err := utils.SafeCallMethod(c.OutboxMessageClient, Create, []reflect.Value{}) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| creator, _ := WrapOutboxMessageCreate(results[0].Interface()) | ||
| return creator | ||
| } | ||
|
|
||
| func (c *outboxMessageClientWrapper) CreateBulk(messageCreators ...MessageCreator) MessageBulkCreator { | ||
| creators := make([]reflect.Value, len(messageCreators)) | ||
| for i, mc := range messageCreators { | ||
| creators[i] = reflect.ValueOf(mc.Unwrap()) | ||
| } | ||
|
|
||
| results, err := utils.SafeCallMethod(c.OutboxMessageClient, CreateBulk, creators) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| return results[0].Interface().(MessageBulkCreator) | ||
| } | ||
|
|
||
| func WrapOutboxMessageClient(client interface{}) (MessageClient, error) { | ||
| if err := validateOutboxMessageClient(client); err != nil { | ||
| return nil, roxy.Wrap(err, "validating client") | ||
| } | ||
|
|
||
| return &outboxMessageClientWrapper{OutboxMessageClient: client}, nil | ||
| } | ||
|
|
||
| func validateOutboxMessageClient(client interface{}) error { | ||
| if client == nil { | ||
| return ErrNilClient | ||
| } | ||
|
|
||
| handleError := func(methodName string) error { | ||
| return roxy.Wrap(utils.ErrMethodNotFound, methodName) | ||
| } | ||
|
|
||
| methods := []string{Create, CreateBulk} | ||
| for _, method := range methods { | ||
| if !utils.HasMethod(client, method) { | ||
| return handleError(method) | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| type OutboxMessageClientInterface[T MessageBuilderInterface[T], U MessageBulkCreator] interface { | ||
| Create() T | ||
| CreateBulk(messageCreators ...T) U | ||
| } |
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,33 @@ | ||
| package outbox | ||
|
|
||
| import ( | ||
| "errors" | ||
| ) | ||
|
|
||
| var ( | ||
| ErrEmptyTopic = errors.New("empty topic") | ||
| ErrEmptyPayload = errors.New("empty payload") | ||
| ) | ||
|
|
||
| type Message struct { | ||
| Topic string | ||
| Payload []byte | ||
| Headers map[string]string | ||
| } | ||
|
|
||
| func (m Message) BuildEntMessage(creator MessageCreator) MessageCreator { | ||
| return creator. | ||
| SetTopic(m.Topic). | ||
| SetPayload(m.Payload). | ||
| SetHeaders(m.Headers) | ||
| } | ||
|
|
||
| func (m Message) Validate() error { | ||
| if m.Topic == "" { | ||
| return ErrEmptyTopic | ||
| } | ||
| if m.Payload == nil || len(m.Payload) == 0 { | ||
| return ErrEmptyPayload | ||
| } | ||
| return nil | ||
| } |
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,5 @@ | ||
| package outbox | ||
|
|
||
| type Relayer interface { | ||
| Start() | ||
| } |
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,89 @@ | ||
| package outbox | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
|
|
||
| "github.com/TheRafaBonin/roxy" | ||
| ) | ||
|
|
||
| var ( | ||
| ErrNoMessages = errors.New("no messages") | ||
| ) | ||
|
|
||
| type Storer interface { | ||
| Store(ctx context.Context, txOutboxMessageClient interface{}, messages []Message) error | ||
| WithTxClient(txOutboxMessageClient interface{}) (TransactionalStorer, error) | ||
| } | ||
|
|
||
| type TransactionalStorer interface { | ||
| Store(ctx context.Context, messages []Message) error | ||
| } | ||
|
|
||
| type storer struct{} | ||
|
|
||
| // Store implements Storer. | ||
| func (s storer) Store(ctx context.Context, txOutboxMessageClient interface{}, messages []Message) error { | ||
| if err := validateMessages(messages); err != nil { | ||
| return roxy.Wrap(err, "validating messages") | ||
| } | ||
|
|
||
| txClient, err := WrapOutboxMessageClient(txOutboxMessageClient) | ||
| if err != nil { | ||
| return roxy.Wrap(err, "wrapping tx client") | ||
| } | ||
|
|
||
| entMessages := make([]MessageCreator, len(messages)) | ||
| for i, msg := range messages { | ||
| entMessages[i] = msg.BuildEntMessage(txClient.Create()) | ||
| } | ||
|
|
||
| err = txClient.CreateBulk(entMessages...).Exec(ctx) | ||
| if err != nil { | ||
| return roxy.Wrap(err, "creating messages") | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // WithTxClient implements Storer. | ||
| func (s *storer) WithTxClient(txOutboxMessageClient interface{}) (TransactionalStorer, error) { | ||
| return newTransactionalStorer(s, txOutboxMessageClient) | ||
| } | ||
|
|
||
| func NewStorer() Storer { | ||
| return &storer{} | ||
| } | ||
|
|
||
| type transactionalStorer struct { | ||
| storer Storer | ||
| txClient interface{} | ||
| } | ||
|
|
||
| // Store implements TransactionalStorer. | ||
| func (t transactionalStorer) Store(ctx context.Context, messages []Message) error { | ||
| return t.storer.Store(ctx, t.txClient, messages) | ||
| } | ||
|
|
||
| func newTransactionalStorer(storer Storer, txOutboxMessageClient interface{}) (TransactionalStorer, error) { | ||
| if err := validateOutboxMessageClient(txOutboxMessageClient); err != nil { | ||
| return nil, roxy.Wrap(err, "validating tx client") | ||
| } | ||
|
|
||
| return &transactionalStorer{ | ||
| storer: storer, | ||
| txClient: txOutboxMessageClient, | ||
| }, nil | ||
| } | ||
|
|
||
| func validateMessages(messages []Message) error { | ||
| if len(messages) == 0 { | ||
| return ErrNoMessages | ||
| } | ||
|
|
||
| errs := make([]error, len(messages)) | ||
| for i, msg := range messages { | ||
| errs[i] = msg.Validate() | ||
| } | ||
| return errors.Join(errs...) | ||
| } |
Oops, something went wrong.