mirror of
https://github.com/fluencelabs/tendermint
synced 2025-04-25 14:52:17 +00:00
* limit number of /subscribe clients and queries per client Add the following config variables (under [rpc] section): * max_subscription_clients * max_subscriptions_per_client * timeout_broadcast_tx_commit Fixes #2826 new HTTPClient interface for subscriptions finalize HTTPClient events interface remove EventSubscriber fix data race ``` WARNING: DATA RACE Read at 0x00c000a36060 by goroutine 129: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe.func1() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:168 +0x1f0 Previous write at 0x00c000a36060 by goroutine 132: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:191 +0x4e0 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 129 (running) created at: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:164 +0x4b7 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 132 (running) created at: testing.(*T).Run() /usr/local/go/src/testing/testing.go:878 +0x659 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:119 +0x186 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 ================== ``` lite client works (tested manually) godoc comments httpclient: do not close the out channel use TimeoutBroadcastTxCommit no timeout for unsubscribe but 1s Local (5s HTTP) timeout for resubscribe format code change Subscribe#out cap to 1 and replace config vars with RPCConfig TimeoutBroadcastTxCommit can't be greater than rpcserver.WriteTimeout rpc: Context as first parameter to all functions reformat code fixes after my own review fixes after Ethan's review add test stubs fix config.toml * fixes after manual testing - rpc: do not recommend to use BroadcastTxCommit because it's slow and wastes Tendermint resources (pubsub) - rpc: better error in Subscribe and BroadcastTxCommit - HTTPClient: do not resubscribe if err = ErrAlreadySubscribed * fixes after Ismail's review * Update rpc/grpc/grpc_test.go Co-Authored-By: melekes <anton.kalyaev@gmail.com>
78 lines
2.2 KiB
Go
78 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
|
|
eventCh, err := c.Subscribe(ctx, subscriber, types.QueryForEvent(evtTyp).String())
|
|
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 event := <-eventCh:
|
|
return event.Data.(types.TMEventData), nil
|
|
case <-ctx.Done():
|
|
return nil, errors.New("timed out waiting for event")
|
|
}
|
|
}
|