Use btcws for parsing btcd notifications.
This commit is contained in:
parent
cdd9bea5db
commit
30db3490c0
1 changed files with 80 additions and 94 deletions
174
sockets.go
174
sockets.go
|
@ -254,6 +254,14 @@ func BtcdHandler(ws *websocket.Conn) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type notificationHandler func(btcws.Notification)
|
||||||
|
|
||||||
|
var notificationHandlers = map[string]notificationHandler{
|
||||||
|
btcws.BlockConnectedNtfnId: NtfnBlockConnected,
|
||||||
|
btcws.BlockDisconnectedNtfnId: NtfnBlockDisconnected,
|
||||||
|
btcws.TxMinedNtfnId: NtfnTxMined,
|
||||||
|
}
|
||||||
|
|
||||||
// ProcessBtcdNotificationReply unmarshalls the JSON notification or
|
// ProcessBtcdNotificationReply unmarshalls the JSON notification or
|
||||||
// reply received from btcd and decides how to handle it. Replies are
|
// reply received from btcd and decides how to handle it. Replies are
|
||||||
// routed back to the frontend who sent the message, and wallet
|
// routed back to the frontend who sent the message, and wallet
|
||||||
|
@ -330,18 +338,20 @@ func ProcessBtcdNotificationReply(b []byte) {
|
||||||
}
|
}
|
||||||
c <- b
|
c <- b
|
||||||
} else {
|
} else {
|
||||||
// btcd notification must either be handled by btcwallet or sent
|
// Message is a btcd notification. Check the id and dispatch
|
||||||
// to all frontends if btcwallet can not handle it.
|
// correct handler, or if no handler, pass up to each wallet.
|
||||||
switch idStr {
|
if ntfnHandler, ok := notificationHandlers[idStr]; ok {
|
||||||
case "btcd:blockconnected":
|
n, err := btcws.ParseMarshaledNtfn(idStr, b)
|
||||||
NtfnBlockConnected(r.Result)
|
if err != nil {
|
||||||
|
log.Errorf("Error unmarshaling expected "+
|
||||||
case "btcd:blockdisconnected":
|
"notification: %v", err)
|
||||||
NtfnBlockDisconnected(r.Result)
|
return
|
||||||
|
|
||||||
default:
|
|
||||||
frontendNotificationMaster <- b
|
|
||||||
}
|
}
|
||||||
|
ntfnHandler(n)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
frontendNotificationMaster <- b
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -359,56 +369,29 @@ func NotifyNewBlockChainHeight(reply chan []byte, height int32) {
|
||||||
|
|
||||||
// NtfnBlockConnected handles btcd notifications resulting from newly
|
// NtfnBlockConnected handles btcd notifications resulting from newly
|
||||||
// connected blocks to the main blockchain.
|
// connected blocks to the main blockchain.
|
||||||
func NtfnBlockConnected(r interface{}) {
|
func NtfnBlockConnected(n btcws.Notification) {
|
||||||
result, ok := r.(map[string]interface{})
|
bcn, ok := n.(*btcws.BlockConnectedNtfn)
|
||||||
if !ok {
|
if !ok {
|
||||||
log.Error("blockconnected notification: invalid result")
|
log.Errorf("%v handler: unexpected type", n.Id())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
hashBE, ok := result["hash"].(string)
|
hash, err := btcwire.NewShaHashFromStr(bcn.Hash)
|
||||||
if !ok {
|
|
||||||
log.Error("blockconnected notification: invalid hash")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
hash, err := btcwire.NewShaHashFromStr(hashBE)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("btcd:blockconnected handler: invalid hash string")
|
log.Errorf("%v handler: invalid hash string", n.Id())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
heightf, ok := result["height"].(float64)
|
|
||||||
if !ok {
|
|
||||||
log.Error("blockconnected notification: invalid height")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
height := int32(heightf)
|
|
||||||
var minedTxs []TXID
|
|
||||||
if iminedTxs, ok := result["minedtxs"].([]interface{}); ok {
|
|
||||||
minedTxs = make([]TXID, len(iminedTxs))
|
|
||||||
for i, iminedTx := range iminedTxs {
|
|
||||||
minedTx, ok := iminedTx.(string)
|
|
||||||
if !ok {
|
|
||||||
log.Error("blockconnected notification: mined tx is not a string")
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
minedTxs[i] = TXID(minedTx)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
curBlock.Lock()
|
|
||||||
curBlock.BlockStamp = wallet.BlockStamp{
|
|
||||||
Height: height,
|
|
||||||
Hash: *hash,
|
|
||||||
}
|
|
||||||
curBlock.Unlock()
|
|
||||||
|
|
||||||
// btcd notifies btcwallet about transactions first, and then sends
|
// btcd notifies btcwallet about transactions first, and then sends
|
||||||
// the block notification. This prevents any races from saving a
|
// the block notification. This prevents any races from saving a
|
||||||
// synced-to block before all notifications from the block have been
|
// synced-to block before all notifications from the block have been
|
||||||
// processed.
|
// processed.
|
||||||
bs := &wallet.BlockStamp{
|
bs := &wallet.BlockStamp{
|
||||||
Height: height,
|
Height: bcn.Height,
|
||||||
Hash: *hash,
|
Hash: *hash,
|
||||||
}
|
}
|
||||||
|
curBlock.Lock()
|
||||||
|
curBlock.BlockStamp = *bs
|
||||||
|
curBlock.Unlock()
|
||||||
for _, w := range wallets.m {
|
for _, w := range wallets.m {
|
||||||
// We do not write synced info immediatelly out to disk.
|
// We do not write synced info immediatelly out to disk.
|
||||||
// If btcd is performing an IBD, that would result in
|
// If btcd is performing an IBD, that would result in
|
||||||
|
@ -425,36 +408,7 @@ func NtfnBlockConnected(r interface{}) {
|
||||||
}
|
}
|
||||||
|
|
||||||
// Notify frontends of new blockchain height.
|
// Notify frontends of new blockchain height.
|
||||||
NotifyNewBlockChainHeight(frontendNotificationMaster, height)
|
NotifyNewBlockChainHeight(frontendNotificationMaster, bcn.Height)
|
||||||
|
|
||||||
// Remove all mined transactions from pool.
|
|
||||||
UnminedTxs.Lock()
|
|
||||||
for _, txid := range minedTxs {
|
|
||||||
delete(UnminedTxs.m, txid)
|
|
||||||
}
|
|
||||||
UnminedTxs.Unlock()
|
|
||||||
}
|
|
||||||
|
|
||||||
// ResendUnminedTxs resends any transactions in the unmined
|
|
||||||
// transaction pool to btcd using the 'sendrawtransaction' RPC
|
|
||||||
// command.
|
|
||||||
func resendUnminedTxs() {
|
|
||||||
for _, createdTx := range UnminedTxs.m {
|
|
||||||
n := <-NewJSONID
|
|
||||||
var id interface{} = fmt.Sprintf("btcwallet(%v)", n)
|
|
||||||
m, err := btcjson.CreateMessageWithId("sendrawtransaction", id, string(createdTx.rawTx))
|
|
||||||
if err != nil {
|
|
||||||
log.Errorf("cannot create resend request: %v", err)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
replyHandlers.Lock()
|
|
||||||
replyHandlers.m[n] = func(result interface{}, err *btcjson.Error) bool {
|
|
||||||
// Do nothing, just remove the handler.
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
replyHandlers.Unlock()
|
|
||||||
btcdMsgs <- m
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// NtfnBlockDisconnected handles btcd notifications resulting from
|
// NtfnBlockDisconnected handles btcd notifications resulting from
|
||||||
|
@ -462,35 +416,45 @@ func resendUnminedTxs() {
|
||||||
// switch and notifies frontends of the new blockchain height.
|
// switch and notifies frontends of the new blockchain height.
|
||||||
//
|
//
|
||||||
// TODO(jrick): Rollback Utxo and Tx data
|
// TODO(jrick): Rollback Utxo and Tx data
|
||||||
func NtfnBlockDisconnected(r interface{}) {
|
func NtfnBlockDisconnected(n btcws.Notification) {
|
||||||
result, ok := r.(map[string]interface{})
|
bdn, ok := n.(*btcws.BlockDisconnectedNtfn)
|
||||||
if !ok {
|
if !ok {
|
||||||
log.Error("blockdisconnected notification: invalid result")
|
log.Errorf("%v handler: unexpected type", n.Id())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
hashBE, ok := result["hash"].(string)
|
hash, err := btcwire.NewShaHashFromStr(bdn.Hash)
|
||||||
if !ok {
|
|
||||||
log.Error("blockdisconnected notification: invalid hash")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
hash, err := btcwire.NewShaHashFromStr(hashBE)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("btcd:blockdisconnected handler: invalid hash string")
|
log.Errorf("%v handler: invalid hash string", n.Id())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
heightf, ok := result["height"].(float64)
|
|
||||||
if !ok {
|
|
||||||
log.Error("blockdisconnected notification: invalid height")
|
|
||||||
}
|
|
||||||
height := int32(heightf)
|
|
||||||
|
|
||||||
// Rollback Utxo and Tx data stores.
|
// Rollback Utxo and Tx data stores.
|
||||||
go func() {
|
go func() {
|
||||||
wallets.Rollback(height, hash)
|
wallets.Rollback(bdn.Height, hash)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// Notify frontends of new blockchain height.
|
// Notify frontends of new blockchain height.
|
||||||
NotifyNewBlockChainHeight(frontendNotificationMaster, height)
|
NotifyNewBlockChainHeight(frontendNotificationMaster, bdn.Height)
|
||||||
|
}
|
||||||
|
|
||||||
|
// NtfnTxMined handles btcd notifications resulting from newly
|
||||||
|
// mined transactions that originated from this wallet.
|
||||||
|
func NtfnTxMined(n btcws.Notification) {
|
||||||
|
tmn, ok := n.(*btcws.TxMinedNtfn)
|
||||||
|
if !ok {
|
||||||
|
log.Errorf("%v handler: unexpected type", n.Id())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
hash, err := btcwire.NewShaHashFromStr(tmn.Hash)
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("%v handler: invalid hash string", n.Id())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Remove mined transaction from pool.
|
||||||
|
UnminedTxs.Lock()
|
||||||
|
delete(UnminedTxs.m, TXID(hash[:]))
|
||||||
|
UnminedTxs.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
var duplicateOnce sync.Once
|
var duplicateOnce sync.Once
|
||||||
|
@ -555,6 +519,28 @@ func BtcdConnect(reply chan error) {
|
||||||
reply <- ErrConnLost
|
reply <- ErrConnLost
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// resendUnminedTxs resends any transactions in the unmined
|
||||||
|
// transaction pool to btcd using the 'sendrawtransaction' RPC
|
||||||
|
// command.
|
||||||
|
func resendUnminedTxs() {
|
||||||
|
for _, createdTx := range UnminedTxs.m {
|
||||||
|
n := <-NewJSONID
|
||||||
|
var id interface{} = fmt.Sprintf("btcwallet(%v)", n)
|
||||||
|
m, err := btcjson.CreateMessageWithId("sendrawtransaction", id, string(createdTx.rawTx))
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("cannot create resend request: %v", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
replyHandlers.Lock()
|
||||||
|
replyHandlers.m[n] = func(result interface{}, err *btcjson.Error) bool {
|
||||||
|
// Do nothing, just remove the handler.
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
replyHandlers.Unlock()
|
||||||
|
btcdMsgs <- m
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// BtcdHandshake first checks that the websocket connection between
|
// BtcdHandshake first checks that the websocket connection between
|
||||||
// btcwallet and btcd is valid, that is, that there are no mismatching
|
// btcwallet and btcd is valid, that is, that there are no mismatching
|
||||||
// settings between the two processes (such as running on different
|
// settings between the two processes (such as running on different
|
||||||
|
|
Loading…
Reference in a new issue