spacecruft
/
blockbook
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Rename subscribeAnyAddresses to subscribeMempool

subscribeMempool
Martin Boehm 2020-05-22 00:14:51 +02:00
parent d2580674fa
commit 2c9abdce42
3 changed files with 121 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
server/public_test.go
View File

@ -1422,16 +1422,16 @@ func websocketTestsBitcoinType(t *testing.T, ts *httptest.Server) {
want: `{"id":"36","data":[{"time":1521514800,"txs":1,"received":"1","sent":"0","sentToSelf":"0","rates":{"eur":1301,"usd":2001}}]}`,
},
{
name: "websocket subscribeAnyAddresses",
name: "websocket subscribeMempool",
req: websocketReq{
Method: "subscribeAnyAddresses",
Method: "subscribeMempool",
},
want: `{"id":"37","data":{"subscribed":true}}`,
},
{
name: "websocket unsubscribeAnyAddresses",
name: "websocket unsubscribeMempool",
req: websocketReq{
Method: "unsubscribeAnyAddresses",
Method: "unsubscribeMempool",
},
want: `{"id":"38","data":{"subscribed":false}}`,
},

149
server/websocket.go
View File

@ -57,25 +57,25 @@ type websocketChannel struct {
// WebsocketServer is a handle to websocket server
type WebsocketServer struct {
socket *websocket.Conn
upgrader *websocket.Upgrader
db *db.RocksDB
txCache *db.TxCache
chain bchain.BlockChain
chainParser bchain.BlockChainParser
mempool bchain.Mempool
metrics *common.Metrics
is *common.InternalState
api *api.Worker
block0hash string
newBlockSubscriptions map[*websocketChannel]string
newBlockSubscriptionsLock sync.Mutex
addressSubscriptions map[string]map[*websocketChannel]string
addressSubscriptionsLock sync.Mutex
anyAddressSubscriptions map[*websocketChannel]string
anyAddressSubscriptionsLock sync.Mutex
fiatRatesSubscriptions map[string]map[*websocketChannel]string
fiatRatesSubscriptionsLock sync.Mutex
socket *websocket.Conn
upgrader *websocket.Upgrader
db *db.RocksDB
txCache *db.TxCache
chain bchain.BlockChain
chainParser bchain.BlockChainParser
mempool bchain.Mempool
metrics *common.Metrics
is *common.InternalState
api *api.Worker
block0hash string
newBlockSubscriptions map[*websocketChannel]string
newBlockSubscriptionsLock sync.Mutex
addressSubscriptions map[string]map[*websocketChannel]string
addressSubscriptionsLock sync.Mutex
mempoolSubscriptions map[*websocketChannel]string
mempoolSubscriptionsLock sync.Mutex
fiatRatesSubscriptions map[string]map[*websocketChannel]string
fiatRatesSubscriptionsLock sync.Mutex
}
// NewWebsocketServer creates new websocket interface to blockbook and returns its handle
@ -94,19 +94,19 @@ func NewWebsocketServer(db *db.RocksDB, chain bchain.BlockChain, mempool bchain.
WriteBufferSize: 1024 * 32,
CheckOrigin: checkOrigin,
},
db: db,
txCache: txCache,
chain: chain,
chainParser: chain.GetChainParser(),
mempool: mempool,
metrics: metrics,
is: is,
api: api,
block0hash: b0,
newBlockSubscriptions: make(map[*websocketChannel]string),
addressSubscriptions: make(map[string]map[*websocketChannel]string),
anyAddressSubscriptions: make(map[*websocketChannel]string),
fiatRatesSubscriptions: make(map[string]map[*websocketChannel]string),
db: db,
txCache: txCache,
chain: chain,
chainParser: chain.GetChainParser(),
mempool: mempool,
metrics: metrics,
is: is,
api: api,
block0hash: b0,
newBlockSubscriptions: make(map[*websocketChannel]string),
addressSubscriptions: make(map[string]map[*websocketChannel]string),
mempoolSubscriptions: make(map[*websocketChannel]string),
fiatRatesSubscriptions: make(map[string]map[*websocketChannel]string),
}
return s, nil
}
@ -336,11 +336,11 @@ var requestHandlers = map[string]func(*WebsocketServer, *websocketChannel, *webs
"unsubscribeAddresses": func(s *WebsocketServer, c *websocketChannel, req *websocketReq) (rv interface{}, err error) {
return s.unsubscribeAddresses(c)
},
"subscribeAnyAddresses": func(s *WebsocketServer, c *websocketChannel, req *websocketReq) (rv interface{}, err error) {
return s.subscribeAnyAddresses(c, req)
"subscribeMempool": func(s *WebsocketServer, c *websocketChannel, req *websocketReq) (rv interface{}, err error) {
return s.subscribeMempool(c, req)
},
"unsubscribeAnyAddresses": func(s *WebsocketServer, c *websocketChannel, req *websocketReq) (rv interface{}, err error) {
return s.unsubscribeAnyAddresses(c)
"unsubscribeMempool": func(s *WebsocketServer, c *websocketChannel, req *websocketReq) (rv interface{}, err error) {
return s.unsubscribeMempool(c)
},
"subscribeFiatRates": func(s *WebsocketServer, c *websocketChannel, req *websocketReq) (rv interface{}, err error) {
r := struct {
@ -706,23 +706,17 @@ func (s *WebsocketServer) unsubscribeAddresses(c *websocketChannel) (res interfa
return &subscriptionResponse{false}, nil
}
func (s *WebsocketServer) subscribeAnyAddresses(c *websocketChannel, req *websocketReq) (res interface{}, err error) {
// unsubscribe first
s.unsubscribeAnyAddresses(c)
s.anyAddressSubscriptionsLock.Lock()
defer s.anyAddressSubscriptionsLock.Unlock()
s.anyAddressSubscriptions[c] = req.ID
func (s *WebsocketServer) subscribeMempool(c *websocketChannel, req *websocketReq) (res interface{}, err error) {
s.mempoolSubscriptionsLock.Lock()
defer s.mempoolSubscriptionsLock.Unlock()
s.mempoolSubscriptions[c] = req.ID
return &subscriptionResponse{true}, nil
}
func (s *WebsocketServer) unsubscribeAnyAddresses(c *websocketChannel) (res interface{}, err error) {
s.anyAddressSubscriptionsLock.Lock()
defer s.anyAddressSubscriptionsLock.Unlock()
for sc := range s.anyAddressSubscriptions {
if sc == c {
delete(s.anyAddressSubscriptions, c)
}
}
func (s *WebsocketServer) unsubscribeMempool(c *websocketChannel) (res interface{}, err error) {
s.mempoolSubscriptionsLock.Lock()
defer s.mempoolSubscriptionsLock.Unlock()
delete(s.mempoolSubscriptions, c)
return &subscriptionResponse{false}, nil
}
@ -784,6 +778,51 @@ func (s *WebsocketServer) OnNewBlock(hash string, height uint32) {
glog.Info("broadcasting new block ", height, " ", hash, " to ", len(s.newBlockSubscriptions), " channels")
}
// // OnNewTxAddr is a callback that broadcasts info about a tx affecting subscribed address
// func (s *WebsocketServer) OnNewTxAddr(tx *bchain.Tx, addrDesc bchain.AddressDescriptor) {
// // check if there is any subscription but release the lock immediately, GetTransactionFromBchainTx may take some time
// s.addressSubscriptionsLock.Lock()
// as, ok := s.addressSubscriptions[string(addrDesc)]
// lenAs := len(as)
// s.addressSubscriptionsLock.Unlock()
// if ok && lenAs > 0 {
// addr, _, err := s.chainParser.GetAddressesFromAddrDesc(addrDesc)
// if err != nil {
// glog.Error("GetAddressesFromAddrDesc error ", err, " for ", addrDesc)
// return
// }
// if len(addr) == 1 {
// atx, err := s.api.GetTransactionFromBchainTx(tx, 0, false, false)
// if err != nil {
// glog.Error("GetTransactionFromBchainTx error ", err, " for ", tx.Txid)
// return
// }
// data := struct {
// Address string `json:"address"`
// Tx *api.Tx `json:"tx"`
// }{
// Address: addr[0],
// Tx: atx,
// }
// // get the list of subscriptions again, this time keep the lock
// s.addressSubscriptionsLock.Lock()
// defer s.addressSubscriptionsLock.Unlock()
// as, ok = s.addressSubscriptions[string(addrDesc)]
// if ok {
// for c, id := range as {
// if c.IsAlive() {
// c.out <- &websocketRes{
// ID: id,
// Data: &data,
// }
// }
// }
// glog.Info("broadcasting new tx ", tx.Txid, " for addr ", addr[0], " to ", len(as), " channels")
// }
// }
// }
// }
// OnNewTxAddr is a callback that broadcasts info about a tx affecting subscribed address
func (s *WebsocketServer) OnNewTxAddr(tx *bchain.Tx, addrDesc bchain.AddressDescriptor) {
// check if there is any subscription, to this address or to any address
@ -798,11 +837,11 @@ func (s *WebsocketServer) OnNewTxAddr(tx *bchain.Tx, addrDesc bchain.AddressDesc
}
s.addressSubscriptionsLock.Unlock()
// check is there is any subscription for any address
s.anyAddressSubscriptionsLock.Lock()
for c, id := range s.anyAddressSubscriptions {
s.mempoolSubscriptionsLock.Lock()
for c, id := range s.mempoolSubscriptions {
subs[c] = id
}
s.anyAddressSubscriptionsLock.Unlock()
s.mempoolSubscriptionsLock.Unlock()
if len(subs) == 0 {
return
@ -830,8 +869,8 @@ func (s *WebsocketServer) OnNewTxAddr(tx *bchain.Tx, addrDesc bchain.AddressDesc
// get the list of subscriptions again, keep locks
s.addressSubscriptionsLock.Lock()
defer s.addressSubscriptionsLock.Unlock()
s.anyAddressSubscriptionsLock.Lock()
defer s.anyAddressSubscriptionsLock.Unlock()
s.mempoolSubscriptionsLock.Lock()
defer s.mempoolSubscriptionsLock.Unlock()
for c, id := range subs {
if c.IsAlive() {
c.out <- &websocketRes{

52
static/test-websocket.html
View File

@ -56,7 +56,7 @@
subscriptions = {};
subscribeNewBlockId = "";
subscribeAddressesId = "";
subscribeAnyAddressesId = "";
subscribeMempoolId = "";
if (server.startsWith("http")) {
server = server.replace("http", "ws");
}
@ -300,33 +300,30 @@
});
}
function subscribeAnyAddresses() {
const method = 'subscribeAnyAddresses';
var addresses = document.getElementById('subscribeAnyAddressesName').value.split(",");
addresses = addresses.map(s => s.trim());
const params = {
addresses
};
if (subscribeAnyAddressesId) {
delete subscriptions[subscribeAnyAddressesId];
subscribeAnyAddressesId = "";
function subscribeMempool() {
const method = 'subscribeMempool';
if (subscribeMempoolId) {
delete subscriptions[subscribeMempoolId];
subscribeMempoolId = "";
}
subscribeAnyAddressesId = subscribe(method, params, function (result) {
document.getElementById('subscribeAnyAddressesResult').innerText += JSON.stringify(result).replace(/,/g, ", ") + "\n";
const params = {
};
subscribeMempoolId = subscribe(method, params, function (result) {
document.getElementById('subscribeMempoolResult').innerText += JSON.stringify(result).replace(/,/g, ", ") + "\n";
});
document.getElementById('subscribeAnyAddressesIds').innerText = subscribeAnyAddressesId;
document.getElementById('unsubscribeAnyAddressesButton').setAttribute("style", "display: inherit;");
document.getElementById('subscribeMempoolIds').innerText = subscribeMempoolId;
document.getElementById('unsubscribeMempoolButton').setAttribute("style", "display: inherit;");
}
function unsubscribeAnyAddresses() {
const method = 'unsubscribeAnyAddresses';
function unsubscribeMempool() {
const method = 'unsubscribeMempool';
const params = {
};
unsubscribe(method, subscribeAnyAddressesId, params, function (result) {
subscribeAnyAddressesId = "";
document.getElementById('subscribeAnyAddressesResult').innerText += JSON.stringify(result).replace(/,/g, ", ") + "\n";
document.getElementById('subscribeAnyAddressesIds').innerText = "";
document.getElementById('unsubscribeAnyAddressesButton').setAttribute("style", "display: none;");
unsubscribe(method, subscribeMempoolId, params, function (result) {
subscribeMempoolId = "";
document.getElementById('subscribeMempoolResult').innerText += JSON.stringify(result).replace(/,/g, ", ") + "\n";
document.getElementById('subscribeMempoolIds').innerText = "";
document.getElementById('unsubscribeMempoolButton').setAttribute("style", "display: none;");
});
}
@ -635,20 +632,17 @@
</div>
<div class="row">
<div class="col">
<input class="btn btn-secondary" type="button" value="subscribe address" onclick="subscribeAnyAddresses()">
</div>
<div class="col-8">
<input type="text" class="form-control" id="subscribeAnyAddressesName" value="0xba98d6a5ac827632e3457de7512d211e4ff7e8bd,0x73d0385f4d8e00c5e6504c6030f47bf6212736a8">
<input class="btn btn-secondary" type="button" value="subscribe mempool" onclick="subscribeMempool()">
</div>
<div class="col">
<span id="subscribeAnyAddressesIds"></span>
<span id="subscribeMempoolIds"></span>
</div>
<div class="col">
<input class="btn btn-secondary" id="unsubscribeAnyAddressesButton" style="display: none;" type="button" value="unsubscribe" onclick="unsubscribeAnyAddresses()">
<input class="btn btn-secondary" id="unsubscribeMempoolButton" style="display: none;" type="button" value="unsubscribe" onclick="unsubscribeMempool()">
</div>
</div>
<div class="row">
<div class="col" id="subscribeAnyAddressesResult"></div>
<div class="col" id="subscribeMempoolResult"></div>
</div>
<div class="row">
<div class="col-3">