-
Notifications
You must be signed in to change notification settings - Fork 244
/
mailrequests.go
130 lines (113 loc) · 3.42 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
package shhext
import (
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
whisper "github.com/status-im/whisper/whisperv6"
)
// EnvelopeState in local tracker
type EnvelopeState int
const (
// NotRegistered returned if asked hash wasn't registered in the tracker.
NotRegistered EnvelopeState = -1
// EnvelopePosted is set when envelope was added to a local whisper queue.
EnvelopePosted EnvelopeState = iota
// EnvelopeSent is set when envelope is sent to atleast one peer.
EnvelopeSent
// MailServerRequestSent is set when p2p request is sent to the mailserver
MailServerRequestSent
)
// MailRequestMonitor is responsible for monitoring history request to mailservers.
type MailRequestMonitor struct {
w *whisper.Whisper
handler EnvelopeEventsHandler
mu sync.Mutex
cache map[common.Hash]EnvelopeState
requestsRegistry *RequestsRegistry
wg sync.WaitGroup
quit chan struct{}
}
// 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 common.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 whisper.EnvelopeEvent, 100) // must be buffered to prevent blocking whisper
sub := m.w.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 whisper.EnvelopeEvent) {
handlers := map[whisper.EventType]func(whisper.EnvelopeEvent){
whisper.EventMailServerRequestSent: m.handleRequestSent,
whisper.EventMailServerRequestCompleted: m.handleEventMailServerRequestCompleted,
whisper.EventMailServerRequestExpired: m.handleEventMailServerRequestExpired,
}
if handler, ok := handlers[event.Event]; ok {
handler(event)
}
}
func (m *MailRequestMonitor) handleRequestSent(event whisper.EnvelopeEvent) {
m.mu.Lock()
defer m.mu.Unlock()
m.cache[event.Hash] = MailServerRequestSent
}
func (m *MailRequestMonitor) handleEventMailServerRequestCompleted(event whisper.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.(*whisper.MailServerResponse); ok {
m.handler.MailServerRequestCompleted(event.Hash, resp.LastEnvelopeHash, resp.Cursor, resp.Error)
}
}
}
func (m *MailRequestMonitor) handleEventMailServerRequestExpired(event whisper.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)
}
}