/
msghandler.go
245 lines (230 loc) · 6.56 KB
/
msghandler.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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
package uspv
import (
"fmt"
"log"
"github.com/roasbeef/btcd/wire"
"github.com/roasbeef/btcutil"
)
func (s *SPVCon) incomingMessageHandler() {
for {
n, xm, _, err := wire.ReadMessageN(s.con, s.localVersion, s.TS.Param.Net)
if err != nil {
log.Printf("ReadMessageN error. Disconnecting: %s\n", err.Error())
return
}
s.RBytes += uint64(n)
// log.Printf("Got %d byte %s message\n", n, xm.Command())
switch m := xm.(type) {
case *wire.MsgVersion:
log.Printf("Got version message. Agent %s, version %d, at height %d\n",
m.UserAgent, m.ProtocolVersion, m.LastBlock)
s.remoteVersion = uint32(m.ProtocolVersion) // weird cast! bug?
case *wire.MsgVerAck:
log.Printf("Got verack. Whatever.\n")
case *wire.MsgAddr:
log.Printf("got %d addresses.\n", len(m.AddrList))
case *wire.MsgPing:
// log.Printf("Got a ping message. We should pong back or they will kick us off.")
go s.PongBack(m.Nonce)
case *wire.MsgPong:
log.Printf("Got a pong response. OK.\n")
case *wire.MsgBlock:
s.IngestBlock(m)
case *wire.MsgMerkleBlock:
s.IngestMerkleBlock(m)
case *wire.MsgHeaders: // concurrent because we keep asking for blocks
go s.HeaderHandler(m)
case *wire.MsgTx: // not concurrent! txs must be in order
s.TxHandler(m)
case *wire.MsgReject:
log.Printf("Rejected! cmd: %s code: %s tx: %s reason: %s",
m.Cmd, m.Code.String(), m.Hash.String(), m.Reason)
case *wire.MsgInv:
s.InvHandler(m)
case *wire.MsgNotFound:
log.Printf("Got not found response from remote:")
for i, thing := range m.InvList {
log.Printf("\t$d) %s: %s", i, thing.Type, thing.Hash)
}
case *wire.MsgGetData:
s.GetDataHandler(m)
default:
log.Printf("Got unknown message type %s\n", m.Command())
}
}
return
}
// this one seems kindof pointless? could get ridf of it and let
// functions call WriteMessageN themselves...
func (s *SPVCon) outgoingMessageHandler() {
for {
msg := <-s.outMsgQueue
n, err := wire.WriteMessageN(s.con, msg, s.localVersion, s.TS.Param.Net)
if err != nil {
log.Printf("Write message error: %s", err.Error())
}
s.WBytes += uint64(n)
}
return
}
// fPositiveHandler monitors false positives and when it gets enough of them,
//
func (s *SPVCon) fPositiveHandler() {
var fpAccumulator int32
for {
fpAccumulator += <-s.fPositives // blocks here
if fpAccumulator > 7 {
filt, err := s.TS.GimmeFilter()
if err != nil {
log.Printf("Filter creation error: %s\n", err.Error())
log.Printf("uhoh, crashing filter handler")
return
}
// send filter
s.SendFilter(filt)
fmt.Printf("sent filter %x\n", filt.MsgFilterLoad().Filter)
// clear the channel
finClear:
for {
select {
case x := <-s.fPositives:
fpAccumulator += x
default:
break finClear
}
}
fmt.Printf("reset %d false positives\n", fpAccumulator)
// reset accumulator
fpAccumulator = 0
}
}
}
func (s *SPVCon) HeaderHandler(m *wire.MsgHeaders) {
moar, err := s.IngestHeaders(m)
if err != nil {
log.Printf("Header error: %s\n", err.Error())
return
}
// more to get? if so, ask for them and return
if moar {
err = s.AskForHeaders()
if err != nil {
log.Printf("AskForHeaders error: %s", err.Error())
}
return
}
// no moar, done w/ headers, get blocks
err = s.AskForBlocks()
if err != nil {
log.Printf("AskForBlocks error: %s", err.Error())
return
}
}
// TxHandler takes in transaction messages that come in from either a request
// after an inv message or after a merkle block message.
func (s *SPVCon) TxHandler(m *wire.MsgTx) {
s.TS.OKMutex.Lock()
height, ok := s.TS.OKTxids[m.TxSha()]
s.TS.OKMutex.Unlock()
if !ok {
log.Printf("Tx %s unknown, will not ingest\n", m.TxSha().String())
return
}
// check for double spends
// allTxs, err := s.TS.GetAllTxs()
// if err != nil {
// log.Printf("Can't get txs from db: %s", err.Error())
// return
// }
// dubs, err := CheckDoubleSpends(m, allTxs)
// if err != nil {
// log.Printf("CheckDoubleSpends error: %s", err.Error())
// return
// }
// if len(dubs) > 0 {
// for i, dub := range dubs {
// fmt.Printf("dub %d known tx %s and new tx %s are exclusive!!!\n",
// i, dub.String(), m.TxSha().String())
// }
// }
utilTx := btcutil.NewTx(m)
if !s.HardMode || s.TS.localFilter.MatchTxAndUpdate(utilTx) {
hits, err := s.TS.Ingest(m, height)
if err != nil {
log.Printf("Incoming Tx error: %s\n", err.Error())
return
}
if hits == 0 && !s.HardMode {
log.Printf("tx %s had no hits, filter false positive.",
m.TxSha().String())
s.fPositives <- 1 // add one false positive to chan
return
}
log.Printf("tx %s ingested and matches %d utxo/adrs.",
m.TxSha().String(), hits)
}
}
// GetDataHandler responds to requests for tx data, which happen after
// advertising our txs via an inv message
func (s *SPVCon) GetDataHandler(m *wire.MsgGetData) {
log.Printf("got GetData. Contains:\n")
var sent int32
for i, thing := range m.InvList {
log.Printf("\t%d)%s : %s",
i, thing.Type.String(), thing.Hash.String())
// separate wittx and tx. needed / combine?
// does the same thing right now
if thing.Type == wire.InvTypeWitnessTx {
tx, err := s.TS.GetTx(&thing.Hash)
if err != nil {
log.Printf("error getting tx %s: %s",
thing.Hash.String(), err.Error())
}
s.outMsgQueue <- tx
sent++
continue
}
if thing.Type == wire.InvTypeTx {
tx, err := s.TS.GetTx(&thing.Hash)
if err != nil {
log.Printf("error getting tx %s: %s",
thing.Hash.String(), err.Error())
}
//tx.Flags = 0x00 // dewitnessify
s.outMsgQueue <- tx
sent++
continue
}
// didn't match, so it's not something we're responding to
log.Printf("We only respond to tx requests, ignoring")
}
log.Printf("sent %d of %d requested items", sent, len(m.InvList))
}
func (s *SPVCon) InvHandler(m *wire.MsgInv) {
log.Printf("got inv. Contains:\n")
for i, thing := range m.InvList {
log.Printf("\t%d)%s : %s",
i, thing.Type.String(), thing.Hash.String())
if thing.Type == wire.InvTypeTx {
if !s.Ironman { // ignore tx invs in ironman mode
// new tx, OK it at 0 and request
s.TS.AddTxid(&thing.Hash, 0) // unconfirmed
s.AskForTx(thing.Hash)
}
}
if thing.Type == wire.InvTypeBlock { // new block what to do?
select {
case <-s.inWaitState:
// start getting headers
fmt.Printf("asking for headers due to inv block\n")
err := s.AskForHeaders()
if err != nil {
log.Printf("AskForHeaders error: %s", err.Error())
}
default:
// drop it as if its component particles had high thermal energies
fmt.Printf("inv block but ignoring; not synched\n")
}
}
}
}