forked from nakagami/firebirdsql
/
event.go
170 lines (151 loc) · 3.8 KB
/
event.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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
package firebirdsql
import (
"database/sql"
"errors"
"fmt"
"sync"
"sync/atomic"
)
// Errors
var (
ErrAlreadySubscribe = errors.New("already subscribe")
ErrFbEventClosed = errors.New("fbevent already closed")
)
//SQLs
const (
sqlPostEvent = `execute block as begin post_event '%s'; end`
)
// FbEvent allows you to subscribe to events, also stores subscribers.
// It is possible to send events to the database.
type FbEvent struct {
mu sync.RWMutex
dsn string
conn *sql.DB
done chan struct{}
closed int32
closer sync.Once
chDoneSubscriber chan *Subscription
subscribers []*Subscription
}
// Event stores event data: the amount since the last time the event was received and id
type Event struct {
Name string
Count int
ID int32
RemoteID int32
}
// EventHandler callback function type
type EventHandler func(e Event)
// NewFBEvent returns FbEvent for event subscription
func NewFBEvent(dsn string) (*FbEvent, error) {
conn, err := sql.Open("firebirdsql", dsn)
if err != nil {
return nil, err
}
fbEvent := &FbEvent{
dsn: dsn,
conn: conn,
done: make(chan struct{}),
chDoneSubscriber: make(chan *Subscription),
}
go fbEvent.run()
return fbEvent, nil
}
// PostEvent posts an event to the database
func (e *FbEvent) PostEvent(name string) error {
_, err := e.conn.Exec(fmt.Sprintf(sqlPostEvent, name))
if err != nil {
return err
}
return nil
}
func (e *FbEvent) newSubscriber(events []string, cb EventHandler, chEvent chan Event) (*Subscription, error) {
subscriber, err := newSubscription(e.dsn, events, cb, chEvent, e.chDoneSubscriber)
if err != nil {
return nil, err
}
e.mu.Lock()
defer e.mu.Unlock()
e.subscribers = append(e.subscribers, subscriber)
return subscriber, nil
}
// Subscribers returns slice of all subscribers
func (e *FbEvent) Subscribers() []*Subscription {
e.mu.RLock()
defer e.mu.RUnlock()
return e.subscribers[:]
}
// Count returns the number of subscribers
func (e *FbEvent) Count() int {
e.mu.RLock()
defer e.mu.RUnlock()
return len(e.subscribers)
}
// Subscribe subscribe to events using the callback function
func (e *FbEvent) Subscribe(events []string, cb EventHandler) (*Subscription, error) {
return e.newSubscriber(events, cb, nil)
}
// SubscribeChan subscribe to events using the channel
func (e *FbEvent) SubscribeChan(events []string, chEvent chan Event) (*Subscription, error) {
return e.newSubscriber(events, nil, chEvent)
}
func (e *FbEvent) run() {
for {
select {
case <-e.done:
return
case subscriber := <-e.chDoneSubscriber:
e.shutdownSubscriber(subscriber)
}
}
}
func (e *FbEvent) shutdownSubscriber(subscriber *Subscription) {
e.mu.Lock()
defer e.mu.Unlock()
for i := range e.subscribers {
if e.subscribers[i] == subscriber {
last := len(e.subscribers) - 1
e.subscribers[i] = e.subscribers[last]
e.subscribers[last] = nil
e.subscribers = e.subscribers[:last]
return
}
}
}
// IsClosed returns a close flag
func (e *FbEvent) IsClosed() bool {
return atomic.LoadInt32(&e.closed) == 1
}
// Close closes FbEvent and all subscribers
func (e *FbEvent) Close() error {
if e.IsClosed() {
return ErrFbEventClosed
}
return e.doClose(nil)
}
func (e *FbEvent) closeWithError(err error) error {
if e.IsClosed() {
return ErrFbEventClosed
}
return e.doClose(err)
}
func (e *FbEvent) doClose(err error) (errResult error) {
atomic.StoreInt32(&e.closed, 1)
e.closer.Do(func() {
e.conn.Close()
e.mu.Lock()
wg := &sync.WaitGroup{}
wg.Add(len(e.subscribers))
for i := range e.subscribers {
go func(subscriber *Subscription) {
defer wg.Done()
subscriber.unsubscribeNoNotify()
}(e.subscribers[i])
}
e.subscribers = make([]*Subscription, 0)
e.mu.Unlock()
wg.Wait()
close(e.done)
})
return
}