diff --git a/rpc/client.go b/rpc/client.go index 18fad8aec1..c43891bf61 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -517,6 +517,16 @@ func (c *Client) ShhSubscribe(ctx context.Context, channel interface{}, args ... // 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) Subscribe(ctx context.Context, namespace string, channel interface{}, args ...interface{}) (*ClientSubscription, error) { + return c.SubscribeWith(ctx, ClientSubscriptionConfig{ + SubscribeMethod: namespace + subscribeMethodSuffix, + UnsubscribeMethod: namespace + unsubscribeMethodSuffix, + }, channel, args...) +} + +// SubscribeWith calls the "" method with the given arguments, +// registering a new Subscription. If `config.UnsubscribeMethod` is not set, it won't +// be possible to unsubscribe. +func (c *Client) SubscribeWith(ctx context.Context, config ClientSubscriptionConfig, 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 { @@ -528,15 +538,18 @@ func (c *Client) Subscribe(ctx context.Context, namespace string, channel interf if c.isHTTP { return nil, ErrNotificationsUnsupported } + if config.SubscribeMethod == "" { + return nil, fmt.Errorf("no subscription method") + } - msg, err := c.newMessage(namespace+subscribeMethodSuffix, args) + msg, err := c.newMessage(config.SubscribeMethod, args) if err != nil { return nil, err } op := &requestOp{ ids: []json.RawMessage{msg.ID}, resp: make(chan []*jsonrpcMessage, 1), - sub: newClientSubscription(c, namespace, chanVal), + sub: newClientSubscription(c, config, chanVal), } // Send the subscription request. diff --git a/rpc/subscription.go b/rpc/subscription.go index 3231c2ceec..bb65977aa5 100644 --- a/rpc/subscription.go +++ b/rpc/subscription.go @@ -24,6 +24,7 @@ import ( "encoding/hex" "encoding/json" "errors" + "fmt" "math/rand" "reflect" "strings" @@ -211,14 +212,20 @@ func (s *Subscription) MarshalJSON() ([]byte, error) { return json.Marshal(s.ID) } +// ClientSubscriptionConfig is a subcrption configuration +type ClientSubscriptionConfig struct { + SubscribeMethod string + UnsubscribeMethod string +} + // 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 + client *Client + etype reflect.Type + channel reflect.Value + config ClientSubscriptionConfig + subid string // The in channel receives notification values from client dispatcher. in chan json.RawMessage @@ -239,10 +246,10 @@ type ClientSubscription struct { // This is the sentinel value sent on sub.quit when Unsubscribe is called. var errUnsubscribed = errors.New("unsubscribed") -func newClientSubscription(c *Client, namespace string, channel reflect.Value) *ClientSubscription { +func newClientSubscription(c *Client, config ClientSubscriptionConfig, channel reflect.Value) *ClientSubscription { sub := &ClientSubscription{ client: c, - namespace: namespace, + config: config, etype: channel.Type().Elem(), channel: channel, in: make(chan json.RawMessage), @@ -381,6 +388,9 @@ func (sub *ClientSubscription) unmarshal(result json.RawMessage) (interface{}, e } func (sub *ClientSubscription) requestUnsubscribe() error { + if sub.config.UnsubscribeMethod == "" { + return fmt.Errorf("no unsubscription method") + } var result interface{} - return sub.client.Call(&result, sub.namespace+unsubscribeMethodSuffix, sub.subid) + return sub.client.Call(&result, sub.config.UnsubscribeMethod, sub.subid) }