LBRYFoundation/lbcd
1
0
Fork 0
mirror of https://github.com/LBRYFoundation/lbcd.git synced 2025-08-23 17:47:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Rework the btcwallet connection.

Browse source 
This changes the protocol between btcd and btcwallet to follow
JSON-RPC specifications sending notifications as requests with an
empty ID.

The notification request context handling has been greatly cleaned up
now that IDs no longer need to be saved when sending notifications.
This commit is contained in:
Josh Rickmar 2014-01-08 11:40:27 -05:00
parent a5cc3196b4
commit cd3084afcd
3 changed files with 142 additions and 224 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
btcd.go
View file

@ -5,8 +5,8 @@
package main package main
import ( import (
"github.com/conformal/btcd/limits"
"fmt" "fmt"
"github.com/conformal/btcd/limits"
"net" "net"
"net/http" "net/http"
_ "net/http/pprof" _ "net/http/pprof"

2
rpcserver.go
View file

@ -316,7 +316,7 @@ func newRPCServer(listenAddrs []string, s *server) (*rpcServer, error) {
} }
// initialize memory for websocket connections // initialize memory for websocket connections
rpc.ws.connections = make(map[chan []byte]*requestContexts) rpc.ws.connections = make(map[walletChan]*requestContexts)
rpc.ws.walletNotificationMaster = make(chan []byte) rpc.ws.walletNotificationMaster = make(chan []byte)
rpc.ws.txNotifications = make(map[string]*list.List) rpc.ws.txNotifications = make(map[string]*list.List)
rpc.ws.spentNotifications = make(map[btcwire.OutPoint]*list.List) rpc.ws.spentNotifications = make(map[btcwire.OutPoint]*list.List)

362
rpcwebsocket.go
View file

@ -8,6 +8,7 @@ import (
"bytes" "bytes"
"code.google.com/p/go.net/websocket" "code.google.com/p/go.net/websocket"
"container/list" "container/list"
"encoding/hex"
"encoding/json" "encoding/json"
"fmt" "fmt"
"github.com/conformal/btcdb" "github.com/conformal/btcdb"
@ -19,7 +20,9 @@ import (
"sync" "sync"
) )
type wsCommandHandler func(*rpcServer, btcjson.Cmd, chan []byte, *requestContexts) error type walletChan chan []byte
type wsCommandHandler func(*rpcServer, btcjson.Cmd, walletChan) error
// wsHandlers maps RPC command strings to appropriate websocket handler // wsHandlers maps RPC command strings to appropriate websocket handler
// functions. // functions.
@ -37,13 +40,13 @@ var wsHandlers = map[string]wsCommandHandler{
type wsContext struct { type wsContext struct {
sync.RWMutex sync.RWMutex
// connections holds a map of each currently connected wallet // connections holds a map of requests for each wallet using the
// listener as the key. // wallet channel as the key.
connections map[chan []byte]*requestContexts connections map[walletChan]*requestContexts
// Any chain notifications meant to be received by every connected // Any chain notifications meant to be received by every connected
// wallet are sent across this channel. // wallet are sent across this channel.
walletNotificationMaster chan []byte walletNotificationMaster walletChan
// Map of address hash to list of notificationCtx. This is the global // Map of address hash to list of notificationCtx. This is the global
// list we actually use for notifications, we also keep a list in the // list we actually use for notifications, we also keep a list in the
@ -58,41 +61,30 @@ type wsContext struct {
minedTxNotifications map[btcwire.ShaHash]*list.List minedTxNotifications map[btcwire.ShaHash]*list.List
} }
type notificationCtx struct {
id interface{}
connection chan []byte
rc *requestContexts
}
// AddTxRequest adds the request context for new transaction notifications. // AddTxRequest adds the request context for new transaction notifications.
func (r *wsContext) AddTxRequest(walletNotification chan []byte, rc *requestContexts, addr string, id interface{}) { func (r *wsContext) AddTxRequest(wallet walletChan, addr string) {
r.Lock() r.Lock()
defer r.Unlock() defer r.Unlock()
nc := &notificationCtx{
id: id,
connection: walletNotification,
rc: rc,
}
clist, ok := r.txNotifications[addr] clist, ok := r.txNotifications[addr]
if !ok { if !ok {
clist = list.New() clist = list.New()
r.txNotifications[addr] = clist r.txNotifications[addr] = clist
} }
clist.PushBack(nc) clist.PushBack(wallet)
rc.txRequests[addr] = id rc := r.connections[wallet]
rc.txRequests[addr] = struct{}{}
} }
func (r *wsContext) removeGlobalTxRequest(walletNotification chan []byte, addr string) { func (r *wsContext) removeGlobalTxRequest(wallet walletChan, addr string) {
clist := r.txNotifications[addr] clist := r.txNotifications[addr]
var enext *list.Element var enext *list.Element
for e := clist.Front(); e != nil; e = enext { for e := clist.Front(); e != nil; e = enext {
enext = e.Next() enext = e.Next()
ctx := e.Value.(*notificationCtx) c := e.Value.(walletChan)
if ctx.connection == walletNotification { if c == wallet {
clist.Remove(e) clist.Remove(e)
break break
} }
@ -105,31 +97,28 @@ func (r *wsContext) removeGlobalTxRequest(walletNotification chan []byte, addr s
// AddSpentRequest adds a request context for notifications of a spent // AddSpentRequest adds a request context for notifications of a spent
// Outpoint. // Outpoint.
func (r *wsContext) AddSpentRequest(walletNotification chan []byte, rc *requestContexts, op *btcwire.OutPoint, id interface{}) { func (r *wsContext) AddSpentRequest(wallet walletChan, op *btcwire.OutPoint) {
r.Lock() r.Lock()
defer r.Unlock() defer r.Unlock()
nc := &notificationCtx{
id: id,
connection: walletNotification,
rc: rc,
}
clist, ok := r.spentNotifications[*op] clist, ok := r.spentNotifications[*op]
if !ok { if !ok {
clist = list.New() clist = list.New()
r.spentNotifications[*op] = clist r.spentNotifications[*op] = clist
} }
clist.PushBack(nc) clist.PushBack(wallet)
rc.spentRequests[*op] = id
rc := r.connections[wallet]
rc.spentRequests[*op] = struct{}{}
} }
func (r *wsContext) removeGlobalSpentRequest(walletNotification chan []byte, op *btcwire.OutPoint) { func (r *wsContext) removeGlobalSpentRequest(wallet walletChan, op *btcwire.OutPoint) {
clist := r.spentNotifications[*op] clist := r.spentNotifications[*op]
var enext *list.Element var enext *list.Element
for e := clist.Front(); e != nil; e = enext { for e := clist.Front(); e != nil; e = enext {
enext = e.Next() enext = e.Next()
ctx := e.Value.(*notificationCtx) c := e.Value.(walletChan)
if ctx.connection == walletNotification { if c == wallet {
clist.Remove(e) clist.Remove(e)
break break
} }
@ -142,42 +131,39 @@ func (r *wsContext) removeGlobalSpentRequest(walletNotification chan []byte, op
// RemoveSpentRequest removes a request context for notifications of a // RemoveSpentRequest removes a request context for notifications of a
// spent Outpoint. // spent Outpoint.
func (r *wsContext) RemoveSpentRequest(walletNotification chan []byte, rc *requestContexts, op *btcwire.OutPoint) { func (r *wsContext) RemoveSpentRequest(wallet walletChan, op *btcwire.OutPoint) {
r.Lock() r.Lock()
defer r.Unlock() defer r.Unlock()
r.removeGlobalSpentRequest(walletNotification, op) r.removeGlobalSpentRequest(wallet, op)
rc := r.connections[wallet]
delete(rc.spentRequests, *op) delete(rc.spentRequests, *op)
} }
// AddMinedTxRequest adds request contexts for notifications of a // AddMinedTxRequest adds request contexts for notifications of a
// mined transaction. // mined transaction.
func (r *wsContext) AddMinedTxRequest(walletNotification chan []byte, txID *btcwire.ShaHash) { func (r *wsContext) AddMinedTxRequest(wallet walletChan, txID *btcwire.ShaHash) {
r.Lock() r.Lock()
defer r.Unlock() defer r.Unlock()
rc := r.connections[walletNotification]
nc := &notificationCtx{
connection: walletNotification,
rc: rc,
}
clist, ok := r.minedTxNotifications[*txID] clist, ok := r.minedTxNotifications[*txID]
if !ok { if !ok {
clist = list.New() clist = list.New()
r.minedTxNotifications[*txID] = clist r.minedTxNotifications[*txID] = clist
} }
clist.PushBack(nc) clist.PushBack(wallet)
rc.minedTxRequests[*txID] = true
rc := r.connections[wallet]
rc.minedTxRequests[*txID] = struct{}{}
} }
func (r *wsContext) removeGlobalMinedTxRequest(walletNotification chan []byte, txID *btcwire.ShaHash) { func (r *wsContext) removeGlobalMinedTxRequest(wallet walletChan, txID *btcwire.ShaHash) {
clist := r.minedTxNotifications[*txID] clist := r.minedTxNotifications[*txID]
var enext *list.Element var enext *list.Element
for e := clist.Front(); e != nil; e = enext { for e := clist.Front(); e != nil; e = enext {
enext = e.Next() enext = e.Next()
ctx := e.Value.(*notificationCtx) c := e.Value.(walletChan)
if ctx.connection == walletNotification { if c == wallet {
clist.Remove(e) clist.Remove(e)
break break
} }
@ -190,44 +176,42 @@ func (r *wsContext) removeGlobalMinedTxRequest(walletNotification chan []byte, t
// RemoveMinedTxRequest removes request contexts for notifications of a // RemoveMinedTxRequest removes request contexts for notifications of a
// mined transaction. // mined transaction.
func (r *wsContext) RemoveMinedTxRequest(walletNotification chan []byte, rc *requestContexts, txID *btcwire.ShaHash) { func (r *wsContext) RemoveMinedTxRequest(wallet walletChan, txID *btcwire.ShaHash) {
r.Lock() r.Lock()
defer r.Unlock() defer r.Unlock()
r.removeMinedTxRequest(walletNotification, rc, txID) r.removeMinedTxRequest(wallet, txID)
} }
// removeMinedTxRequest removes request contexts for notifications of a // removeMinedTxRequest removes request contexts for notifications of a
// mined transaction without grabbing any locks. // mined transaction without grabbing any locks.
func (r *wsContext) removeMinedTxRequest(walletNotification chan []byte, rc *requestContexts, txID *btcwire.ShaHash) { func (r *wsContext) removeMinedTxRequest(wallet walletChan, txID *btcwire.ShaHash) {
r.removeGlobalMinedTxRequest(walletNotification, txID) r.removeGlobalMinedTxRequest(wallet, txID)
rc := r.connections[wallet]
delete(rc.minedTxRequests, *txID) delete(rc.minedTxRequests, *txID)
} }
// CloseListeners removes all request contexts for notifications sent // CloseListeners removes all request contexts for notifications sent
// to a wallet notification channel and closes the channel to stop all // to a wallet notification channel and closes the channel to stop all
// goroutines currently serving that wallet. // goroutines currently serving that wallet.
func (r *wsContext) CloseListeners(walletNotification chan []byte) { func (r *wsContext) CloseListeners(wallet walletChan) {
r.Lock() r.Lock()
defer r.Unlock() defer r.Unlock()
delete(r.connections, walletNotification) delete(r.connections, wallet)
close(walletNotification) close(wallet)
} }
// requestContexts holds all requests for a single wallet connection. // requestContexts holds all requests for a single wallet connection.
type requestContexts struct { type requestContexts struct {
// txRequests maps between a 160-byte pubkey hash and the JSON // txRequests is a set of addresses a wallet has requested transactions
// id of the requester so replies can be correctly routed back // updates for. It is maintained here so all requests can be removed
// to the correct btcwallet callback. The key must be a stringified // when a wallet disconnects.
// address hash. txRequests map[string]struct{}
txRequests map[string]interface{}
// spentRequests maps between an Outpoint of an unspent // spentRequests is a set of unspent Outpoints a wallet has requested
// transaction output and the JSON id of the requester so // notifications for when they are spent by a processed transaction.
// replies can be correctly routed back to the correct spentRequests map[btcwire.OutPoint]struct{}
// btcwallet callback.
spentRequests map[btcwire.OutPoint]interface{}
// minedTxRequests holds a set of transaction IDs (tx hashes) of // minedTxRequests holds a set of transaction IDs (tx hashes) of
// transactions created by a wallet. A wallet may request // transactions created by a wallet. A wallet may request
@ -235,35 +219,33 @@ type requestContexts struct {
// removed from the mempool. Once a tx has been mined into a // removed from the mempool. Once a tx has been mined into a
// block, wallet may remove the raw transaction from its unmined tx // block, wallet may remove the raw transaction from its unmined tx
// pool. // pool.
minedTxRequests map[btcwire.ShaHash]bool minedTxRequests map[btcwire.ShaHash]struct{}
} }
// respondToAnyCmd checks that a parsed command is a standard or // respondToAnyCmd checks that a parsed command is a standard or
// extension JSON-RPC command and runs the proper handler to reply to // extension JSON-RPC command and runs the proper handler to reply to
// the command. Any and all responses are sent to the wallet from // the command. Any and all responses are sent to the wallet from
// this function. // this function.
func respondToAnyCmd(cmd btcjson.Cmd, s *rpcServer, func respondToAnyCmd(cmd btcjson.Cmd, s *rpcServer, wallet walletChan) {
walletNotification chan []byte, rc *requestContexts) {
// Lookup the websocket extension for the command and if it doesn't // Lookup the websocket extension for the command and if it doesn't
// exist fallback to handling the command as a standard command. // exist fallback to handling the command as a standard command.
wsHandler, ok := wsHandlers[cmd.Method()] wsHandler, ok := wsHandlers[cmd.Method()]
if !ok { if !ok {
reply := standardCmdReply(cmd, s) reply := standardCmdReply(cmd, s)
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
return return
} }
// Call the appropriate handler which responds unless there was an // Call the appropriate handler which responds unless there was an
// error in which case the error is marshalled and sent here. // error in which case the error is marshalled and sent here.
if err := wsHandler(s, cmd, walletNotification, rc); err != nil { if err := wsHandler(s, cmd, wallet); err != nil {
var reply btcjson.Reply var reply btcjson.Reply
jsonErr, ok := err.(btcjson.Error) jsonErr, ok := err.(btcjson.Error)
if ok { if ok {
reply.Error = &jsonErr reply.Error = &jsonErr
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
return return
} }
@ -276,15 +258,13 @@ func respondToAnyCmd(cmd btcjson.Cmd, s *rpcServer,
} }
reply.Error = &jsonErr reply.Error = &jsonErr
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
} }
} }
// handleGetCurrentNet implements the getcurrentnet command extension // handleGetCurrentNet implements the getcurrentnet command extension
// for websocket connections. // for websocket connections.
func handleGetCurrentNet(s *rpcServer, cmd btcjson.Cmd, func handleGetCurrentNet(s *rpcServer, cmd btcjson.Cmd, wallet walletChan) error {
walletNotification chan []byte, rc *requestContexts) error {
id := cmd.Id() id := cmd.Id()
reply := &btcjson.Reply{Id: &id} reply := &btcjson.Reply{Id: &id}
@ -297,15 +277,13 @@ func handleGetCurrentNet(s *rpcServer, cmd btcjson.Cmd,
reply.Result = float64(net) reply.Result = float64(net)
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
return nil return nil
} }
// handleGetBestBlock implements the getbestblock command extension // handleGetBestBlock implements the getbestblock command extension
// for websocket connections. // for websocket connections.
func handleGetBestBlock(s *rpcServer, cmd btcjson.Cmd, func handleGetBestBlock(s *rpcServer, cmd btcjson.Cmd, wallet walletChan) error {
walletNotification chan []byte, rc *requestContexts) error {
id := cmd.Id() id := cmd.Id()
reply := &btcjson.Reply{Id: &id} reply := &btcjson.Reply{Id: &id}
@ -322,15 +300,13 @@ func handleGetBestBlock(s *rpcServer, cmd btcjson.Cmd,
"height": height, "height": height,
} }
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
return nil return nil
} }
// handleNotifyNewTXs implements the notifynewtxs command extension for // handleNotifyNewTXs implements the notifynewtxs command extension for
// websocket connections. // websocket connections.
func handleNotifyNewTXs(s *rpcServer, cmd btcjson.Cmd, func handleNotifyNewTXs(s *rpcServer, cmd btcjson.Cmd, wallet walletChan) error {
walletNotification chan []byte, rc *requestContexts) error {
id := cmd.Id() id := cmd.Id()
reply := &btcjson.Reply{Id: &id} reply := &btcjson.Reply{Id: &id}
@ -351,20 +327,17 @@ func handleNotifyNewTXs(s *rpcServer, cmd btcjson.Cmd,
return fmt.Errorf("address is not P2PKH: %v", addr.EncodeAddress()) return fmt.Errorf("address is not P2PKH: %v", addr.EncodeAddress())
} }
s.ws.AddTxRequest(walletNotification, rc, addr.EncodeAddress(), s.ws.AddTxRequest(wallet, addr.EncodeAddress())
cmd.Id())
} }
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
return nil return nil
} }
// handleNotifySpent implements the notifyspent command extension for // handleNotifySpent implements the notifyspent command extension for
// websocket connections. // websocket connections.
func handleNotifySpent(s *rpcServer, cmd btcjson.Cmd, func handleNotifySpent(s *rpcServer, cmd btcjson.Cmd, wallet walletChan) error {
walletNotification chan []byte, rc *requestContexts) error {
id := cmd.Id() id := cmd.Id()
reply := &btcjson.Reply{Id: &id} reply := &btcjson.Reply{Id: &id}
@ -373,22 +346,16 @@ func handleNotifySpent(s *rpcServer, cmd btcjson.Cmd,
return btcjson.ErrInternal return btcjson.ErrInternal
} }
s.ws.AddSpentRequest(walletNotification, rc, notifyCmd.OutPoint, s.ws.AddSpentRequest(wallet, notifyCmd.OutPoint)
cmd.Id())
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
return nil return nil
} }
// handleRescan implements the rescan command extension for websocket // handleRescan implements the rescan command extension for websocket
// connections. // connections.
func handleRescan(s *rpcServer, cmd btcjson.Cmd, func handleRescan(s *rpcServer, cmd btcjson.Cmd, wallet walletChan) error {
walletNotification chan []byte, rc *requestContexts) error {
id := cmd.Id()
reply := &btcjson.Reply{Id: &id}
rescanCmd, ok := cmd.(*btcws.RescanCmd) rescanCmd, ok := cmd.(*btcws.RescanCmd)
if !ok { if !ok {
return btcjson.ErrInternal return btcjson.ErrInternal
@ -425,11 +392,12 @@ func handleRescan(s *rpcServer, cmd btcjson.Cmd,
} }
for _, tx := range blk.Transactions() { for _, tx := range blk.Transactions() {
var txReply *btcdb.TxListReply var txReply *btcdb.TxListReply
txouts:
for txOutIdx, txout := range tx.MsgTx().TxOut { for txOutIdx, txout := range tx.MsgTx().TxOut {
_, addrs, _, err := btcscript.ExtractPkScriptAddrs( _, addrs, _, err := btcscript.ExtractPkScriptAddrs(
txout.PkScript, s.server.btcnet) txout.PkScript, s.server.btcnet)
if err != nil { if err != nil {
continue continue txouts
} }
for i, addr := range addrs { for i, addr := range addrs {
@ -443,7 +411,7 @@ func handleRescan(s *rpcServer, cmd btcjson.Cmd,
txReplyList, err := s.server.db.FetchTxBySha(tx.Sha()) txReplyList, err := s.server.db.FetchTxBySha(tx.Sha())
if err != nil { if err != nil {
rpcsLog.Errorf("Tx Sha %v not found by db.", tx.Sha()) rpcsLog.Errorf("Tx Sha %v not found by db.", tx.Sha())
return err continue txouts
} }
for i := range txReplyList { for i := range txReplyList {
if txReplyList[i].Height == blk.Height() { if txReplyList[i].Height == blk.Height() {
@ -451,33 +419,23 @@ func handleRescan(s *rpcServer, cmd btcjson.Cmd,
break break
} }
} }
} }
reply.Result = struct { ntfn := &btcws.ProcessedTxNtfn{
Receiver string `json:"receiver"` Receiver: encodedAddr,
Height int64 `json:"height"` Amount: txout.Value,
BlockHash string `json:"blockhash"` TxID: tx.Sha().String(),
BlockIndex int `json:"blockindex"` TxOutIndex: uint32(txOutIdx),
BlockTime int64 `json:"blocktime"` PkScript: hex.EncodeToString(txout.PkScript),
TxID string `json:"txid"` BlockHash: blkshalist[i].String(),
TxOutIndex uint32 `json:"txoutindex"` BlockHeight: int32(blk.Height()),
Amount int64 `json:"amount"` BlockIndex: tx.Index(),
PkScript string `json:"pkscript"` BlockTime: blk.MsgBlock().Header.Timestamp.Unix(),
Spent bool `json:"spent"` Spent: txReply.TxSpent[txOutIdx],
}{
Receiver: encodedAddr,
Height: blk.Height(),
BlockHash: blkshalist[i].String(),
BlockIndex: tx.Index(),
BlockTime: blk.MsgBlock().Header.Timestamp.Unix(),
TxID: tx.Sha().String(),
TxOutIndex: uint32(txOutIdx),
Amount: txout.Value,
PkScript: btcutil.Base58Encode(txout.PkScript),
Spent: txReply.TxSpent[txOutIdx],
} }
mreply, _ := json.Marshal(reply) mntfn, _ := ntfn.MarshalJSON()
walletNotification <- mreply wallet <- mntfn
} }
} }
} }
@ -491,19 +449,23 @@ func handleRescan(s *rpcServer, cmd btcjson.Cmd,
} }
} }
reply.Result = nil
mreply, _ := json.Marshal(reply)
walletNotification <- mreply
rpcsLog.Info("Finished rescan") rpcsLog.Info("Finished rescan")
id := cmd.Id()
response := &btcjson.Reply{
Id: &id,
Result: nil,
Error: nil,
}
mresponse, _ := json.Marshal(response)
wallet <- mresponse
return nil return nil
} }
// handleWalletSendRawTransaction implements the websocket extended version of // handleWalletSendRawTransaction implements the websocket extended version of
// the sendrawtransaction command. // the sendrawtransaction command.
func handleWalletSendRawTransaction(s *rpcServer, cmd btcjson.Cmd, func handleWalletSendRawTransaction(s *rpcServer, cmd btcjson.Cmd, wallet walletChan) error {
walletNotification chan []byte, rc *requestContexts) error {
result, err := handleSendRawTransaction(s, cmd) result, err := handleSendRawTransaction(s, cmd)
if err != nil { if err != nil {
return err return err
@ -514,31 +476,28 @@ func handleWalletSendRawTransaction(s *rpcServer, cmd btcjson.Cmd,
txSha, _ := btcwire.NewShaHashFromStr(result.(string)) txSha, _ := btcwire.NewShaHashFromStr(result.(string))
// Request to be notified when the transaction is mined. // Request to be notified when the transaction is mined.
s.ws.AddMinedTxRequest(walletNotification, txSha) s.ws.AddMinedTxRequest(wallet, txSha)
return nil return nil
} }
// AddWalletListener adds a channel to listen for new messages from a // AddWalletListener adds a channel to listen for new messages from a
// wallet. // wallet.
func (s *rpcServer) AddWalletListener(c chan []byte) *requestContexts { func (s *rpcServer) AddWalletListener(c walletChan) {
s.ws.Lock() s.ws.Lock()
rc := &requestContexts{ rc := &requestContexts{
// The key is a stringified addressHash. txRequests: make(map[string]struct{}),
txRequests: make(map[string]interface{}), spentRequests: make(map[btcwire.OutPoint]struct{}),
minedTxRequests: make(map[btcwire.ShaHash]struct{}),
spentRequests: make(map[btcwire.OutPoint]interface{}),
minedTxRequests: make(map[btcwire.ShaHash]bool),
} }
s.ws.connections[c] = rc s.ws.connections[c] = rc
s.ws.Unlock() s.ws.Unlock()
return rc
} }
// RemoveWalletListener removes a wallet listener channel. // RemoveWalletListener removes a wallet listener channel.
func (s *rpcServer) RemoveWalletListener(c chan []byte, rc *requestContexts) { func (s *rpcServer) RemoveWalletListener(c walletChan) {
s.ws.Lock() s.ws.Lock()
rc := s.ws.connections[c]
for k := range rc.txRequests { for k := range rc.txRequests {
s.ws.removeGlobalTxRequest(c, k) s.ws.removeGlobalTxRequest(c, k)
} }
@ -580,9 +539,9 @@ func (s *rpcServer) walletListenerDuplicator() {
func (s *rpcServer) walletReqsNotifications(ws *websocket.Conn) { func (s *rpcServer) walletReqsNotifications(ws *websocket.Conn) {
// Add wallet notification channel so this handler receives btcd chain // Add wallet notification channel so this handler receives btcd chain
// notifications. // notifications.
c := make(chan []byte) c := make(walletChan)
rc := s.AddWalletListener(c) s.AddWalletListener(c)
defer s.RemoveWalletListener(c, rc) defer s.RemoveWalletListener(c)
// msgs is a channel for all messages received over the websocket. // msgs is a channel for all messages received over the websocket.
msgs := make(chan []byte) msgs := make(chan []byte)
@ -614,7 +573,7 @@ func (s *rpcServer) walletReqsNotifications(ws *websocket.Conn) {
return return
} }
// Handle request here. // Handle request here.
go s.websocketJSONHandler(c, rc, m) go s.websocketJSONHandler(c, m)
case ntfn, _ := <-c: case ntfn, _ := <-c:
// Send btcd notification to btcwallet instance over // Send btcd notification to btcwallet instance over
// websocket. // websocket.
@ -631,9 +590,7 @@ func (s *rpcServer) walletReqsNotifications(ws *websocket.Conn) {
// websocketJSONHandler parses and handles a marshalled json message, // websocketJSONHandler parses and handles a marshalled json message,
// sending the marshalled reply to a wallet notification channel. // sending the marshalled reply to a wallet notification channel.
func (s *rpcServer) websocketJSONHandler(walletNotification chan []byte, func (s *rpcServer) websocketJSONHandler(wallet walletChan, msg []byte) {
rc *requestContexts, msg []byte) {
s.wg.Add(1) s.wg.Add(1)
defer s.wg.Done() defer s.wg.Done()
@ -648,11 +605,11 @@ func (s *rpcServer) websocketJSONHandler(walletNotification chan []byte,
} }
reply.Error = jsonErr reply.Error = jsonErr
mreply, _ := json.Marshal(reply) mreply, _ := json.Marshal(reply)
walletNotification <- mreply wallet <- mreply
return return
} }
respondToAnyCmd(cmd, s, walletNotification, rc) respondToAnyCmd(cmd, s, wallet)
} }
// NotifyBlockConnected creates and marshalls a JSON message to notify // NotifyBlockConnected creates and marshalls a JSON message to notify
@ -678,7 +635,7 @@ func (s *rpcServer) NotifyBlockConnected(block *btcutil.Block) {
var enext *list.Element var enext *list.Element
for e := clist.Front(); e != nil; e = enext { for e := clist.Front(); e != nil; e = enext {
enext = e.Next() enext = e.Next()
ctx := e.Value.(*notificationCtx) c := e.Value.(walletChan)
// TODO: remove int32 type conversion after // TODO: remove int32 type conversion after
// the int64 -> int32 switch is made. // the int64 -> int32 switch is made.
ntfn := btcws.NewTxMinedNtfn(tx.Sha().String(), ntfn := btcws.NewTxMinedNtfn(tx.Sha().String(),
@ -686,9 +643,8 @@ func (s *rpcServer) NotifyBlockConnected(block *btcutil.Block) {
block.MsgBlock().Header.Timestamp.Unix(), block.MsgBlock().Header.Timestamp.Unix(),
tx.Index()) tx.Index())
mntfn, _ := json.Marshal(ntfn) mntfn, _ := json.Marshal(ntfn)
ctx.connection <- mntfn c <- mntfn
s.ws.removeMinedTxRequest(ctx.connection, ctx.rc, s.ws.removeMinedTxRequest(c, tx.Sha())
tx.Sha())
} }
} }
} }
@ -722,39 +678,19 @@ func (s *rpcServer) NotifyBlockTXs(db btcdb.Db, block *btcutil.Block) {
} }
} }
func notifySpentData(ctx *notificationCtx, txhash *btcwire.ShaHash, index uint32, func notifySpentData(wallet walletChan, txhash *btcwire.ShaHash, index uint32,
spender *btcutil.Tx) { spender *btcutil.Tx) {
txStr := ""
if spender != nil {
var buf bytes.Buffer
err := spender.MsgTx().Serialize(&buf)
if err != nil {
// This really shouldn't ever happen...
rpcsLog.Warnf("Can't serialize tx: %v", err)
return
}
txStr = string(buf.Bytes())
}
reply := &btcjson.Reply{ var buf bytes.Buffer
Result: struct { // Ignore Serialize's error, as writing to a bytes.buffer
TxHash string `json:"txhash"` // cannot fail.
Index uint32 `json:"index"` spender.MsgTx().Serialize(&buf)
SpendingTx string `json:"spender,omitempty"` txStr := hex.EncodeToString(buf.Bytes())
}{
TxHash: txhash.String(), // TODO(jrick): create a new notification in btcws and use that.
Index: index, ntfn := btcws.NewTxSpentNtfn(txhash.String(), int(index), txStr)
SpendingTx: txStr, mntfn, _ := ntfn.MarshalJSON()
}, wallet <- mntfn
Error: nil,
Id: &ctx.id,
}
replyBytes, err := json.Marshal(reply)
if err != nil {
rpcsLog.Errorf("Unable to marshal spent notification: %v", err)
return
}
ctx.connection <- replyBytes
} }
// newBlockNotifyCheckTxIn is a helper function to iterate through // newBlockNotifyCheckTxIn is a helper function to iterate through
@ -766,11 +702,10 @@ func (s *rpcServer) newBlockNotifyCheckTxIn(tx *btcutil.Tx) {
var enext *list.Element var enext *list.Element
for e := clist.Front(); e != nil; e = enext { for e := clist.Front(); e != nil; e = enext {
enext = e.Next() enext = e.Next()
ctx := e.Value.(*notificationCtx) c := e.Value.(walletChan)
notifySpentData(ctx, &txin.PreviousOutpoint.Hash, notifySpentData(c, &txin.PreviousOutpoint.Hash,
uint32(txin.PreviousOutpoint.Index), tx) txin.PreviousOutpoint.Index, tx)
s.ws.RemoveSpentRequest(ctx.connection, ctx.rc, s.ws.RemoveSpentRequest(c, &txin.PreviousOutpoint)
&txin.PreviousOutpoint)
} }
} }
} }
@ -796,25 +731,17 @@ func (s *rpcServer) NotifyForTxOuts(tx *btcutil.Tx, block *btcutil.Block) {
encodedAddr := addr.EncodeAddress() encodedAddr := addr.EncodeAddress()
if idlist, ok := s.ws.txNotifications[encodedAddr]; ok { if idlist, ok := s.ws.txNotifications[encodedAddr]; ok {
for e := idlist.Front(); e != nil; e = e.Next() { for e := idlist.Front(); e != nil; e = e.Next() {
ctx := e.Value.(*notificationCtx) wallet := e.Value.(walletChan)
// TODO(jrick): shove this in btcws ntfn := &btcws.ProcessedTxNtfn{
result := struct {
Receiver string `json:"receiver"`
Height int64 `json:"height"`
BlockHash string `json:"blockhash"`
BlockIndex int `json:"blockindex"`
BlockTime int64 `json:"blocktime"`
TxID string `json:"txid"`
TxOutIndex uint32 `json:"txoutindex"`
Amount int64 `json:"amount"`
PkScript string `json:"pkscript"`
}{
Receiver: encodedAddr, Receiver: encodedAddr,
TxID: tx.Sha().String(), TxID: tx.Sha().String(),
TxOutIndex: uint32(i), TxOutIndex: uint32(i),
Amount: txout.Value, Amount: txout.Value,
PkScript: btcutil.Base58Encode(txout.PkScript), PkScript: hex.EncodeToString(txout.PkScript),
// TODO(jrick): hardcoding unspent is WRONG and needs
// to be either calculated from other block txs, or dropped.
Spent: false,
} }
if block != nil { if block != nil {
@ -823,26 +750,17 @@ func (s *rpcServer) NotifyForTxOuts(tx *btcutil.Tx, block *btcutil.Block) {
rpcsLog.Error("Error getting block sha; dropping Tx notification.") rpcsLog.Error("Error getting block sha; dropping Tx notification.")
break break
} }
result.Height = block.Height() ntfn.BlockHeight = int32(block.Height())
result.BlockHash = blkhash.String() ntfn.BlockHash = blkhash.String()
result.BlockIndex = tx.Index() ntfn.BlockIndex = tx.Index()
result.BlockTime = block.MsgBlock().Header.Timestamp.Unix() ntfn.BlockTime = block.MsgBlock().Header.Timestamp.Unix()
} else { } else {
result.Height = -1 ntfn.BlockHeight = -1
result.BlockIndex = -1 ntfn.BlockIndex = -1
} }
reply := &btcjson.Reply{ mntfn, _ := ntfn.MarshalJSON()
Result: result, wallet <- mntfn
Error: nil,
Id: &ctx.id,
}
mreply, err := json.Marshal(reply)
if err != nil {
rpcsLog.Errorf("Unable to marshal tx notification: %v", err)
continue
}
ctx.connection <- mreply
} }
} }
} }