diff --git a/api/gql_ev/msgbus.graphqls b/api/gql_ev/msgbus.graphqls index 8c1c59c..53c0f3d 100644 --- a/api/gql_ev/msgbus.graphqls +++ b/api/gql_ev/msgbus.graphqls @@ -2,7 +2,8 @@ extend type Query { posts(streamID: String! paging: PageInput): Connection! } extend type Subscription { - postAdded(streamID: String!): PostEvent + """after == 0 start from begining, after == -1 start from end""" + postAdded(streamID: String! after: Int! = -1): PostEvent } type PostEvent implements Edge { id: ID! diff --git a/api/gql_ev/resolver.go b/api/gql_ev/resolver.go index a75aa66..cb23e8d 100644 --- a/api/gql_ev/resolver.go +++ b/api/gql_ev/resolver.go @@ -3,7 +3,6 @@ package gql_ev import ( "context" "fmt" - "log" "time" "github.com/sour-is/ev/pkg/es" @@ -66,13 +65,13 @@ func (r *Resolver) Posts(ctx context.Context, streamID string, paging *PageInput }, nil } -func (r *Resolver) PostAdded(ctx context.Context, streamID string) (<-chan *PostEvent, error) { +func (r *Resolver) PostAdded(ctx context.Context, streamID string, after int64) (<-chan *PostEvent, error) { es := r.es.EventStream() if es == nil { return nil, fmt.Errorf("EventStore does not implement streaming") } - sub, err := es.Subscribe(ctx, streamID) + sub, err := es.Subscribe(ctx, streamID, after) if err != nil { return nil, err } @@ -83,7 +82,7 @@ func (r *Resolver) PostAdded(ctx context.Context, streamID string) (<-chan *Post defer func() { ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) defer cancel() - log.Print(sub.Close(ctx)) + sub.Close(ctx) }() for sub.Recv(ctx) { diff --git a/gqlgen.yml b/gqlgen.yml index e7907ac..f0329d0 100644 --- a/gqlgen.yml +++ b/gqlgen.yml @@ -54,12 +54,12 @@ models: - github.com/99designs/gqlgen/graphql.Uint32 Int: model: - - github.com/99designs/gqlgen/graphql.Int - github.com/99designs/gqlgen/graphql.Int64 - github.com/99designs/gqlgen/graphql.Int32 - - github.com/99designs/gqlgen/graphql.Uint + - github.com/99designs/gqlgen/graphql.Int - github.com/99designs/gqlgen/graphql.Uint64 - github.com/99designs/gqlgen/graphql.Uint32 + - github.com/99designs/gqlgen/graphql.Uint Time: model: - github.com/99designs/gqlgen/graphql.Time diff --git a/internal/graph/generated/generated.go b/internal/graph/generated/generated.go index 693a1a1..b7419b8 100644 --- a/internal/graph/generated/generated.go +++ b/internal/graph/generated/generated.go @@ -81,7 +81,7 @@ type ComplexityRoot struct { } Subscription struct { - PostAdded func(childComplexity int, streamID string) int + PostAdded func(childComplexity int, streamID string, after int64) int } _Service struct { @@ -93,7 +93,7 @@ type QueryResolver interface { Posts(ctx context.Context, streamID string, paging *gql_ev.PageInput) (*gql_ev.Connection, error) } type SubscriptionResolver interface { - PostAdded(ctx context.Context, streamID string) (<-chan *gql_ev.PostEvent, error) + PostAdded(ctx context.Context, streamID string, after int64) (<-chan *gql_ev.PostEvent, error) } type executableSchema struct { @@ -245,7 +245,7 @@ func (e *executableSchema) Complexity(typeName, field string, childComplexity in return 0, false } - return e.complexity.Subscription.PostAdded(childComplexity, args["streamID"].(string)), true + return e.complexity.Subscription.PostAdded(childComplexity, args["streamID"].(string), args["after"].(int64)), true case "_Service.sdl": if e.complexity._Service.SDL == nil { @@ -372,7 +372,8 @@ directive @goTag( posts(streamID: String! paging: PageInput): Connection! } extend type Subscription { - postAdded(streamID: String!): PostEvent + """after == 0 start from begining, after == -1 start from end""" + postAdded(streamID: String! after: Int! = -1): PostEvent } type PostEvent implements Edge { id: ID! @@ -461,6 +462,15 @@ func (ec *executionContext) field_Subscription_postAdded_args(ctx context.Contex } } args["streamID"] = arg0 + var arg1 int64 + if tmp, ok := rawArgs["after"]; ok { + ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("after")) + arg1, err = ec.unmarshalNInt2int64(ctx, tmp) + if err != nil { + return nil, err + } + } + args["after"] = arg1 return args, nil } @@ -1434,7 +1444,7 @@ func (ec *executionContext) _Subscription_postAdded(ctx context.Context, field g }() resTmp, err := ec.ResolverMiddleware(ctx, func(rctx context.Context) (interface{}, error) { ctx = rctx // use context from middleware stack in children - return ec.resolvers.Subscription().PostAdded(rctx, fc.Args["streamID"].(string)) + return ec.resolvers.Subscription().PostAdded(rctx, fc.Args["streamID"].(string), fc.Args["after"].(int64)) }) if err != nil { ec.Error(ctx, err) @@ -4130,6 +4140,21 @@ func (ec *executionContext) marshalNID2string(ctx context.Context, sel ast.Selec return res } +func (ec *executionContext) unmarshalNInt2int64(ctx context.Context, v interface{}) (int64, error) { + res, err := graphql.UnmarshalInt64(v) + return res, graphql.ErrorOnPath(ctx, err) +} + +func (ec *executionContext) marshalNInt2int64(ctx context.Context, sel ast.SelectionSet, v int64) graphql.Marshaler { + res := graphql.MarshalInt64(v) + if res == graphql.Null { + if !graphql.HasFieldError(ctx, graphql.GetFieldContext(ctx)) { + ec.Errorf(ctx, "the requested element is null which the schema does not allow") + } + } + return res +} + func (ec *executionContext) unmarshalNInt2uint64(ctx context.Context, v interface{}) (uint64, error) { res, err := graphql.UnmarshalUint64(v) return res, graphql.ErrorOnPath(ctx, err) diff --git a/pkg/es/driver/disk-store/disk-store.go b/pkg/es/driver/disk-store/disk-store.go index cb6743a..7f151d7 100644 --- a/pkg/es/driver/disk-store/disk-store.go +++ b/pkg/es/driver/disk-store/disk-store.go @@ -114,31 +114,9 @@ func (es *eventLog) Read(ctx context.Context, pos, count int64) (event.Events, e return nil } - if count == AllEvents { - count = int64(first - last) - } - - var start uint64 - - switch { - case pos >= 0 && count > 0: - start = first + uint64(pos) - case pos < 0 && count > 0: - start = uint64(int64(last) + pos + 1) - - case pos >= 0 && count < 0: - start = first + uint64(pos) - if pos > 1 { - start -= 2 // if pos is positive and count negative start before - } - if pos <= 1 { - return nil // if pos is one or zero and negative count nothing to return - } - case pos < 0 && count < 0: - start = uint64(int64(last) + pos) - } - if start >= last { - return nil // if start is after last and positive count nothing to return + start, count := math.PagerBox(first, last, pos, count) + if count == 0 { + return nil } events = make([]event.Event, math.Abs(count)) diff --git a/pkg/es/driver/driver.go b/pkg/es/driver/driver.go index ddaf696..625baf6 100644 --- a/pkg/es/driver/driver.go +++ b/pkg/es/driver/driver.go @@ -25,6 +25,6 @@ type Subscription interface { } type EventStream interface { - Subscribe(ctx context.Context, streamID string) (Subscription, error) + Subscribe(ctx context.Context, streamID string, start int64) (Subscription, error) Send(ctx context.Context, streamID string, events event.Events) error } diff --git a/pkg/es/driver/mem-store/mem-store.go b/pkg/es/driver/mem-store/mem-store.go index 13d0869..9192b7e 100644 --- a/pkg/es/driver/mem-store/mem-store.go +++ b/pkg/es/driver/mem-store/mem-store.go @@ -87,31 +87,9 @@ func (es *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Eve return nil } - if count == AllEvents { - count = int64(first - last) - } - - var start uint64 - - switch { - case pos >= 0 && count > 0: - start = first + uint64(pos) - case pos < 0 && count > 0: - start = uint64(int64(last) + pos + 1) - - case pos >= 0 && count < 0: - start = first + uint64(pos) - if pos > 1 { - start -= 2 // if pos is positive and count negative start before - } - if pos <= 1 { - return nil // if pos is one or zero and negative count nothing to return - } - case pos < 0 && count < 0: - start = uint64(int64(last) + pos) - } - if start >= last { - return nil // if start is after last and positive count nothing to return + start, count := math.PagerBox(first, last, pos, count) + if count == 0 { + return nil } events = make([]event.Event, math.Abs(count)) diff --git a/pkg/es/driver/streamer/streamer.go b/pkg/es/driver/streamer/streamer.go index 3342c7a..26e362e 100644 --- a/pkg/es/driver/streamer/streamer.go +++ b/pkg/es/driver/streamer/streamer.go @@ -2,7 +2,6 @@ package streamer import ( "context" - "log" "github.com/sour-is/ev/pkg/es" "github.com/sour-is/ev/pkg/es/driver" @@ -45,26 +44,24 @@ func (s *streamer) EventLog(ctx context.Context, streamID string) (driver.EventL var _ driver.EventStream = (*streamer)(nil) -func (s *streamer) Subscribe(ctx context.Context, streamID string) (driver.Subscription, error) { - log.Println("subscribe", streamID) +func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64) (driver.Subscription, error) { events, err := s.up.EventLog(ctx, streamID) if err != nil { return nil, err } sub := &subscription{topic: streamID, events: events} sub.position = locker.New(&position{ + idx: start, size: es.AllEvents, }) sub.unsub = s.delete(streamID, sub) return sub, s.state.Modify(ctx, func(state *state) error { state.subscribers[streamID] = append(state.subscribers[streamID], sub) - log.Println("subs=", len(state.subscribers[streamID])) return nil }) } func (s *streamer) Send(ctx context.Context, streamID string, events event.Events) error { - log.Println("send", streamID, len(events)) return s.state.Modify(ctx, func(state *state) error { for _, sub := range state.subscribers[streamID] { return sub.position.Modify(ctx, func(position *position) error { @@ -83,7 +80,6 @@ func (s *streamer) Send(ctx context.Context, streamID string, events event.Event func (s *streamer) delete(streamID string, sub *subscription) func(context.Context) error { return func(ctx context.Context) error { - log.Println("unsub", streamID) if err := ctx.Err(); err != nil { return err } @@ -93,7 +89,6 @@ func (s *streamer) delete(streamID string, sub *subscription) func(context.Conte if lis[i] == sub { lis[i] = lis[len(lis)-1] state.subscribers[streamID] = lis[:len(lis)-1] - log.Println("subs=", len(state.subscribers[streamID])) return nil } @@ -148,7 +143,6 @@ type subscription struct { func (s *subscription) Recv(ctx context.Context) bool { var wait func(context.Context) bool - log.Println("recv more") err := s.position.Modify(ctx, func(position *position) error { if position.size == es.AllEvents { return nil @@ -156,14 +150,11 @@ func (s *subscription) Recv(ctx context.Context) bool { if position.size == 0 { position.wait = make(chan struct{}) wait = func(ctx context.Context) bool { - log.Println("waiting", s.topic) select { case <-position.wait: - log.Println("got some") return true case <-ctx.Done(): - log.Println("got cancel") return false } @@ -185,15 +176,12 @@ func (s *subscription) Recv(ctx context.Context) bool { } func (s *subscription) Events(ctx context.Context) (event.Events, error) { var events event.Events - log.Println("get events") return events, s.position.Modify(ctx, func(position *position) error { var err error events, err = s.events.Read(ctx, int64(position.idx), position.size) - log.Printf("got events=%d %#v", len(events), position) position.size = int64(len(events)) if len(events) > 0 { position.idx = int64(events.First().EventMeta().Position - 1) - log.Println(position, events.First()) } return err }) diff --git a/pkg/locker/locker.go b/pkg/locker/locker.go index 6495e44..b7bf629 100644 --- a/pkg/locker/locker.go +++ b/pkg/locker/locker.go @@ -2,7 +2,6 @@ package locker import ( "context" - "log" ) type Locked[T any] struct { @@ -26,9 +25,6 @@ func (s *Locked[T]) Modify(ctx context.Context, fn func(*T) error) error { select { case state := <-s.state: defer func() { s.state <- state }() - log.Printf("locker %T to %p", state, fn) - defer log.Printf("locker %T from %p", state, fn) - return fn(state) case <-ctx.Done(): return ctx.Err() diff --git a/pkg/math/math.go b/pkg/math/math.go index 7e20efd..56ff8d0 100644 --- a/pkg/math/math.go +++ b/pkg/math/math.go @@ -38,3 +38,30 @@ func Min[T ordered](i T, candidates ...T) T { } return i } + +func PagerBox(first, last uint64, pos, count int64) (uint64, int64) { + var start uint64 + + if pos >= 0 { + start = first + uint64(pos) + } else { + start = uint64(int64(last) + pos + 1) + } + + switch { + case count > 0: + count = Min(count, int64(last-start)+1) + + case pos >= 0 && count < 0: + count = Max(count, int64(first-start)) + + case pos < 0 && count < 0: + count = Max(count, int64(first-start)-1) + } + + if count == 0 || (start < first && count <= 0) || (start > last && count >= 0) { + return 0, 0 + } + + return start, count +} diff --git a/pkg/math/math_test.go b/pkg/math/math_test.go index 3c183a6..83cd4fc 100644 --- a/pkg/math/math_test.go +++ b/pkg/math/math_test.go @@ -1,9 +1,11 @@ package math_test import ( + "log" "testing" "github.com/matryer/is" + "github.com/sour-is/ev/pkg/es" "github.com/sour-is/ev/pkg/math" ) @@ -46,3 +48,46 @@ func TestMath(t *testing.T) { )) } + +func TestPagerBox(t *testing.T) { + is := is.New(t) + + tests := []struct { + first uint64 + last uint64 + pos int64 + n int64 + + start uint64 + count int64 + }{ + {1, 10, 0, 10, 1, 10}, + {1, 10, 0, 11, 1, 10}, + {1, 5, 0, 10, 1, 5}, + {1, 10, 4, 10, 5, 6}, + {1, 10, 5, 10, 6, 5}, + {1, 10, 0, -10, 0, 0}, + {1, 10, 1, -1, 2, -1}, + {1, 10, 1, -10, 2, -1}, + {1, 10, -1, 1, 10, 1}, + {1, 10, -2, 10, 9, 2}, + {1, 10, -1, -1, 10, -1}, + {1, 10, -2, -10, 9, -9}, + {1, 10, 0, -10, 0, 0}, + {1, 10, 10, 10, 0, 0}, + {1, 10, 0, es.AllEvents, 1, 10}, + {1, 10, -1, -es.AllEvents, 10, -10}, + } + + for _, tt := range tests { + start, count := math.PagerBox(tt.first, tt.last, tt.pos, tt.n) + if count > 0 { + log.Print(tt, "|", start, count, int64(start)+count-1) + } else { + log.Print(tt, "|", start, count, int64(start)+count+1) + } + + is.Equal(start, tt.start) + is.Equal(count, tt.count) + } +}