From e32925397b3110f6dd5e18c79f6641bd2a55776f Mon Sep 17 00:00:00 2001
From: Felix Lange <fjl@twurst.com>
Date: Fri, 5 Aug 2016 13:24:48 +0200
Subject: rpc: add context argument to EthSubscribe
It's inconsistent not to pass it and most callers will
work with contexts anyway.
---
rpc/client.go | 6 ++---
rpc/client_example_test.go | 57 +++++++++++++++++++++++++---------------------
rpc/client_test.go | 8 +++----
3 files changed, 38 insertions(+), 33 deletions(-)
(limited to 'rpc')
diff --git a/rpc/client.go b/rpc/client.go
index 6846e1dda..34a3b7831 100644
--- a/rpc/client.go
+++ b/rpc/client.go
@@ -354,12 +354,14 @@ func (c *Client) BatchCallContext(ctx context.Context, b []BatchElem) error {
// sent to the given channel. The element type of the channel must match the
// expected type of content returned by the subscription.
//
+// The context argument cancels the RPC request that sets up the subscription but has no
+// effect on the subscription after EthSubscribe has returned.
//
// Slow subscribers will be dropped eventually. Client buffers up to 8000 notifications
// before considering the subscriber dead. The subscription Err channel will receive
// ErrSubscriptionQueueOverflow. Use a sufficiently large buffer on the channel or ensure
// that the channel usually has at least one reader to prevent this issue.
-func (c *Client) EthSubscribe(channel interface{}, args ...interface{}) (*ClientSubscription, error) {
+func (c *Client) EthSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error) {
// Check type of channel first.
chanVal := reflect.ValueOf(channel)
if chanVal.Kind() != reflect.Chan || chanVal.Type().ChanDir()&reflect.SendDir == 0 {
@@ -381,8 +383,6 @@ func (c *Client) EthSubscribe(channel interface{}, args ...interface{}) (*Client
resp: make(chan *jsonrpcMessage),
sub: newClientSubscription(c, chanVal),
}
- ctx, cancel := context.WithTimeout(context.Background(), subscribeTimeout)
- defer cancel()
// Send the subscription request.
// The arrival and validity of the response is signaled on sub.quit.
diff --git a/rpc/client_example_test.go b/rpc/client_example_test.go
index 84b4b67bb..3462b3685 100644
--- a/rpc/client_example_test.go
+++ b/rpc/client_example_test.go
@@ -22,6 +22,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/rpc"
+ "golang.org/x/net/context"
)
// In this example, our client whishes to track the latest 'block number'
@@ -41,7 +42,16 @@ func ExampleClientSubscription() {
// Connect the client.
client, _ := rpc.Dial("ws://127.0.0.1:8485")
subch := make(chan Block)
- go subscribeBlocks(client, subch)
+
+ // Ensure that subch receives the latest block.
+ go func() {
+ for i := 0; ; i++ {
+ if i > 0 {
+ time.Sleep(2 * time.Second)
+ }
+ subscribeBlocks(client, subch)
+ }
+ }()
// Print events from the subscription as they arrive.
for block := range subch {
@@ -52,32 +62,27 @@ func ExampleClientSubscription() {
// subscribeBlocks runs in its own goroutine and maintains
// a subscription for new blocks.
func subscribeBlocks(client *rpc.Client, subch chan Block) {
- for i := 0; ; i++ {
- if i > 0 {
- time.Sleep(2 * time.Second)
- }
-
- // Subscribe to new blocks.
- sub, err := client.EthSubscribe(subch, "newBlocks")
- if err == rpc.ErrClientQuit {
- return // Stop reconnecting if the client was closed.
- } else if err != nil {
- fmt.Println("subscribe error:", err)
- continue
- }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
- // The connection is established now.
- // Update the channel with the current block.
- var lastBlock Block
- if err := client.Call(&lastBlock, "eth_getBlockByNumber", "latest"); err != nil {
- fmt.Println("can't get latest block:", err)
- continue
- }
- subch <- lastBlock
+ // Subscribe to new blocks.
+ sub, err := client.EthSubscribe(ctx, subch, "newBlocks")
+ if err != nil {
+ fmt.Println("subscribe error:", err)
+ return
+ }
- // The subscription will deliver events to the channel. Wait for the
- // subscription to end for any reason, then loop around to re-establish
- // the connection.
- fmt.Println("connection lost: ", <-sub.Err())
+ // The connection is established now.
+ // Update the channel with the current block.
+ var lastBlock Block
+ if err := client.CallContext(ctx, &lastBlock, "eth_getBlockByNumber", "latest"); err != nil {
+ fmt.Println("can't get latest block:", err)
+ return
}
+ subch <- lastBlock
+
+ // The subscription will deliver events to the channel. Wait for the
+ // subscription to end for any reason, then loop around to re-establish
+ // the connection.
+ fmt.Println("connection lost: ", <-sub.Err())
}
diff --git a/rpc/client_test.go b/rpc/client_test.go
index 424d7f5bc..476c8c6f3 100644
--- a/rpc/client_test.go
+++ b/rpc/client_test.go
@@ -215,7 +215,7 @@ func TestClientSubscribeInvalidArg(t *testing.T) {
t.Error(string(buf))
}
}()
- client.EthSubscribe(arg, "foo_bar")
+ client.EthSubscribe(context.Background(), arg, "foo_bar")
}
check(true, nil)
check(true, 1)
@@ -233,7 +233,7 @@ func TestClientSubscribe(t *testing.T) {
nc := make(chan int)
count := 10
- sub, err := client.EthSubscribe(nc, "someSubscription", count, 0)
+ sub, err := client.EthSubscribe(context.Background(), nc, "someSubscription", count, 0)
if err != nil {
t.Fatal("can't subscribe:", err)
}
@@ -275,7 +275,7 @@ func TestClientSubscribeClose(t *testing.T) {
err error
)
go func() {
- sub, err = client.EthSubscribe(nc, "hangSubscription", 999)
+ sub, err = client.EthSubscribe(context.Background(), nc, "hangSubscription", 999)
errc <- err
}()
@@ -311,7 +311,7 @@ func TestClientNotificationStorm(t *testing.T) {
// Subscribe on the server. It will start sending many notifications
// very quickly.
nc := make(chan int)
- sub, err := client.EthSubscribe(nc, "someSubscription", count, 0)
+ sub, err := client.EthSubscribe(ctx, nc, "someSubscription", count, 0)
if err != nil {
t.Fatal("can't subscribe:", err)
}
--
cgit v1.2.3