RanchiMall/blockbook
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve ZeroMQ startup/shutdown

Browse Source
This commit is contained in:
Martin Boehm 2018-02-23 11:56:44 +01:00
parent e8b260a393
commit 531da09227
1 changed files with 14 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
bchain/mq.go
View File

@ -33,13 +33,22 @@ func NewMQ(binding string, callback func(*MQMessage)) (*MQ, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
socket.SetSubscribe("hashblock") err = socket.SetSubscribe("hashblock")
socket.SetSubscribe("hashtx") if err != nil {
return nil, err
}
err = socket.SetSubscribe("hashtx")
if err != nil {
return nil, err
}
// for now do not use raw subscriptions - we would have to handle skipped/lost notifications from zeromq // for now do not use raw subscriptions - we would have to handle skipped/lost notifications from zeromq
// on each notification we do sync or syncmempool respectively // on each notification we do sync or syncmempool respectively
// socket.SetSubscribe("rawblock") // socket.SetSubscribe("rawblock")
// socket.SetSubscribe("rawtx") // socket.SetSubscribe("rawtx")
socket.Connect(binding) err = socket.Connect(binding)
if err != nil {
return nil, err
}
glog.Info("MQ listening to ", binding) glog.Info("MQ listening to ", binding)
mq := &MQ{context, socket, true, make(chan bool)} mq := &MQ{context, socket, true, make(chan bool)}
go mq.run(callback) go mq.run(callback)
@ -78,6 +87,8 @@ func (mq *MQ) run(callback func(*MQMessage)) {
func (mq *MQ) Shutdown() error { func (mq *MQ) Shutdown() error {
glog.Info("MQ server shutdown") glog.Info("MQ server shutdown")
if mq.isRunning { if mq.isRunning {
mq.socket.SetUnsubscribe("hashtx")
mq.socket.SetUnsubscribe("hashblock")
// if errors in socket.Close or context.Term, let it close ungracefully // if errors in socket.Close or context.Term, let it close ungracefully
if err := mq.socket.Close(); err != nil { if err := mq.socket.Close(); err != nil {
return err return err