73 lines
1.7 KiB
Go
73 lines
1.7 KiB
Go
package application
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
)
|
|
|
|
var (
|
|
// ErrInvalidEvent indicates that an event is not part of the event contract.
|
|
ErrInvalidEvent = errors.New("invalid application event")
|
|
// ErrRuntimeClosed indicates that the runtime no longer accepts events.
|
|
ErrRuntimeClosed = errors.New("application runtime closed")
|
|
)
|
|
|
|
// Runtime is the minimal event bus shared by background use cases and adapters.
|
|
type Runtime struct {
|
|
events chan Event
|
|
done chan struct{}
|
|
closeOnce sync.Once
|
|
}
|
|
|
|
// NewRuntime creates an event runtime with the requested queue capacity.
|
|
func NewRuntime(buffer int) *Runtime {
|
|
if buffer < 0 {
|
|
panic("application runtime buffer must not be negative")
|
|
}
|
|
return &Runtime{
|
|
events: make(chan Event, buffer),
|
|
done: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// Publish queues an event or returns when the context/runtime is closed.
|
|
func (runtime *Runtime) Publish(ctx context.Context, event Event) error {
|
|
if !event.Type.Valid() {
|
|
return fmt.Errorf("%w: unknown type %q", ErrInvalidEvent, event.Type)
|
|
}
|
|
|
|
select {
|
|
case <-runtime.done:
|
|
return ErrRuntimeClosed
|
|
default:
|
|
}
|
|
|
|
select {
|
|
case <-runtime.done:
|
|
return ErrRuntimeClosed
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case runtime.events <- event:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// Events exposes the read-only event stream to adapters.
|
|
func (runtime *Runtime) Events() <-chan Event {
|
|
return runtime.events
|
|
}
|
|
|
|
// Done is closed when the runtime stops accepting events.
|
|
func (runtime *Runtime) Done() <-chan struct{} {
|
|
return runtime.done
|
|
}
|
|
|
|
// Close stops future publishes. It is safe to call more than once.
|
|
func (runtime *Runtime) Close() {
|
|
runtime.closeOnce.Do(func() {
|
|
close(runtime.done)
|
|
})
|
|
}
|