-
Notifications
You must be signed in to change notification settings - Fork 245
/
mailrequests.go
138 lines (118 loc) · 3.57 KB
/
mailrequests.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
// +build !nimbus
package ext
import (
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/services/ext/mailservers"
)
// EnvelopeState in local tracker
type EnvelopeState int
const (
// NotRegistered returned if asked hash wasn't registered in the tracker.
NotRegistered EnvelopeState = -1
// MailServerRequestSent is set when p2p request is sent to the mailserver
MailServerRequestSent
)
// MailRequestMonitor is responsible for monitoring history request to mailservers.
type MailRequestMonitor struct {
eventSub mailservers.EnvelopeEventSubscriber
handler EnvelopeEventsHandler
mu sync.Mutex
cache map[types.Hash]EnvelopeState
requestsRegistry *RequestsRegistry
wg sync.WaitGroup
quit chan struct{}
}
func NewMailRequestMonitor(eventSub mailservers.EnvelopeEventSubscriber, h EnvelopeEventsHandler, reg *RequestsRegistry) *MailRequestMonitor {
return &MailRequestMonitor{
eventSub: eventSub,
handler: h,
cache: make(map[types.Hash]EnvelopeState),
requestsRegistry: reg,
}
}
// Start processing events.
func (m *MailRequestMonitor) Start() {
m.quit = make(chan struct{})
m.wg.Add(1)
go func() {
m.handleEnvelopeEvents()
m.wg.Done()
}()
}
// Stop process events.
func (m *MailRequestMonitor) Stop() {
close(m.quit)
m.wg.Wait()
}
func (m *MailRequestMonitor) GetState(hash types.Hash) EnvelopeState {
m.mu.Lock()
defer m.mu.Unlock()
state, exist := m.cache[hash]
if !exist {
return NotRegistered
}
return state
}
// handleEnvelopeEvents processes whisper envelope events
func (m *MailRequestMonitor) handleEnvelopeEvents() {
events := make(chan types.EnvelopeEvent, 100) // must be buffered to prevent blocking whisper
sub := m.eventSub.SubscribeEnvelopeEvents(events)
defer sub.Unsubscribe()
for {
select {
case <-m.quit:
return
case event := <-events:
m.handleEvent(event)
}
}
}
// handleEvent based on type of the event either triggers
// confirmation handler or removes hash from MailRequestMonitor
func (m *MailRequestMonitor) handleEvent(event types.EnvelopeEvent) {
handlers := map[types.EventType]func(types.EnvelopeEvent){
types.EventMailServerRequestSent: m.handleRequestSent,
types.EventMailServerRequestCompleted: m.handleEventMailServerRequestCompleted,
types.EventMailServerRequestExpired: m.handleEventMailServerRequestExpired,
}
if handler, ok := handlers[event.Event]; ok {
handler(event)
}
}
func (m *MailRequestMonitor) handleRequestSent(event types.EnvelopeEvent) {
m.mu.Lock()
defer m.mu.Unlock()
m.cache[event.Hash] = MailServerRequestSent
}
func (m *MailRequestMonitor) handleEventMailServerRequestCompleted(event types.EnvelopeEvent) {
m.mu.Lock()
defer m.mu.Unlock()
m.requestsRegistry.Unregister(event.Hash)
state, ok := m.cache[event.Hash]
if !ok || state != MailServerRequestSent {
return
}
log.Debug("mailserver response received", "hash", event.Hash)
delete(m.cache, event.Hash)
if m.handler != nil {
if resp, ok := event.Data.(*types.MailServerResponse); ok {
m.handler.MailServerRequestCompleted(event.Hash, resp.LastEnvelopeHash, resp.Cursor, resp.Error)
}
}
}
func (m *MailRequestMonitor) handleEventMailServerRequestExpired(event types.EnvelopeEvent) {
m.mu.Lock()
defer m.mu.Unlock()
m.requestsRegistry.Unregister(event.Hash)
state, ok := m.cache[event.Hash]
if !ok || state != MailServerRequestSent {
return
}
log.Debug("mailserver response expired", "hash", event.Hash)
delete(m.cache, event.Hash)
if m.handler != nil {
m.handler.MailServerRequestExpired(event.Hash)
}
}