aboutsummaryrefslogtreecommitdiff
path: root/rpc/subscription.go
diff options
context:
space:
mode:
authorDeterminant <[email protected]>2020-06-28 14:47:41 -0400
committerDeterminant <[email protected]>2020-06-28 14:47:41 -0400
commitd235e2c6a5788ec4a6cff15a16f56b38a3876a0d (patch)
tree5f2727f7a50ee5840f889c82776d3a30a88dd59b /rpc/subscription.go
parent13ebd8bd9468e9d769d598b0ca2afb72ba78cb97 (diff)
...
Diffstat (limited to 'rpc/subscription.go')
-rw-r--r--rpc/subscription.go327
1 files changed, 327 insertions, 0 deletions
diff --git a/rpc/subscription.go b/rpc/subscription.go
new file mode 100644
index 0000000..c1e869b
--- /dev/null
+++ b/rpc/subscription.go
@@ -0,0 +1,327 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package rpc
+
+import (
+ "bufio"
+ "container/list"
+ "context"
+ crand "crypto/rand"
+ "encoding/binary"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "math/rand"
+ "reflect"
+ "strings"
+ "sync"
+ "time"
+)
+
+var (
+ // ErrNotificationsUnsupported is returned when the connection doesn't support notifications
+ ErrNotificationsUnsupported = errors.New("notifications not supported")
+ // ErrNotificationNotFound is returned when the notification for the given id is not found
+ ErrSubscriptionNotFound = errors.New("subscription not found")
+)
+
+var globalGen = randomIDGenerator()
+
+// ID defines a pseudo random number that is used to identify RPC subscriptions.
+type ID string
+
+// NewID returns a new, random ID.
+func NewID() ID {
+ return globalGen()
+}
+
+// randomIDGenerator returns a function generates a random IDs.
+func randomIDGenerator() func() ID {
+ seed, err := binary.ReadVarint(bufio.NewReader(crand.Reader))
+ if err != nil {
+ seed = int64(time.Now().Nanosecond())
+ }
+ var (
+ mu sync.Mutex
+ rng = rand.New(rand.NewSource(seed))
+ )
+ return func() ID {
+ mu.Lock()
+ defer mu.Unlock()
+ id := make([]byte, 16)
+ rng.Read(id)
+ return encodeID(id)
+ }
+}
+
+func encodeID(b []byte) ID {
+ id := hex.EncodeToString(b)
+ id = strings.TrimLeft(id, "0")
+ if id == "" {
+ id = "0" // ID's are RPC quantities, no leading zero's and 0 is 0x0.
+ }
+ return ID("0x" + id)
+}
+
+type notifierKey struct{}
+
+// NotifierFromContext returns the Notifier value stored in ctx, if any.
+func NotifierFromContext(ctx context.Context) (*Notifier, bool) {
+ n, ok := ctx.Value(notifierKey{}).(*Notifier)
+ return n, ok
+}
+
+// Notifier is tied to a RPC connection that supports subscriptions.
+// Server callbacks use the notifier to send notifications.
+type Notifier struct {
+ h *handler
+ namespace string
+
+ mu sync.Mutex
+ sub *Subscription
+ buffer []json.RawMessage
+ callReturned bool
+ activated bool
+}
+
+// CreateSubscription returns a new subscription that is coupled to the
+// RPC connection. By default subscriptions are inactive and notifications
+// are dropped until the subscription is marked as active. This is done
+// by the RPC server after the subscription ID is send to the client.
+func (n *Notifier) CreateSubscription() *Subscription {
+ n.mu.Lock()
+ defer n.mu.Unlock()
+
+ if n.sub != nil {
+ panic("can't create multiple subscriptions with Notifier")
+ } else if n.callReturned {
+ panic("can't create subscription after subscribe call has returned")
+ }
+ n.sub = &Subscription{ID: n.h.idgen(), namespace: n.namespace, err: make(chan error, 1)}
+ return n.sub
+}
+
+// Notify sends a notification to the client with the given data as payload.
+// If an error occurs the RPC connection is closed and the error is returned.
+func (n *Notifier) Notify(id ID, data interface{}) error {
+ enc, err := json.Marshal(data)
+ if err != nil {
+ return err
+ }
+
+ n.mu.Lock()
+ defer n.mu.Unlock()
+
+ if n.sub == nil {
+ panic("can't Notify before subscription is created")
+ } else if n.sub.ID != id {
+ panic("Notify with wrong ID")
+ }
+ if n.activated {
+ return n.send(n.sub, enc)
+ }
+ n.buffer = append(n.buffer, enc)
+ return nil
+}
+
+// Closed returns a channel that is closed when the RPC connection is closed.
+// Deprecated: use subscription error channel
+func (n *Notifier) Closed() <-chan interface{} {
+ return n.h.conn.Closed()
+}
+
+// takeSubscription returns the subscription (if one has been created). No subscription can
+// be created after this call.
+func (n *Notifier) takeSubscription() *Subscription {
+ n.mu.Lock()
+ defer n.mu.Unlock()
+ n.callReturned = true
+ return n.sub
+}
+
+// acticate is called after the subscription ID was sent to client. Notifications are
+// buffered before activation. This prevents notifications being sent to the client before
+// the subscription ID is sent to the client.
+func (n *Notifier) activate() error {
+ n.mu.Lock()
+ defer n.mu.Unlock()
+
+ for _, data := range n.buffer {
+ if err := n.send(n.sub, data); err != nil {
+ return err
+ }
+ }
+ n.activated = true
+ return nil
+}
+
+func (n *Notifier) send(sub *Subscription, data json.RawMessage) error {
+ params, _ := json.Marshal(&subscriptionResult{ID: string(sub.ID), Result: data})
+ ctx := context.Background()
+ return n.h.conn.Write(ctx, &jsonrpcMessage{
+ Version: vsn,
+ Method: n.namespace + notificationMethodSuffix,
+ Params: params,
+ })
+}
+
+// A Subscription is created by a notifier and tight to that notifier. The client can use
+// this subscription to wait for an unsubscribe request for the client, see Err().
+type Subscription struct {
+ ID ID
+ namespace string
+ err chan error // closed on unsubscribe
+}
+
+// Err returns a channel that is closed when the client send an unsubscribe request.
+func (s *Subscription) Err() <-chan error {
+ return s.err
+}
+
+// MarshalJSON marshals a subscription as its ID.
+func (s *Subscription) MarshalJSON() ([]byte, error) {
+ return json.Marshal(s.ID)
+}
+
+// ClientSubscription is a subscription established through the Client's Subscribe or
+// EthSubscribe methods.
+type ClientSubscription struct {
+ client *Client
+ etype reflect.Type
+ channel reflect.Value
+ namespace string
+ subid string
+ in chan json.RawMessage
+
+ quitOnce sync.Once // ensures quit is closed once
+ quit chan struct{} // quit is closed when the subscription exits
+ errOnce sync.Once // ensures err is closed once
+ err chan error
+}
+
+func newClientSubscription(c *Client, namespace string, channel reflect.Value) *ClientSubscription {
+ sub := &ClientSubscription{
+ client: c,
+ namespace: namespace,
+ etype: channel.Type().Elem(),
+ channel: channel,
+ quit: make(chan struct{}),
+ err: make(chan error, 1),
+ in: make(chan json.RawMessage),
+ }
+ return sub
+}
+
+// Err returns the subscription error channel. The intended use of Err is to schedule
+// resubscription when the client connection is closed unexpectedly.
+//
+// The error channel receives a value when the subscription has ended due
+// to an error. The received error is nil if Close has been called
+// on the underlying client and no other error has occurred.
+//
+// The error channel is closed when Unsubscribe is called on the subscription.
+func (sub *ClientSubscription) Err() <-chan error {
+ return sub.err
+}
+
+// Unsubscribe unsubscribes the notification and closes the error channel.
+// It can safely be called more than once.
+func (sub *ClientSubscription) Unsubscribe() {
+ sub.quitWithError(nil, true)
+ sub.errOnce.Do(func() { close(sub.err) })
+}
+
+func (sub *ClientSubscription) quitWithError(err error, unsubscribeServer bool) {
+ sub.quitOnce.Do(func() {
+ // The dispatch loop won't be able to execute the unsubscribe call
+ // if it is blocked on deliver. Close sub.quit first because it
+ // unblocks deliver.
+ close(sub.quit)
+ if unsubscribeServer {
+ sub.requestUnsubscribe()
+ }
+ if err != nil {
+ if err == ErrClientQuit {
+ err = nil // Adhere to subscription semantics.
+ }
+ sub.err <- err
+ }
+ })
+}
+
+func (sub *ClientSubscription) deliver(result json.RawMessage) (ok bool) {
+ select {
+ case sub.in <- result:
+ return true
+ case <-sub.quit:
+ return false
+ }
+}
+
+func (sub *ClientSubscription) start() {
+ sub.quitWithError(sub.forward())
+}
+
+func (sub *ClientSubscription) forward() (err error, unsubscribeServer bool) {
+ cases := []reflect.SelectCase{
+ {Dir: reflect.SelectRecv, Chan: reflect.ValueOf(sub.quit)},
+ {Dir: reflect.SelectRecv, Chan: reflect.ValueOf(sub.in)},
+ {Dir: reflect.SelectSend, Chan: sub.channel},
+ }
+ buffer := list.New()
+ defer buffer.Init()
+ for {
+ var chosen int
+ var recv reflect.Value
+ if buffer.Len() == 0 {
+ // Idle, omit send case.
+ chosen, recv, _ = reflect.Select(cases[:2])
+ } else {
+ // Non-empty buffer, send the first queued item.
+ cases[2].Send = reflect.ValueOf(buffer.Front().Value)
+ chosen, recv, _ = reflect.Select(cases)
+ }
+
+ switch chosen {
+ case 0: // <-sub.quit
+ return nil, false
+ case 1: // <-sub.in
+ val, err := sub.unmarshal(recv.Interface().(json.RawMessage))
+ if err != nil {
+ return err, true
+ }
+ if buffer.Len() == maxClientSubscriptionBuffer {
+ return ErrSubscriptionQueueOverflow, true
+ }
+ buffer.PushBack(val)
+ case 2: // sub.channel<-
+ cases[2].Send = reflect.Value{} // Don't hold onto the value.
+ buffer.Remove(buffer.Front())
+ }
+ }
+}
+
+func (sub *ClientSubscription) unmarshal(result json.RawMessage) (interface{}, error) {
+ val := reflect.New(sub.etype)
+ err := json.Unmarshal(result, val.Interface())
+ return val.Elem().Interface(), err
+}
+
+func (sub *ClientSubscription) requestUnsubscribe() error {
+ var result interface{}
+ return sub.client.Call(&result, sub.namespace+unsubscribeMethodSuffix, sub.subid)
+}