rpc: Add SubscribeWith method for custom (un)sub methods

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 "<config.SubscribeMethod>" 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.

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)
 }