mirror of
https://github.com/fluencelabs/tendermint
synced 2025-04-24 14:22:16 +00:00
90465f727f
Continues from #3280 in building support for batched requests/responses in the JSON RPC (as per issue #3213). * Add JSON RPC batching for client and server As per #3213, this adds support for [JSON RPC batch requests and responses](https://www.jsonrpc.org/specification#batch). * Add additional checks to ensure client responses are the same as results * Fix case where a notification is sent and no response is expected * Add test to check that JSON RPC notifications in a batch are left out in responses * Update CHANGELOG_PENDING.md * Update PR number now that PR has been created * Make errors start with lowercase letter * Refactor batch functionality to be standalone This refactors the batching functionality to rather act in a standalone way. In light of supporting concurrent goroutines making use of the same client, it would make sense to have batching functionality where one could create a batch of requests per goroutine and send that batch without interfering with a batch from another goroutine. * Add examples for simple and batch HTTP client usage * Check errors from writer and remove nolinter directives * Make error strings start with lowercase letter * Refactor examples to make them testable * Use safer deferred shutdown for example Tendermint test node * Recompose rpcClient interface from pre-existing interface components * Rename WaitGroup for brevity * Replace empty ID string with request ID * Remove extraneous test case * Convert first letter of errors.Wrap() messages to lowercase * Remove extraneous function parameter * Make variable declaration terse * Reorder WaitGroup.Done call to help prevent race conditions in the face of failure * Swap mutex to value representation and remove initialization * Restore empty JSONRPC string ID in response to prevent nil * Make JSONRPCBufferedRequest private * Revert PR hard link in CHANGELOG_PENDING * Add client ID for JSONRPCClient This adds code to automatically generate a randomized client ID for the JSONRPCClient, and adds a check of the IDs in the responses (if one was set in the requests). * Extract response ID validation into separate function * Remove extraneous comments * Reorder fields to indicate clearly which are protected by the mutex * Refactor for loop to remove indexing * Restructure and combine loop * Flatten conditional block for better readability * Make multi-variable declaration slightly more readable * Change for loop style * Compress error check statements * Make function description more generic to show that we support different protocols * Preallocate memory for request and result objects
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")
|
|
}
|
|
}
|