1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
package pds
import (
"context"
"errors"
"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/atprotorepo"
"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/events"
"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise"
)
const SourcePrefix = "pds/"
var ErrEnded = errors.New("pds: the repo subscription ended")
type Open func(ctx context.Context, collections []string) (atprotorepo.Subscriber, error)
type Config struct {
Repo string
}
type Deps struct {
Open Open
}
type Source struct {
repo string
open Open
}
var (
_ events.Source = (*Source)(nil)
_ events.Stream = (*stream)(nil)
)
func New(cfg Config, deps Deps) (*Source, error) {
if cfg.Repo == "" {
return nil, errors.New("pds: a repo is required")
}
if deps.Open == nil {
return nil, errors.New("pds: an Open function is required")
}
return &Source{repo: cfg.Repo, open: deps.Open}, nil
}
func (s *Source) Subscribe(ctx context.Context, filter events.Filter) (events.Stream, error) {
sub, err := s.open(ctx, filter.Collections)
if err != nil {
return nil, err
}
out := &stream{commits: make(chan events.Event), done: make(chan struct{}), filter: filter, source: SourcePrefix + s.repo}
out.group = supervise.New(ctx)
out.group.Go("pds subscription "+s.repo, func(ctx context.Context) error {
defer close(out.done)
out.err = sub.Subscribe(ctx, atprotorepo.SubscribeOptions{Collections: filter.Collections, Cursor: filter.Cursor}, func(c atprotorepo.Commit) error {
select {
case out.commits <- FromCommit(c):
return nil
case <-ctx.Done():
return ctx.Err()
}
})
return nil
})
return out, nil
}
type stream struct {
group *supervise.Group
commits chan events.Event
done chan struct{}
err error
filter events.Filter
source string
}
func (s *stream) Next(ctx context.Context) (events.Event, error) {
select {
case <-ctx.Done():
return events.Event{}, ctx.Err()
case e := <-s.commits:
e.Source = s.source
return s.filter.Narrow(e), nil
case <-s.done:
if s.err != nil {
return events.Event{}, s.err
}
return events.Event{}, ErrEnded
}
}
func (s *stream) Close() error {
return s.group.Stop()
}
func FromCommit(c atprotorepo.Commit) events.Event {
out := events.Event{Seq: c.Seq, Repo: c.Repo, Commit: c.CID, Rev: c.Rev, Time: c.Time}
for _, op := range c.Ops {
out.Ops = append(out.Ops, events.Op{Action: op.Action, Path: op.Collection + "/" + op.RKey, CID: op.CID})
}
return out
}
