feat(rpc/subscription): implement state_unsubscribeStorage

dot/rpc/subscription/messages.go

@@ -28,6 +28,12 @@ type Params struct {
 	SubscriptionID uint        `json:"subscription"`
 }
 
+// InvalidRequestCode error code returned for invalid request parameters, value derived from Substrate node output
+const InvalidRequestCode = -32600
+
+// InvalidRequestMessage error message for invalid request parameters
+const InvalidRequestMessage = "Invalid request"
+
 func newSubcriptionBaseResponseJSON() BaseResponseJSON {
 	return BaseResponseJSON{
 		Jsonrpc: "2.0",
@@ -59,3 +65,18 @@ func newSubscriptionResponseJSON(subID uint, reqID float64) ResponseJSON {
 		ID:      reqID,
 	}
 }
+
+// BooleanResponseJSON for responses that return boolean values
+type BooleanResponseJSON struct {
+	Jsonrpc string  `json:"jsonrpc"`
+	Result  bool    `json:"result"`
+	ID      float64 `json:"id"`
+}
+
+func newBooleanResponseJSON(value bool, reqID float64) BooleanResponseJSON {
+	return BooleanResponseJSON{
+		Jsonrpc: "2.0",
+		Result:  value,
+		ID:      reqID,
+	}
+}

dot/rpc/subscription/websocket.go

@@ -23,10 +23,12 @@ import (
 	"io/ioutil"
 	"math/big"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 
 	"github.com/ChainSafe/gossamer/dot/rpc/modules"
+	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/common"
 	log "github.com/ChainSafe/log15"
@@ -106,6 +108,9 @@ func (c *WSConn) HandleComm() {
 					continue
 				}
 				c.startListener(rvl)
+			case "state_unsubscribeStorage":
+				c.unsubscribeStorageListener(reqid, params)
+
 			}
 			continue
 		}
@@ -212,6 +217,45 @@ func (c *WSConn) initStorageChangeListener(reqID float64, params interface{}) (u
 	return myObs.id, nil
 }
 
+func (c *WSConn) unsubscribeStorageListener(reqID float64, params interface{}) {
+	switch v := params.(type) {
+	case []interface{}:
+		if len(v) == 0 {
+			c.safeSendError(reqID, big.NewInt(InvalidRequestCode), InvalidRequestMessage)
+			return
+		}
+	default:
+		c.safeSendError(reqID, big.NewInt(InvalidRequestCode), InvalidRequestMessage)
+		return
+	}
+
+	var id uint
+	switch v := params.([]interface{})[0].(type) {
+	case float64:
+		id = uint(v)
+	case string:
+		i, err := strconv.ParseUint(v, 10, 32)
+		if err != nil {
+			c.safeSend(newBooleanResponseJSON(false, reqID))
+			return
+		}
+		id = uint(i)
+	default:
+		c.safeSendError(reqID, big.NewInt(InvalidRequestCode), InvalidRequestMessage)
+		return
+	}
+
+	observer, ok := c.Subscriptions[id].(state.Observer)
+	if !ok {
+		initRes := newBooleanResponseJSON(false, reqID)
+		c.safeSend(initRes)
+		return
+	}
+
+	c.StorageAPI.UnregisterStorageObserver(observer)
+	c.safeSend(newBooleanResponseJSON(true, reqID))
+}
+
 func (c *WSConn) initBlockListener(reqID float64) (uint, error) {
 	bl := &BlockListener{
 		Channel: make(chan *types.Block),

dot/rpc/subscription/websocket_test.go

@@ -113,6 +113,61 @@ func TestWSConn_HandleComm(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []byte(`{"jsonrpc":"2.0","result":4,"id":7}`+"\n"), msg)
 
+	// test state_unsubscribeStorage
+	c.WriteMessage(websocket.TextMessage, []byte(`{
+    "jsonrpc": "2.0",
+    "method": "state_unsubscribeStorage",
+    "params": "foo",
+    "id": 7}`))
+	_, msg, err = c.ReadMessage()
+	require.NoError(t, err)
+	require.Equal(t, []byte(`{"jsonrpc":"2.0","error":{"code":-32600,"message":"Invalid request"},"id":7}`+"\n"), msg)
+
+	c.WriteMessage(websocket.TextMessage, []byte(`{
+    "jsonrpc": "2.0",
+    "method": "state_unsubscribeStorage",
+    "params": [],
+    "id": 7}`))
+	_, msg, err = c.ReadMessage()
+	require.NoError(t, err)
+	require.Equal(t, []byte(`{"jsonrpc":"2.0","error":{"code":-32600,"message":"Invalid request"},"id":7}`+"\n"), msg)
+
+	c.WriteMessage(websocket.TextMessage, []byte(`{
+    "jsonrpc": "2.0",
+    "method": "state_unsubscribeStorage",
+    "params": ["6"],
+    "id": 7}`))
+	_, msg, err = c.ReadMessage()
+	require.NoError(t, err)
+	require.Equal(t, []byte(`{"jsonrpc":"2.0","result":false,"id":7}`+"\n"), msg)
+
+	c.WriteMessage(websocket.TextMessage, []byte(`{
+    "jsonrpc": "2.0",
+    "method": "state_unsubscribeStorage",
+    "params": ["4"],
+    "id": 7}`))
+	_, msg, err = c.ReadMessage()
+	require.NoError(t, err)
+	require.Equal(t, []byte(`{"jsonrpc":"2.0","result":true,"id":7}`+"\n"), msg)
+
+	c.WriteMessage(websocket.TextMessage, []byte(`{
+    "jsonrpc": "2.0",
+    "method": "state_unsubscribeStorage",
+    "params": [6],
+    "id": 7}`))
+	_, msg, err = c.ReadMessage()
+	require.NoError(t, err)
+	require.Equal(t, []byte(`{"jsonrpc":"2.0","result":false,"id":7}`+"\n"), msg)
+
+	c.WriteMessage(websocket.TextMessage, []byte(`{
+    "jsonrpc": "2.0",
+    "method": "state_unsubscribeStorage",
+    "params": [4],
+    "id": 7}`))
+	_, msg, err = c.ReadMessage()
+	require.NoError(t, err)
+	require.Equal(t, []byte(`{"jsonrpc":"2.0","result":true,"id":7}`+"\n"), msg)
+
 	// test initBlockListener
 	res, err = wsconn.initBlockListener(1)
 	require.EqualError(t, err, "error BlockAPI not set")