mirror of
https://github.com/fluencelabs/tendermint
synced 2025-06-30 13:11:38 +00:00
* green pubsub tests :OK:
* get rid of clientToQueryMap
* Subscribe and SubscribeUnbuffered
* start adapting other pkgs to new pubsub
* nope
* rename MsgAndTags to Message
* remove TagMap
it does not bring any additional benefits
* bring back EventSubscriber
* fix test
* fix data race in TestStartNextHeightCorrectly
```
Write at 0x00c0001c7418 by goroutine 796:
github.com/tendermint/tendermint/consensus.TestStartNextHeightCorrectly()
/go/src/github.com/tendermint/tendermint/consensus/state_test.go:1296 +0xad
testing.tRunner()
/usr/local/go/src/testing/testing.go:827 +0x162
Previous read at 0x00c0001c7418 by goroutine 858:
github.com/tendermint/tendermint/consensus.(*ConsensusState).addVote()
/go/src/github.com/tendermint/tendermint/consensus/state.go:1631 +0x1366
github.com/tendermint/tendermint/consensus.(*ConsensusState).tryAddVote()
/go/src/github.com/tendermint/tendermint/consensus/state.go:1476 +0x8f
github.com/tendermint/tendermint/consensus.(*ConsensusState).handleMsg()
/go/src/github.com/tendermint/tendermint/consensus/state.go:667 +0xa1e
github.com/tendermint/tendermint/consensus.(*ConsensusState).receiveRoutine()
/go/src/github.com/tendermint/tendermint/consensus/state.go:628 +0x794
Goroutine 796 (running) created at:
testing.(*T).Run()
/usr/local/go/src/testing/testing.go:878 +0x659
testing.runTests.func1()
/usr/local/go/src/testing/testing.go:1119 +0xa8
testing.tRunner()
/usr/local/go/src/testing/testing.go:827 +0x162
testing.runTests()
/usr/local/go/src/testing/testing.go:1117 +0x4ee
testing.(*M).Run()
/usr/local/go/src/testing/testing.go:1034 +0x2ee
main.main()
_testmain.go:214 +0x332
Goroutine 858 (running) created at:
github.com/tendermint/tendermint/consensus.(*ConsensusState).startRoutines()
/go/src/github.com/tendermint/tendermint/consensus/state.go:334 +0x221
github.com/tendermint/tendermint/consensus.startTestRound()
/go/src/github.com/tendermint/tendermint/consensus/common_test.go:122 +0x63
github.com/tendermint/tendermint/consensus.TestStateFullRound1()
/go/src/github.com/tendermint/tendermint/consensus/state_test.go:255 +0x397
testing.tRunner()
/usr/local/go/src/testing/testing.go:827 +0x162
```
* fixes after my own review
* fix formatting
* wait 100ms before kicking a subscriber out
+ a test for indexer_service
* fixes after my second review
* no timeout
* add changelog entries
* fix merge conflicts
* fix typos after Thane's review
Co-Authored-By: melekes <anton.kalyaev@gmail.com>
* reformat code
* rewrite indexer service in the attempt to fix failing test
https://github.com/tendermint/tendermint/pull/3227/#issuecomment-462316527
* Revert "rewrite indexer service in the attempt to fix failing test"
This reverts commit 0d9107a098
.
* another attempt to fix indexer
* fixes after Ethan's review
* use unbuffered channel when indexing transactions
Refs https://github.com/tendermint/tendermint/pull/3227#discussion_r258786716
* add a comment for EventBus#SubscribeUnbuffered
* format code
80 lines
2.2 KiB
Go
80 lines
2.2 KiB
Go
package client
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/pkg/errors"
|
|
"github.com/tendermint/tendermint/types"
|
|
)
|
|
|
|
// Waiter is informed of current height, decided whether to quit early
|
|
type Waiter func(delta int64) (abort error)
|
|
|
|
// DefaultWaitStrategy is the standard backoff algorithm,
|
|
// but you can plug in another one
|
|
func DefaultWaitStrategy(delta int64) (abort error) {
|
|
if delta > 10 {
|
|
return errors.Errorf("Waiting for %d blocks... aborting", delta)
|
|
} else if delta > 0 {
|
|
// estimate of wait time....
|
|
// wait half a second for the next block (in progress)
|
|
// plus one second for every full block
|
|
delay := time.Duration(delta-1)*time.Second + 500*time.Millisecond
|
|
time.Sleep(delay)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Wait for height will poll status at reasonable intervals until
|
|
// the block at the given height is available.
|
|
//
|
|
// If waiter is nil, we use DefaultWaitStrategy, but you can also
|
|
// provide your own implementation
|
|
func WaitForHeight(c StatusClient, h int64, waiter Waiter) error {
|
|
if waiter == nil {
|
|
waiter = DefaultWaitStrategy
|
|
}
|
|
delta := int64(1)
|
|
for delta > 0 {
|
|
s, err := c.Status()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
delta = h - s.SyncInfo.LatestBlockHeight
|
|
// wait for the time, or abort early
|
|
if err := waiter(delta); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// WaitForOneEvent subscribes to a websocket event for the given
|
|
// event time and returns upon receiving it one time, or
|
|
// when the timeout duration has expired.
|
|
//
|
|
// This handles subscribing and unsubscribing under the hood
|
|
func WaitForOneEvent(c EventsClient, evtTyp string, timeout time.Duration) (types.TMEventData, error) {
|
|
const subscriber = "helpers"
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
|
|
// register for the next event of this type
|
|
sub, err := c.Subscribe(ctx, subscriber, types.QueryForEvent(evtTyp))
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "failed to subscribe")
|
|
}
|
|
// make sure to unregister after the test is over
|
|
defer c.UnsubscribeAll(ctx, subscriber)
|
|
|
|
select {
|
|
case msg := <-sub.Out():
|
|
return msg.Data().(types.TMEventData), nil
|
|
case <-sub.Cancelled():
|
|
return nil, errors.New("subscription was cancelled")
|
|
case <-ctx.Done():
|
|
return nil, errors.New("timed out waiting for event")
|
|
}
|
|
}
|