forked from btcsuite/btcd
-
Notifications
You must be signed in to change notification settings - Fork 29
/
rpcwebsocket.go
2643 lines (2344 loc) · 81.4 KB
/
rpcwebsocket.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
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Copyright (c) 2013-2017 The btcsuite developers
// Copyright (c) 2015-2017 The Decred developers
// Use of this source code is governed by an ISC
// license that can be found in the LICENSE file.
package main
import (
"bytes"
"container/list"
"crypto/sha256"
"crypto/subtle"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"math"
"sync"
"time"
"golang.org/x/crypto/ripemd160"
"github.com/btcsuite/websocket"
"github.com/roasbeef/btcd/blockchain"
"github.com/roasbeef/btcd/btcjson"
"github.com/roasbeef/btcd/chaincfg"
"github.com/roasbeef/btcd/chaincfg/chainhash"
"github.com/roasbeef/btcd/database"
"github.com/roasbeef/btcd/txscript"
"github.com/roasbeef/btcd/wire"
"github.com/roasbeef/btcutil"
)
const (
// websocketSendBufferSize is the number of elements the send channel
// can queue before blocking. Note that this only applies to requests
// handled directly in the websocket client input handler or the async
// handler since notifications have their own queuing mechanism
// independent of the send channel buffer.
websocketSendBufferSize = 50
)
type semaphore chan struct{}
func makeSemaphore(n int) semaphore {
return make(chan struct{}, n)
}
func (s semaphore) acquire() { s <- struct{}{} }
func (s semaphore) release() { <-s }
// timeZeroVal is simply the zero value for a time.Time and is used to avoid
// creating multiple instances.
var timeZeroVal time.Time
// wsCommandHandler describes a callback function used to handle a specific
// command.
type wsCommandHandler func(*wsClient, interface{}) (interface{}, error)
// wsHandlers maps RPC command strings to appropriate websocket handler
// functions. This is set by init because help references wsHandlers and thus
// causes a dependency loop.
var wsHandlers map[string]wsCommandHandler
var wsHandlersBeforeInit = map[string]wsCommandHandler{
"loadtxfilter": handleLoadTxFilter,
"help": handleWebsocketHelp,
"notifyblocks": handleNotifyBlocks,
"notifynewtransactions": handleNotifyNewTransactions,
"notifyreceived": handleNotifyReceived,
"notifyspent": handleNotifySpent,
"session": handleSession,
"stopnotifyblocks": handleStopNotifyBlocks,
"stopnotifynewtransactions": handleStopNotifyNewTransactions,
"stopnotifyspent": handleStopNotifySpent,
"stopnotifyreceived": handleStopNotifyReceived,
"rescan": handleRescan,
"rescanblocks": handleRescanBlocks,
}
// WebsocketHandler handles a new websocket client by creating a new wsClient,
// starting it, and blocking until the connection closes. Since it blocks, it
// must be run in a separate goroutine. It should be invoked from the websocket
// server handler which runs each new connection in a new goroutine thereby
// satisfying the requirement.
func (s *rpcServer) WebsocketHandler(conn *websocket.Conn, remoteAddr string,
authenticated bool, isAdmin bool) {
// Clear the read deadline that was set before the websocket hijacked
// the connection.
conn.SetReadDeadline(timeZeroVal)
// Limit max number of websocket clients.
rpcsLog.Infof("New websocket client %s", remoteAddr)
if s.ntfnMgr.NumClients()+1 > cfg.RPCMaxWebsockets {
rpcsLog.Infof("Max websocket clients exceeded [%d] - "+
"disconnecting client %s", cfg.RPCMaxWebsockets,
remoteAddr)
conn.Close()
return
}
// Create a new websocket client to handle the new websocket connection
// and wait for it to shutdown. Once it has shutdown (and hence
// disconnected), remove it and any notifications it registered for.
client, err := newWebsocketClient(s, conn, remoteAddr, authenticated, isAdmin)
if err != nil {
rpcsLog.Errorf("Failed to serve client %s: %v", remoteAddr, err)
conn.Close()
return
}
s.ntfnMgr.AddClient(client)
client.Start()
client.WaitForShutdown()
s.ntfnMgr.RemoveClient(client)
rpcsLog.Infof("Disconnected websocket client %s", remoteAddr)
}
// wsNotificationManager is a connection and notification manager used for
// websockets. It allows websocket clients to register for notifications they
// are interested in. When an event happens elsewhere in the code such as
// transactions being added to the memory pool or block connects/disconnects,
// the notification manager is provided with the relevant details needed to
// figure out which websocket clients need to be notified based on what they
// have registered for and notifies them accordingly. It is also used to keep
// track of all connected websocket clients.
type wsNotificationManager struct {
// server is the RPC server the notification manager is associated with.
server *rpcServer
// queueNotification queues a notification for handling.
queueNotification chan interface{}
// notificationMsgs feeds notificationHandler with notifications
// and client (un)registeration requests from a queue as well as
// registeration and unregisteration requests from clients.
notificationMsgs chan interface{}
// Access channel for current number of connected clients.
numClients chan int
// Shutdown handling
wg sync.WaitGroup
quit chan struct{}
}
// queueHandler manages a queue of empty interfaces, reading from in and
// sending the oldest unsent to out. This handler stops when either of the
// in or quit channels are closed, and closes out before returning, without
// waiting to send any variables still remaining in the queue.
func queueHandler(in <-chan interface{}, out chan<- interface{}, quit <-chan struct{}) {
var q []interface{}
var dequeue chan<- interface{}
skipQueue := out
var next interface{}
out:
for {
select {
case n, ok := <-in:
if !ok {
// Sender closed input channel.
break out
}
// Either send to out immediately if skipQueue is
// non-nil (queue is empty) and reader is ready,
// or append to the queue and send later.
select {
case skipQueue <- n:
default:
q = append(q, n)
dequeue = out
skipQueue = nil
next = q[0]
}
case dequeue <- next:
copy(q, q[1:])
q[len(q)-1] = nil // avoid leak
q = q[:len(q)-1]
if len(q) == 0 {
dequeue = nil
skipQueue = out
} else {
next = q[0]
}
case <-quit:
break out
}
}
close(out)
}
// queueHandler maintains a queue of notifications and notification handler
// control messages.
func (m *wsNotificationManager) queueHandler() {
queueHandler(m.queueNotification, m.notificationMsgs, m.quit)
m.wg.Done()
}
// NotifyBlockConnected passes a block newly-connected to the best chain
// to the notification manager for block and transaction notification
// processing.
func (m *wsNotificationManager) NotifyBlockConnected(block *btcutil.Block) {
// As NotifyBlockConnected will be called by the block manager
// and the RPC server may no longer be running, use a select
// statement to unblock enqueuing the notification once the RPC
// server has begun shutting down.
select {
case m.queueNotification <- (*notificationBlockConnected)(block):
case <-m.quit:
}
}
// NotifyBlockDisconnected passes a block disconnected from the best chain
// to the notification manager for block notification processing.
func (m *wsNotificationManager) NotifyBlockDisconnected(block *btcutil.Block) {
// As NotifyBlockDisconnected will be called by the block manager
// and the RPC server may no longer be running, use a select
// statement to unblock enqueuing the notification once the RPC
// server has begun shutting down.
select {
case m.queueNotification <- (*notificationBlockDisconnected)(block):
case <-m.quit:
}
}
// NotifyMempoolTx passes a transaction accepted by mempool to the
// notification manager for transaction notification processing. If
// isNew is true, the tx is is a new transaction, rather than one
// added to the mempool during a reorg.
func (m *wsNotificationManager) NotifyMempoolTx(tx *btcutil.Tx, isNew bool) {
n := ¬ificationTxAcceptedByMempool{
isNew: isNew,
tx: tx,
}
// As NotifyMempoolTx will be called by mempool and the RPC server
// may no longer be running, use a select statement to unblock
// enqueuing the notification once the RPC server has begun
// shutting down.
select {
case m.queueNotification <- n:
case <-m.quit:
}
}
// wsClientFilter tracks relevant addresses for each websocket client for
// the `rescanblocks` extension. It is modified by the `loadtxfilter` command.
//
// NOTE: This extension was ported from github.com/decred/dcrd
type wsClientFilter struct {
mu sync.Mutex
// Implemented fast paths for address lookup.
pubKeyHashes map[[ripemd160.Size]byte]struct{}
scriptHashes map[[ripemd160.Size]byte]struct{}
compressedPubKeys map[[33]byte]struct{}
uncompressedPubKeys map[[65]byte]struct{}
// A fallback address lookup map in case a fast path doesn't exist.
// Only exists for completeness. If using this shows up in a profile,
// there's a good chance a fast path should be added.
otherAddresses map[string]struct{}
// Outpoints of unspent outputs.
unspent map[wire.OutPoint]struct{}
}
// newWSClientFilter creates a new, empty wsClientFilter struct to be used
// for a websocket client.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func newWSClientFilter(addresses []string, unspentOutPoints []wire.OutPoint, params *chaincfg.Params) *wsClientFilter {
filter := &wsClientFilter{
pubKeyHashes: map[[ripemd160.Size]byte]struct{}{},
scriptHashes: map[[ripemd160.Size]byte]struct{}{},
compressedPubKeys: map[[33]byte]struct{}{},
uncompressedPubKeys: map[[65]byte]struct{}{},
otherAddresses: map[string]struct{}{},
unspent: make(map[wire.OutPoint]struct{}, len(unspentOutPoints)),
}
for _, s := range addresses {
filter.addAddressStr(s, params)
}
for i := range unspentOutPoints {
filter.addUnspentOutPoint(&unspentOutPoints[i])
}
return filter
}
// addAddress adds an address to a wsClientFilter, treating it correctly based
// on the type of address passed as an argument.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) addAddress(a btcutil.Address) {
switch a := a.(type) {
case *btcutil.AddressPubKeyHash:
f.pubKeyHashes[*a.Hash160()] = struct{}{}
return
case *btcutil.AddressScriptHash:
f.scriptHashes[*a.Hash160()] = struct{}{}
return
case *btcutil.AddressPubKey:
serializedPubKey := a.ScriptAddress()
switch len(serializedPubKey) {
case 33: // compressed
var compressedPubKey [33]byte
copy(compressedPubKey[:], serializedPubKey)
f.compressedPubKeys[compressedPubKey] = struct{}{}
return
case 65: // uncompressed
var uncompressedPubKey [65]byte
copy(uncompressedPubKey[:], serializedPubKey)
f.uncompressedPubKeys[uncompressedPubKey] = struct{}{}
return
}
}
f.otherAddresses[a.EncodeAddress()] = struct{}{}
}
// addAddressStr parses an address from a string and then adds it to the
// wsClientFilter using addAddress.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) addAddressStr(s string, params *chaincfg.Params) {
// If address can't be decoded, no point in saving it since it should also
// impossible to create the address from an inspected transaction output
// script.
a, err := btcutil.DecodeAddress(s, params)
if err != nil {
return
}
f.addAddress(a)
}
// existsAddress returns true if the passed address has been added to the
// wsClientFilter.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) existsAddress(a btcutil.Address) bool {
switch a := a.(type) {
case *btcutil.AddressPubKeyHash:
_, ok := f.pubKeyHashes[*a.Hash160()]
return ok
case *btcutil.AddressScriptHash:
_, ok := f.scriptHashes[*a.Hash160()]
return ok
case *btcutil.AddressPubKey:
serializedPubKey := a.ScriptAddress()
switch len(serializedPubKey) {
case 33: // compressed
var compressedPubKey [33]byte
copy(compressedPubKey[:], serializedPubKey)
_, ok := f.compressedPubKeys[compressedPubKey]
if !ok {
_, ok = f.pubKeyHashes[*a.AddressPubKeyHash().Hash160()]
}
return ok
case 65: // uncompressed
var uncompressedPubKey [65]byte
copy(uncompressedPubKey[:], serializedPubKey)
_, ok := f.uncompressedPubKeys[uncompressedPubKey]
if !ok {
_, ok = f.pubKeyHashes[*a.AddressPubKeyHash().Hash160()]
}
return ok
}
}
_, ok := f.otherAddresses[a.EncodeAddress()]
return ok
}
// removeAddress removes the passed address, if it exists, from the
// wsClientFilter.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) removeAddress(a btcutil.Address) {
switch a := a.(type) {
case *btcutil.AddressPubKeyHash:
delete(f.pubKeyHashes, *a.Hash160())
return
case *btcutil.AddressScriptHash:
delete(f.scriptHashes, *a.Hash160())
return
case *btcutil.AddressPubKey:
serializedPubKey := a.ScriptAddress()
switch len(serializedPubKey) {
case 33: // compressed
var compressedPubKey [33]byte
copy(compressedPubKey[:], serializedPubKey)
delete(f.compressedPubKeys, compressedPubKey)
return
case 65: // uncompressed
var uncompressedPubKey [65]byte
copy(uncompressedPubKey[:], serializedPubKey)
delete(f.uncompressedPubKeys, uncompressedPubKey)
return
}
}
delete(f.otherAddresses, a.EncodeAddress())
}
// removeAddressStr parses an address from a string and then removes it from the
// wsClientFilter using removeAddress.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) removeAddressStr(s string, params *chaincfg.Params) {
a, err := btcutil.DecodeAddress(s, params)
if err == nil {
f.removeAddress(a)
} else {
delete(f.otherAddresses, s)
}
}
// addUnspentOutPoint adds an outpoint to the wsClientFilter.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) addUnspentOutPoint(op *wire.OutPoint) {
f.unspent[*op] = struct{}{}
}
// existsUnspentOutPoint returns true if the passed outpoint has been added to
// the wsClientFilter.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) existsUnspentOutPoint(op *wire.OutPoint) bool {
_, ok := f.unspent[*op]
return ok
}
// removeUnspentOutPoint removes the passed outpoint, if it exists, from the
// wsClientFilter.
//
// NOTE: This extension was ported from github.com/decred/dcrd
func (f *wsClientFilter) removeUnspentOutPoint(op *wire.OutPoint) {
delete(f.unspent, *op)
}
// Notification types
type notificationBlockConnected btcutil.Block
type notificationBlockDisconnected btcutil.Block
type notificationTxAcceptedByMempool struct {
isNew bool
tx *btcutil.Tx
}
// Notification control requests
type notificationRegisterClient wsClient
type notificationUnregisterClient wsClient
type notificationRegisterBlocks wsClient
type notificationUnregisterBlocks wsClient
type notificationRegisterNewMempoolTxs wsClient
type notificationUnregisterNewMempoolTxs wsClient
type notificationRegisterSpent struct {
wsc *wsClient
ops []*wire.OutPoint
}
type notificationUnregisterSpent struct {
wsc *wsClient
op *wire.OutPoint
}
type notificationRegisterAddr struct {
wsc *wsClient
addrs []string
}
type notificationUnregisterAddr struct {
wsc *wsClient
addr string
}
// notificationHandler reads notifications and control messages from the queue
// handler and processes one at a time.
func (m *wsNotificationManager) notificationHandler() {
// clients is a map of all currently connected websocket clients.
clients := make(map[chan struct{}]*wsClient)
// Maps used to hold lists of websocket clients to be notified on
// certain events. Each websocket client also keeps maps for the events
// which have multiple triggers to make removal from these lists on
// connection close less horrendously expensive.
//
// Where possible, the quit channel is used as the unique id for a client
// since it is quite a bit more efficient than using the entire struct.
blockNotifications := make(map[chan struct{}]*wsClient)
txNotifications := make(map[chan struct{}]*wsClient)
watchedOutPoints := make(map[wire.OutPoint]map[chan struct{}]*wsClient)
watchedAddrs := make(map[string]map[chan struct{}]*wsClient)
out:
for {
select {
case n, ok := <-m.notificationMsgs:
if !ok {
// queueHandler quit.
break out
}
switch n := n.(type) {
case *notificationBlockConnected:
block := (*btcutil.Block)(n)
// Skip iterating through all txs if no
// tx notification requests exist.
if len(watchedOutPoints) != 0 || len(watchedAddrs) != 0 {
for _, tx := range block.Transactions() {
m.notifyForTx(watchedOutPoints,
watchedAddrs, tx, block)
}
}
if len(blockNotifications) != 0 {
m.notifyBlockConnected(blockNotifications,
block)
m.notifyFilteredBlockConnected(blockNotifications,
block)
}
case *notificationBlockDisconnected:
block := (*btcutil.Block)(n)
if len(blockNotifications) != 0 {
m.notifyBlockDisconnected(blockNotifications,
block)
m.notifyFilteredBlockDisconnected(blockNotifications,
block)
}
case *notificationTxAcceptedByMempool:
if n.isNew && len(txNotifications) != 0 {
m.notifyForNewTx(txNotifications, n.tx)
}
m.notifyForTx(watchedOutPoints, watchedAddrs, n.tx, nil)
m.notifyRelevantTxAccepted(n.tx, clients)
case *notificationRegisterBlocks:
wsc := (*wsClient)(n)
blockNotifications[wsc.quit] = wsc
case *notificationUnregisterBlocks:
wsc := (*wsClient)(n)
delete(blockNotifications, wsc.quit)
case *notificationRegisterClient:
wsc := (*wsClient)(n)
clients[wsc.quit] = wsc
case *notificationUnregisterClient:
wsc := (*wsClient)(n)
// Remove any requests made by the client as well as
// the client itself.
delete(blockNotifications, wsc.quit)
delete(txNotifications, wsc.quit)
for k := range wsc.spentRequests {
op := k
m.removeSpentRequest(watchedOutPoints, wsc, &op)
}
for addr := range wsc.addrRequests {
m.removeAddrRequest(watchedAddrs, wsc, addr)
}
delete(clients, wsc.quit)
case *notificationRegisterSpent:
m.addSpentRequests(watchedOutPoints, n.wsc, n.ops)
case *notificationUnregisterSpent:
m.removeSpentRequest(watchedOutPoints, n.wsc, n.op)
case *notificationRegisterAddr:
m.addAddrRequests(watchedAddrs, n.wsc, n.addrs)
case *notificationUnregisterAddr:
m.removeAddrRequest(watchedAddrs, n.wsc, n.addr)
case *notificationRegisterNewMempoolTxs:
wsc := (*wsClient)(n)
txNotifications[wsc.quit] = wsc
case *notificationUnregisterNewMempoolTxs:
wsc := (*wsClient)(n)
delete(txNotifications, wsc.quit)
default:
rpcsLog.Warn("Unhandled notification type")
}
case m.numClients <- len(clients):
case <-m.quit:
// RPC server shutting down.
break out
}
}
for _, c := range clients {
c.Disconnect()
}
m.wg.Done()
}
// NumClients returns the number of clients actively being served.
func (m *wsNotificationManager) NumClients() (n int) {
select {
case n = <-m.numClients:
case <-m.quit: // Use default n (0) if server has shut down.
}
return
}
// RegisterBlockUpdates requests block update notifications to the passed
// websocket client.
func (m *wsNotificationManager) RegisterBlockUpdates(wsc *wsClient) {
m.queueNotification <- (*notificationRegisterBlocks)(wsc)
}
// UnregisterBlockUpdates removes block update notifications for the passed
// websocket client.
func (m *wsNotificationManager) UnregisterBlockUpdates(wsc *wsClient) {
m.queueNotification <- (*notificationUnregisterBlocks)(wsc)
}
// subscribedClients returns the set of all websocket client quit channels that
// are registered to receive notifications regarding tx, either due to tx
// spending a watched output or outputting to a watched address. Matching
// client's filters are updated based on this transaction's outputs and output
// addresses that may be relevant for a client.
func (m *wsNotificationManager) subscribedClients(tx *btcutil.Tx,
clients map[chan struct{}]*wsClient) map[chan struct{}]struct{} {
// Use a map of client quit channels as keys to prevent duplicates when
// multiple inputs and/or outputs are relevant to the client.
subscribed := make(map[chan struct{}]struct{})
msgTx := tx.MsgTx()
for _, input := range msgTx.TxIn {
for quitChan, wsc := range clients {
wsc.Lock()
filter := wsc.filterData
wsc.Unlock()
if filter == nil {
continue
}
filter.mu.Lock()
if filter.existsUnspentOutPoint(&input.PreviousOutPoint) {
subscribed[quitChan] = struct{}{}
}
filter.mu.Unlock()
}
}
for i, output := range msgTx.TxOut {
_, addrs, _, err := txscript.ExtractPkScriptAddrs(
output.PkScript, m.server.cfg.ChainParams)
if err != nil {
// Clients are not able to subscribe to
// nonstandard or non-address outputs.
continue
}
for quitChan, wsc := range clients {
wsc.Lock()
filter := wsc.filterData
wsc.Unlock()
if filter == nil {
continue
}
filter.mu.Lock()
for _, a := range addrs {
if filter.existsAddress(a) {
subscribed[quitChan] = struct{}{}
op := wire.OutPoint{
Hash: *tx.Hash(),
Index: uint32(i),
}
filter.addUnspentOutPoint(&op)
}
}
filter.mu.Unlock()
}
}
return subscribed
}
// notifyBlockConnected notifies websocket clients that have registered for
// block updates when a block is connected to the main chain.
func (*wsNotificationManager) notifyBlockConnected(clients map[chan struct{}]*wsClient,
block *btcutil.Block) {
// Notify interested websocket clients about the connected block.
ntfn := btcjson.NewBlockConnectedNtfn(block.Hash().String(), block.Height(),
block.MsgBlock().Header.Timestamp.Unix())
marshalledJSON, err := btcjson.MarshalCmd(nil, ntfn)
if err != nil {
rpcsLog.Errorf("Failed to marshal block connected notification: "+
"%v", err)
return
}
for _, wsc := range clients {
wsc.QueueNotification(marshalledJSON)
}
}
// notifyBlockDisconnected notifies websocket clients that have registered for
// block updates when a block is disconnected from the main chain (due to a
// reorganize).
func (*wsNotificationManager) notifyBlockDisconnected(clients map[chan struct{}]*wsClient, block *btcutil.Block) {
// Skip notification creation if no clients have requested block
// connected/disconnected notifications.
if len(clients) == 0 {
return
}
// Notify interested websocket clients about the disconnected block.
ntfn := btcjson.NewBlockDisconnectedNtfn(block.Hash().String(),
block.Height(), block.MsgBlock().Header.Timestamp.Unix())
marshalledJSON, err := btcjson.MarshalCmd(nil, ntfn)
if err != nil {
rpcsLog.Errorf("Failed to marshal block disconnected "+
"notification: %v", err)
return
}
for _, wsc := range clients {
wsc.QueueNotification(marshalledJSON)
}
}
// notifyFilteredBlockConnected notifies websocket clients that have registered for
// block updates when a block is connected to the main chain.
func (m *wsNotificationManager) notifyFilteredBlockConnected(clients map[chan struct{}]*wsClient,
block *btcutil.Block) {
// Create the common portion of the notification that is the same for
// every client.
var w bytes.Buffer
err := block.MsgBlock().Header.Serialize(&w)
if err != nil {
rpcsLog.Errorf("Failed to serialize header for filtered block "+
"connected notification: %v", err)
return
}
ntfn := btcjson.NewFilteredBlockConnectedNtfn(block.Height(),
hex.EncodeToString(w.Bytes()), nil)
// Search for relevant transactions for each client and save them
// serialized in hex encoding for the notification.
subscribedTxs := make(map[chan struct{}][]string)
for _, tx := range block.Transactions() {
var txHex string
for quitChan := range m.subscribedClients(tx, clients) {
if txHex == "" {
txHex = txHexString(tx.MsgTx())
}
subscribedTxs[quitChan] = append(subscribedTxs[quitChan], txHex)
}
}
for quitChan, wsc := range clients {
// Add all discovered transactions for this client. For clients
// that have no new-style filter, add the empty string slice.
ntfn.SubscribedTxs = subscribedTxs[quitChan]
// Marshal and queue notification.
marshalledJSON, err := btcjson.MarshalCmd(nil, ntfn)
if err != nil {
rpcsLog.Errorf("Failed to marshal filtered block "+
"connected notification: %v", err)
return
}
wsc.QueueNotification(marshalledJSON)
}
}
// notifyFilteredBlockDisconnected notifies websocket clients that have registered for
// block updates when a block is disconnected from the main chain (due to a
// reorganize).
func (*wsNotificationManager) notifyFilteredBlockDisconnected(clients map[chan struct{}]*wsClient,
block *btcutil.Block) {
// Skip notification creation if no clients have requested block
// connected/disconnected notifications.
if len(clients) == 0 {
return
}
// Notify interested websocket clients about the disconnected block.
var w bytes.Buffer
err := block.MsgBlock().Header.Serialize(&w)
if err != nil {
rpcsLog.Errorf("Failed to serialize header for filtered block "+
"disconnected notification: %v", err)
return
}
ntfn := btcjson.NewFilteredBlockDisconnectedNtfn(block.Height(),
hex.EncodeToString(w.Bytes()))
marshalledJSON, err := btcjson.MarshalCmd(nil, ntfn)
if err != nil {
rpcsLog.Errorf("Failed to marshal filtered block disconnected "+
"notification: %v", err)
return
}
for _, wsc := range clients {
wsc.QueueNotification(marshalledJSON)
}
}
// RegisterNewMempoolTxsUpdates requests notifications to the passed websocket
// client when new transactions are added to the memory pool.
func (m *wsNotificationManager) RegisterNewMempoolTxsUpdates(wsc *wsClient) {
m.queueNotification <- (*notificationRegisterNewMempoolTxs)(wsc)
}
// UnregisterNewMempoolTxsUpdates removes notifications to the passed websocket
// client when new transaction are added to the memory pool.
func (m *wsNotificationManager) UnregisterNewMempoolTxsUpdates(wsc *wsClient) {
m.queueNotification <- (*notificationUnregisterNewMempoolTxs)(wsc)
}
// notifyForNewTx notifies websocket clients that have registered for updates
// when a new transaction is added to the memory pool.
func (m *wsNotificationManager) notifyForNewTx(clients map[chan struct{}]*wsClient, tx *btcutil.Tx) {
txHashStr := tx.Hash().String()
mtx := tx.MsgTx()
var amount int64
for _, txOut := range mtx.TxOut {
amount += txOut.Value
}
ntfn := btcjson.NewTxAcceptedNtfn(txHashStr, btcutil.Amount(amount).ToBTC())
marshalledJSON, err := btcjson.MarshalCmd(nil, ntfn)
if err != nil {
rpcsLog.Errorf("Failed to marshal tx notification: %s", err.Error())
return
}
var verboseNtfn *btcjson.TxAcceptedVerboseNtfn
var marshalledJSONVerbose []byte
for _, wsc := range clients {
if wsc.verboseTxUpdates {
if marshalledJSONVerbose != nil {
wsc.QueueNotification(marshalledJSONVerbose)
continue
}
net := m.server.cfg.ChainParams
rawTx, err := createTxRawResult(net, mtx, txHashStr, nil,
"", 0, 0)
if err != nil {
return
}
verboseNtfn = btcjson.NewTxAcceptedVerboseNtfn(*rawTx)
marshalledJSONVerbose, err = btcjson.MarshalCmd(nil,
verboseNtfn)
if err != nil {
rpcsLog.Errorf("Failed to marshal verbose tx "+
"notification: %s", err.Error())
return
}
wsc.QueueNotification(marshalledJSONVerbose)
} else {
wsc.QueueNotification(marshalledJSON)
}
}
}
// RegisterSpentRequests requests a notification when each of the passed
// outpoints is confirmed spent (contained in a block connected to the main
// chain) for the passed websocket client. The request is automatically
// removed once the notification has been sent.
func (m *wsNotificationManager) RegisterSpentRequests(wsc *wsClient, ops []*wire.OutPoint) {
m.queueNotification <- ¬ificationRegisterSpent{
wsc: wsc,
ops: ops,
}
}
// addSpentRequests modifies a map of watched outpoints to sets of websocket
// clients to add a new request watch all of the outpoints in ops and create
// and send a notification when spent to the websocket client wsc.
func (m *wsNotificationManager) addSpentRequests(opMap map[wire.OutPoint]map[chan struct{}]*wsClient,
wsc *wsClient, ops []*wire.OutPoint) {
for _, op := range ops {
// Track the request in the client as well so it can be quickly
// be removed on disconnect.
wsc.spentRequests[*op] = struct{}{}
// Add the client to the list to notify when the outpoint is seen.
// Create the list as needed.
cmap, ok := opMap[*op]
if !ok {
cmap = make(map[chan struct{}]*wsClient)
opMap[*op] = cmap
}
cmap[wsc.quit] = wsc
}
// Check if any transactions spending these outputs already exists in
// the mempool, if so send the notification immediately.
spends := make(map[chainhash.Hash]*btcutil.Tx)
for _, op := range ops {
spend := m.server.cfg.TxMemPool.CheckSpend(*op)
if spend != nil {
rpcsLog.Debugf("Found existing mempool spend for "+
"outpoint<%v>: %v", op, spend.Hash())
spends[*spend.Hash()] = spend
}
}
for _, spend := range spends {
m.notifyForTx(opMap, nil, spend, nil)
}
}
// UnregisterSpentRequest removes a request from the passed websocket client
// to be notified when the passed outpoint is confirmed spent (contained in a
// block connected to the main chain).
func (m *wsNotificationManager) UnregisterSpentRequest(wsc *wsClient, op *wire.OutPoint) {
m.queueNotification <- ¬ificationUnregisterSpent{
wsc: wsc,
op: op,
}
}
// removeSpentRequest modifies a map of watched outpoints to remove the
// websocket client wsc from the set of clients to be notified when a
// watched outpoint is spent. If wsc is the last client, the outpoint
// key is removed from the map.
func (*wsNotificationManager) removeSpentRequest(ops map[wire.OutPoint]map[chan struct{}]*wsClient,
wsc *wsClient, op *wire.OutPoint) {
// Remove the request tracking from the client.
delete(wsc.spentRequests, *op)
// Remove the client from the list to notify.
notifyMap, ok := ops[*op]
if !ok {
rpcsLog.Warnf("Attempt to remove nonexistent spent request "+
"for websocket client %s", wsc.addr)
return
}
delete(notifyMap, wsc.quit)
// Remove the map entry altogether if there are
// no more clients interested in it.
if len(notifyMap) == 0 {
delete(ops, *op)
}
}
// txHexString returns the serialized transaction encoded in hexadecimal.
func txHexString(tx *wire.MsgTx) string {
buf := bytes.NewBuffer(make([]byte, 0, tx.SerializeSize()))
// Ignore Serialize's error, as writing to a bytes.buffer cannot fail.
tx.Serialize(buf)
return hex.EncodeToString(buf.Bytes())
}
// blockDetails creates a BlockDetails struct to include in btcws notifications
// from a block and a transaction's block index.
func blockDetails(block *btcutil.Block, txIndex int) *btcjson.BlockDetails {
if block == nil {
return nil
}
return &btcjson.BlockDetails{
Height: block.Height(),
Hash: block.Hash().String(),
Index: txIndex,
Time: block.MsgBlock().Header.Timestamp.Unix(),
}
}
// newRedeemingTxNotification returns a new marshalled redeemingtx notification
// with the passed parameters.
func newRedeemingTxNotification(txHex string, index int, block *btcutil.Block) ([]byte, error) {
// Create and marshal the notification.
ntfn := btcjson.NewRedeemingTxNtfn(txHex, blockDetails(block, index))
return btcjson.MarshalCmd(nil, ntfn)
}
// notifyForTxOuts examines each transaction output, notifying interested
// websocket clients of the transaction if an output spends to a watched
// address. A spent notification request is automatically registered for
// the client for each matching output.
func (m *wsNotificationManager) notifyForTxOuts(ops map[wire.OutPoint]map[chan struct{}]*wsClient,
addrs map[string]map[chan struct{}]*wsClient, tx *btcutil.Tx, block *btcutil.Block) {
// Nothing to do if nobody is listening for address notifications.
if len(addrs) == 0 {
return
}
txHex := ""
wscNotified := make(map[chan struct{}]struct{})