2022-10-13 15:32:25 -06:00
|
|
|
// package streamer provides a driver to allow awaiting for new events to be added to a stream.
|
2022-08-09 16:23:33 -06:00
|
|
|
package streamer
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
2022-09-06 10:35:14 -06:00
|
|
|
"fmt"
|
2022-11-23 13:51:55 -07:00
|
|
|
"sync"
|
2022-08-09 16:23:33 -06:00
|
|
|
|
2022-11-20 10:15:51 -07:00
|
|
|
"go.opentelemetry.io/otel/attribute"
|
|
|
|
"go.opentelemetry.io/otel/trace"
|
|
|
|
|
2023-02-26 22:33:01 -07:00
|
|
|
"go.sour.is/ev"
|
|
|
|
"go.sour.is/ev/internal/lg"
|
|
|
|
"go.sour.is/ev/pkg/es/driver"
|
|
|
|
"go.sour.is/ev/pkg/es/event"
|
|
|
|
"go.sour.is/ev/pkg/locker"
|
2022-08-09 16:23:33 -06:00
|
|
|
)
|
|
|
|
|
|
|
|
type state struct {
|
|
|
|
subscribers map[string][]*subscription
|
|
|
|
}
|
|
|
|
|
|
|
|
type streamer struct {
|
|
|
|
state *locker.Locked[state]
|
|
|
|
up driver.Driver
|
|
|
|
}
|
|
|
|
|
|
|
|
func New(ctx context.Context) *streamer {
|
2022-09-06 10:35:14 -06:00
|
|
|
_, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
return &streamer{state: locker.New(&state{subscribers: map[string][]*subscription{}})}
|
|
|
|
}
|
|
|
|
|
2023-01-09 11:30:02 -07:00
|
|
|
var _ ev.Option = (*streamer)(nil)
|
2022-08-09 16:23:33 -06:00
|
|
|
|
2023-01-09 11:30:02 -07:00
|
|
|
func (s *streamer) Apply(e *ev.EventStore) {
|
2022-08-09 16:23:33 -06:00
|
|
|
s.up = e.Driver
|
|
|
|
e.Driver = s
|
|
|
|
}
|
|
|
|
func (s *streamer) Unwrap() driver.Driver {
|
|
|
|
return s.up
|
|
|
|
}
|
|
|
|
|
|
|
|
var _ driver.Driver = (*streamer)(nil)
|
|
|
|
|
|
|
|
func (s *streamer) Open(ctx context.Context, dsn string) (driver.Driver, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
return s.up.Open(ctx, dsn)
|
|
|
|
}
|
|
|
|
func (s *streamer) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
l, err := s.up.EventLog(ctx, streamID)
|
|
|
|
return &wrapper{streamID, l, s}, err
|
|
|
|
}
|
|
|
|
|
|
|
|
var _ driver.EventStream = (*streamer)(nil)
|
|
|
|
|
2022-08-10 10:09:58 -06:00
|
|
|
func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64) (driver.Subscription, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
events, err := s.up.EventLog(ctx, streamID)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2022-11-23 13:51:55 -07:00
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
sub := &subscription{topic: streamID, events: events}
|
|
|
|
sub.position = locker.New(&position{
|
2022-08-10 10:09:58 -06:00
|
|
|
idx: start,
|
2023-01-09 11:30:02 -07:00
|
|
|
size: ev.AllEvents,
|
2022-08-09 16:23:33 -06:00
|
|
|
})
|
|
|
|
sub.unsub = s.delete(streamID, sub)
|
|
|
|
|
2022-10-30 09:18:08 -06:00
|
|
|
return sub, s.state.Modify(ctx, func(ctx context.Context, state *state) error {
|
2022-08-09 16:23:33 -06:00
|
|
|
state.subscribers[streamID] = append(state.subscribers[streamID], sub)
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
}
|
|
|
|
func (s *streamer) Send(ctx context.Context, streamID string, events event.Events) error {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-10-30 09:18:08 -06:00
|
|
|
return s.state.Modify(ctx, func(ctx context.Context, state *state) error {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
2022-11-23 13:51:55 -07:00
|
|
|
|
2022-09-06 10:35:14 -06:00
|
|
|
span.AddEvent(fmt.Sprint("subscribers=", len(state.subscribers[streamID])))
|
2022-08-13 18:59:15 -06:00
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
for _, sub := range state.subscribers[streamID] {
|
2022-10-30 09:18:08 -06:00
|
|
|
err := sub.position.Modify(ctx, func(ctx context.Context, position *position) error {
|
2022-11-20 10:15:51 -07:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
2022-11-23 13:51:55 -07:00
|
|
|
|
2022-11-20 10:15:51 -07:00
|
|
|
span.SetAttributes(
|
|
|
|
attribute.String("streamID", streamID),
|
|
|
|
attribute.Int64("actualPosition", int64(events.Last().EventMeta().ActualPosition)),
|
|
|
|
attribute.String("actualStreamID", events.Last().EventMeta().ActualStreamID),
|
|
|
|
attribute.Int64("position", int64(events.Last().EventMeta().Position)),
|
|
|
|
)
|
2022-08-13 18:59:15 -06:00
|
|
|
|
2022-11-20 10:15:51 -07:00
|
|
|
position.size = int64(events.Last().EventMeta().ActualPosition - uint64(position.idx))
|
2022-08-09 16:23:33 -06:00
|
|
|
|
|
|
|
if position.wait != nil {
|
|
|
|
close(position.wait)
|
2022-09-06 10:35:14 -06:00
|
|
|
position.link = trace.LinkFromContext(ctx, attribute.String("src", "event"))
|
2022-08-09 16:23:33 -06:00
|
|
|
position.wait = nil
|
|
|
|
}
|
2022-11-23 13:51:55 -07:00
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
return nil
|
|
|
|
})
|
2022-08-12 15:53:16 -06:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
}
|
|
|
|
func (s *streamer) delete(streamID string, sub *subscription) func(context.Context) error {
|
|
|
|
return func(ctx context.Context) error {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
if err := ctx.Err(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2022-10-30 09:18:08 -06:00
|
|
|
return s.state.Modify(ctx, func(ctx context.Context, state *state) error {
|
2022-09-06 10:35:14 -06:00
|
|
|
_, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
lis := state.subscribers[streamID]
|
|
|
|
for i := range lis {
|
|
|
|
if lis[i] == sub {
|
|
|
|
lis[i] = lis[len(lis)-1]
|
|
|
|
state.subscribers[streamID] = lis[:len(lis)-1]
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
type wrapper struct {
|
|
|
|
topic string
|
|
|
|
up driver.EventLog
|
|
|
|
streamer *streamer
|
|
|
|
}
|
|
|
|
|
|
|
|
var _ driver.EventLog = (*wrapper)(nil)
|
|
|
|
|
2022-11-20 10:15:51 -07:00
|
|
|
func (r *wrapper) Unwrap() driver.EventLog {
|
|
|
|
return r.up
|
|
|
|
}
|
2022-10-25 20:15:57 -06:00
|
|
|
func (w *wrapper) Read(ctx context.Context, after int64, count int64) (event.Events, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-10-25 20:15:57 -06:00
|
|
|
return w.up.Read(ctx, after, count)
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|
2022-10-30 09:18:08 -06:00
|
|
|
func (w *wrapper) ReadN(ctx context.Context, index ...uint64) (event.Events, error) {
|
|
|
|
ctx, span := lg.Span(ctx)
|
|
|
|
defer span.End()
|
|
|
|
|
|
|
|
return w.up.ReadN(ctx, index...)
|
|
|
|
}
|
2022-08-09 16:23:33 -06:00
|
|
|
func (w *wrapper) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
i, err := w.up.Append(ctx, events, version)
|
|
|
|
if err != nil {
|
|
|
|
return i, err
|
|
|
|
}
|
|
|
|
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span = lg.Fork(ctx)
|
|
|
|
go func() {
|
|
|
|
defer span.End()
|
|
|
|
|
|
|
|
err := w.streamer.Send(ctx, w.topic, events)
|
|
|
|
span.RecordError(err)
|
|
|
|
}()
|
|
|
|
|
|
|
|
return i, nil
|
|
|
|
}
|
2022-08-09 16:23:33 -06:00
|
|
|
func (w *wrapper) FirstIndex(ctx context.Context) (uint64, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
return w.up.FirstIndex(ctx)
|
|
|
|
}
|
|
|
|
func (w *wrapper) LastIndex(ctx context.Context) (uint64, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
return w.up.LastIndex(ctx)
|
|
|
|
}
|
2022-08-14 10:04:15 -06:00
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
type position struct {
|
|
|
|
size int64
|
|
|
|
idx int64
|
2022-09-06 10:35:14 -06:00
|
|
|
link trace.Link
|
2022-08-09 16:23:33 -06:00
|
|
|
wait chan struct{}
|
|
|
|
}
|
|
|
|
|
|
|
|
type subscription struct {
|
|
|
|
topic string
|
|
|
|
|
|
|
|
position *locker.Locked[position]
|
|
|
|
|
|
|
|
events driver.EventLog
|
|
|
|
unsub func(context.Context) error
|
2022-11-23 13:51:55 -07:00
|
|
|
once sync.Once
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|
|
|
|
|
2022-11-23 13:51:55 -07:00
|
|
|
func (s *subscription) Recv(ctx context.Context) <-chan bool {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-11-23 13:51:55 -07:00
|
|
|
done := make(chan bool)
|
2022-08-13 18:59:15 -06:00
|
|
|
|
2022-11-23 13:51:55 -07:00
|
|
|
go func() {
|
|
|
|
var wait func(context.Context) bool
|
|
|
|
defer close(done)
|
2022-08-13 18:59:15 -06:00
|
|
|
|
2022-11-23 13:51:55 -07:00
|
|
|
err := s.position.Modify(ctx, func(ctx context.Context, position *position) error {
|
|
|
|
_, span := lg.Span(ctx)
|
|
|
|
defer span.End()
|
2022-08-13 18:59:15 -06:00
|
|
|
|
2023-01-09 11:30:02 -07:00
|
|
|
if position.size == ev.AllEvents {
|
2022-11-23 13:51:55 -07:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
if position.size == 0 {
|
|
|
|
position.wait = make(chan struct{})
|
|
|
|
wait = func(ctx context.Context) bool {
|
|
|
|
ctx, span := lg.Span(ctx)
|
|
|
|
defer span.End()
|
|
|
|
|
|
|
|
select {
|
|
|
|
case <-position.wait:
|
|
|
|
if position.link.SpanContext.IsValid() {
|
|
|
|
_, span := lg.Span(ctx, trace.WithLinks(position.link))
|
|
|
|
span.AddEvent("recv event")
|
|
|
|
span.End()
|
|
|
|
position.link = trace.Link{}
|
|
|
|
}
|
|
|
|
return true
|
|
|
|
|
|
|
|
case <-ctx.Done():
|
|
|
|
return false
|
2022-09-06 10:35:14 -06:00
|
|
|
}
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|
|
|
|
}
|
2022-11-23 13:51:55 -07:00
|
|
|
position.idx += position.size
|
|
|
|
position.size = 0
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
done <- false
|
|
|
|
return
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|
|
|
|
|
2022-11-23 13:51:55 -07:00
|
|
|
if wait != nil {
|
|
|
|
done <- wait(ctx)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
done <- true
|
|
|
|
}()
|
2022-08-09 16:23:33 -06:00
|
|
|
|
2022-11-23 13:51:55 -07:00
|
|
|
return done
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|
|
|
|
func (s *subscription) Events(ctx context.Context) (event.Events, error) {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
var events event.Events
|
2022-10-30 09:18:08 -06:00
|
|
|
return events, s.position.Modify(ctx, func(ctx context.Context, position *position) error {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
var err error
|
2022-08-10 21:18:04 -06:00
|
|
|
events, err = s.events.Read(ctx, position.idx, position.size)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2022-08-09 16:23:33 -06:00
|
|
|
position.size = int64(len(events))
|
|
|
|
if len(events) > 0 {
|
2022-11-20 10:15:51 -07:00
|
|
|
position.idx = int64(events.First().EventMeta().ActualPosition - 1)
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|
2022-11-20 10:15:51 -07:00
|
|
|
span.SetAttributes(
|
|
|
|
attribute.Int64("position.idx", position.idx),
|
|
|
|
attribute.Int64("position.size", position.size),
|
|
|
|
attribute.Int64("meta.ActualPosition", int64(events.First().EventMeta().ActualPosition)),
|
|
|
|
attribute.Int64("meta.Position", int64(events.First().EventMeta().Position)),
|
|
|
|
)
|
|
|
|
|
2022-08-09 16:23:33 -06:00
|
|
|
return err
|
|
|
|
})
|
|
|
|
}
|
|
|
|
func (s *subscription) Close(ctx context.Context) error {
|
2022-09-06 10:35:14 -06:00
|
|
|
ctx, span := lg.Span(ctx)
|
2022-08-13 18:59:15 -06:00
|
|
|
defer span.End()
|
|
|
|
|
2022-11-23 13:51:55 -07:00
|
|
|
if s == nil || s.unsub == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
var err error
|
|
|
|
s.once.Do(func() { err = s.unsub(ctx) })
|
|
|
|
|
|
|
|
return err
|
2022-08-09 16:23:33 -06:00
|
|
|
}
|