-
Notifications
You must be signed in to change notification settings - Fork 244
/
publisher.go
127 lines (103 loc) · 2.48 KB
/
publisher.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
package publisher
import (
"crypto/ecdsa"
"errors"
"time"
"go.uber.org/zap"
"github.com/status-im/status-go/eth-node/crypto"
)
const (
// How often a ticker fires in seconds.
tickerInterval = 120
// How often we should publish a contact code in seconds.
publishInterval = 21600
// Cooldown period on acking messages when not targeting our device.
deviceNotFoundAckInterval = 7200
)
var (
errNotEnoughTimePassed = errors.New("not enough time passed")
)
type Publisher struct {
persistence *persistence
logger *zap.Logger
notifyCh chan struct{}
quit chan struct{}
}
func New(logger *zap.Logger) *Publisher {
if logger == nil {
logger = zap.NewNop()
}
return &Publisher{
persistence: newPersistence(),
logger: logger.With(zap.Namespace("Publisher")),
}
}
func (p *Publisher) Start() <-chan struct{} {
logger := p.logger.With(zap.String("site", "Start"))
logger.Info("starting publisher")
p.notifyCh = make(chan struct{}, 100)
p.quit = make(chan struct{})
go p.tickerLoop()
return p.notifyCh
}
func (p *Publisher) Stop() {
// If hasn't started, ignore
if p.quit == nil {
return
}
select {
case _, ok := <-p.quit:
if !ok {
// channel already closed
return
}
default:
close(p.quit)
}
}
func (p *Publisher) tickerLoop() {
ticker := time.NewTicker(tickerInterval * time.Second)
go func() {
logger := p.logger.With(zap.String("site", "tickerLoop"))
for {
select {
case <-ticker.C:
err := p.notify()
switch err {
case errNotEnoughTimePassed:
logger.Debug("not enough time passed")
case nil:
// skip
default:
logger.Error("error while sending a contact code", zap.Error(err))
}
case <-p.quit:
ticker.Stop()
return
}
}
}()
}
func (p *Publisher) notify() error {
lastPublished := p.persistence.getLastPublished()
now := time.Now().Unix()
if now-lastPublished < publishInterval {
return errNotEnoughTimePassed
}
select {
case p.notifyCh <- struct{}{}:
default:
p.logger.Warn("publisher channel full, dropping message")
}
p.persistence.setLastPublished(now)
return nil
}
func (p *Publisher) ShouldAdvertiseBundle(publicKey *ecdsa.PublicKey, now int64) (bool, error) {
identity := crypto.CompressPubkey(publicKey)
lastAcked := p.persistence.lastAck(identity)
return now-lastAcked < deviceNotFoundAckInterval, nil
}
func (p *Publisher) SetLastAck(publicKey *ecdsa.PublicKey, now int64) {
identity := crypto.CompressPubkey(publicKey)
p.persistence.setLastAck(identity, now)
}