-
Notifications
You must be signed in to change notification settings - Fork 3
/
watcher.go
96 lines (76 loc) · 2.06 KB
/
watcher.go
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
package spark
import (
"fmt"
"time"
"github.com/256dpi/fire"
"github.com/256dpi/fire/coal"
"github.com/globalsign/mgo/bson"
)
// TODO: How to close a watcher?
// Watcher will watch multiple collections and serve watch requests by clients.
type Watcher struct {
manager *manager
streams map[string]*Stream
// The function gets invoked by the watcher with critical errors.
Reporter func(error)
}
// NewWatcher creates and returns a new watcher.
func NewWatcher() *Watcher {
// prepare watcher
w := &Watcher{
streams: make(map[string]*Stream),
}
// create and add manager
w.manager = newManager(w)
return w
}
// Add will add a stream to the watcher.
func (w *Watcher) Add(stream *Stream) {
// initialize model
coal.Init(stream.Model)
// check existence
if w.streams[stream.Name()] != nil {
panic(fmt.Sprintf(`spark: stream with name "%s" already exists`, stream.Name()))
}
// save stream
w.streams[stream.Name()] = stream
// open stream
coal.OpenStream(stream.Store, stream.Model, nil, func(e coal.Event, id bson.ObjectId, m coal.Model, token []byte) {
// ignore real deleted events when soft delete has been enabled
if stream.SoftDelete && e == coal.Deleted {
return
}
// handle soft deleted documents
if stream.SoftDelete && e == coal.Updated {
// get soft delete field
softDeleteField := coal.L(stream.Model, "fire-soft-delete", true)
// get deleted time
t := m.MustGet(softDeleteField).(*time.Time)
// change type if document has been soft deleted
if t != nil && !t.IsZero() {
e = coal.Deleted
}
}
// create event
evt := &Event{
Type: e,
ID: id,
Model: m,
Stream: stream,
}
// broadcast event
w.manager.broadcast(evt)
}, nil, w.Reporter)
}
// Action returns an action that should be registered in the group under
// the "watch" name.
func (w *Watcher) Action() *fire.Action {
return &fire.Action{
Methods: []string{"GET"},
Callback: fire.C("spark/Watcher.Action", fire.All(), func(ctx *fire.Context) error {
// handle connection
w.manager.handle(ctx)
return nil
}),
}
}