fluencelabs/tendermint
1
0
Fork 0
mirror of https://github.com/fluencelabs/tendermint synced 2025-04-26 07:12:16 +00:00
Code Issues Projects Releases Wiki Activity

fixes from Bucky's review 2

Browse Source
This commit is contained in:
Anton Kaliaev 2017-11-02 13:12:40 -05:00
parent 7c3cf316f1
commit fcdd30b2d3
No known key found for this signature in database
GPG Key ID: 7B6881D965918214
4 changed files with 13 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
rpc/client/httpclient.go
View File

@ -311,13 +311,13 @@ func (w *WSEvents) eventListener() {
continue continue
} }
result := new(ctypes.ResultEvent) result := new(ctypes.ResultEvent)
err = json.Unmarshal(*resp.Result, result) err := json.Unmarshal(*resp.Result, result)
if err != nil { if err != nil {
// ignore silently (eg. subscribe, unsubscribe and maybe other events) // ignore silently (eg. subscribe, unsubscribe and maybe other events)
// TODO: ? // TODO: ?
continue continue
} }
if ch := getSubscription(result.Query); ch != nil { if ch := w.getSubscription(result.Query); ch != nil {
ch <- result.Data ch <- result.Data
} }
case <-w.quit: case <-w.quit:

8
rpc/core/mempool.go
View File

@ -158,8 +158,8 @@ func BroadcastTxCommit(tx types.Tx) (*ctypes.ResultBroadcastTxCommit, error) {
err := eventBus.Subscribe(ctx, "mempool", q, deliverTxResCh) err := eventBus.Subscribe(ctx, "mempool", q, deliverTxResCh)
if err != nil { if err != nil {
err = errors.Wrap(err, "failed to subscribe to tx") err = errors.Wrap(err, "failed to subscribe to tx")
logger.Error("Error broadcasting transaction", "err", err) logger.Error("Error on broadcastTxCommit", "err", err)
return nil, fmt.Errorf("Error broadcasting transaction: %v", err) return nil, fmt.Errorf("Error on broadcastTxCommit: %v", err)
} }
defer eventBus.Unsubscribe(context.Background(), "mempool", q) defer eventBus.Unsubscribe(context.Background(), "mempool", q)
@ -169,8 +169,8 @@ func BroadcastTxCommit(tx types.Tx) (*ctypes.ResultBroadcastTxCommit, error) {
checkTxResCh <- res checkTxResCh <- res
}) })
if err != nil { if err != nil {
logger.Error("Error broadcasting transaction", "err", err) logger.Error("Error on broadcastTxCommit", "err", err)
return nil, fmt.Errorf("Error broadcasting transaction: %v", err) return nil, fmt.Errorf("Error on broadcastTxCommit: %v", err)
} }
checkTxRes := <-checkTxResCh checkTxRes := <-checkTxResCh
checkTxR := checkTxRes.GetCheckTx() checkTxR := checkTxRes.GetCheckTx()

2
rpc/core/pipe.go
View File

@ -50,7 +50,7 @@ var (
addrBook *p2p.AddrBook addrBook *p2p.AddrBook
txIndexer txindex.TxIndexer txIndexer txindex.TxIndexer
consensusReactor *consensus.ConsensusReactor consensusReactor *consensus.ConsensusReactor
eventBus *types.EventBus eventBus *types.EventBus // thread safe
logger log.Logger logger log.Logger
) )

16
types/event_bus.go
View File

@ -59,11 +59,9 @@ func (b *EventBus) UnsubscribeAll(ctx context.Context, subscriber string) error
} }
func (b *EventBus) Publish(eventType string, eventData TMEventData) error { func (b *EventBus) Publish(eventType string, eventData TMEventData) error {
if b.pubsub != nil { // no explicit deadline for publishing events
// no explicit deadline for publishing events ctx := context.Background()
ctx := context.Background() b.pubsub.PublishWithTags(ctx, eventData, map[string]interface{}{EventTypeKey: eventType})
b.pubsub.PublishWithTags(ctx, eventData, map[string]interface{}{EventTypeKey: eventType})
}
return nil return nil
} }
@ -82,11 +80,9 @@ func (b *EventBus) PublishEventVote(vote EventDataVote) error {
} }
func (b *EventBus) PublishEventTx(tx EventDataTx) error { func (b *EventBus) PublishEventTx(tx EventDataTx) error {
if b.pubsub != nil { // no explicit deadline for publishing events
// no explicit deadline for publishing events ctx := context.Background()
ctx := context.Background() b.pubsub.PublishWithTags(ctx, TMEventData{tx}, map[string]interface{}{EventTypeKey: EventTx, TxHashKey: fmt.Sprintf("%X", tx.Tx.Hash())})
b.pubsub.PublishWithTags(ctx, TMEventData{tx}, map[string]interface{}{EventTypeKey: EventTx, TxHashKey: fmt.Sprintf("%X", tx.Tx.Hash())})
}
return nil return nil
} }