dot/rpc implement state_subscribeStorage RPC WebSocket method

dot/rpc/http.go

@@ -22,6 +22,7 @@ import (
 	"os"
 
 	"github.com/ChainSafe/gossamer/dot/rpc/modules"
+	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/gorilla/mux"
 	"github.com/gorilla/rpc/v2"
@@ -32,11 +33,13 @@ import (
 
 // HTTPServer gateway for RPC server
 type HTTPServer struct {
-	logger       log.Logger
-	rpcServer    *rpc.Server // Actual RPC call handler
-	serverConfig *HTTPServerConfig
-	blockChan    chan *types.Block
-	chanID       byte // channel ID
+	logger        log.Logger
+	rpcServer     *rpc.Server // Actual RPC call handler
+	serverConfig  *HTTPServerConfig
+	blockChan     chan *types.Block
+	chanID        byte // channel ID
+	storageChan   chan *state.KeyValue
+	storageChanID byte // storage channel ID
 }
 
 // HTTPServerConfig configures the HTTPServer
@@ -63,6 +66,7 @@ type HTTPServerConfig struct {
 type WebSocketSubscription struct {
 	WSConnection     *websocket.Conn
 	SubscriptionType int
+	Filter           map[string]bool
 }
 
 // NewHTTPServer creates a new http server and registers an associated rpc server
@@ -162,6 +166,17 @@ func (h *HTTPServer) Start() error {
 		go h.blockReceivedListener()
 	}
 
+	// init and start storage change listener routine
+	if h.serverConfig.StorageAPI != nil {
+		var err error
+		h.storageChan = make(chan *state.KeyValue)
+		h.storageChanID, err = h.serverConfig.StorageAPI.RegisterStorageChangeChannel(h.storageChan)
+		if err != nil {
+			return err
+		}
+		go h.storageChangeListener()
+	}
+
 	return nil
 }
 
@@ -170,6 +185,9 @@ func (h *HTTPServer) Stop() error {
 	if h.serverConfig.WSEnabled {
 		h.serverConfig.BlockAPI.UnregisterImportedChannel(h.chanID)
 		close(h.blockChan)
+
+		h.serverConfig.StorageAPI.UnregisterStorageChangeChannel(h.storageChanID)
+		close(h.storageChan)
 	}
 	return nil
 }

dot/rpc/modules/api.go

@@ -3,6 +3,7 @@ package modules
 import (
 	"math/big"
 
+	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/common"
 	"github.com/ChainSafe/gossamer/lib/crypto"
@@ -14,6 +15,8 @@ import (
 type StorageAPI interface {
 	GetStorage(key []byte) ([]byte, error)
 	Entries() map[string][]byte
+	RegisterStorageChangeChannel(ch chan<- *state.KeyValue) (byte, error)
+	UnregisterStorageChangeChannel(id byte)
 }
 
 // BlockAPI is the interface for the block state

dot/rpc/modules/state.go

@@ -285,9 +285,11 @@ func (sm *StateModule) SubscribeRuntimeVersion(r *http.Request, req *StateStorag
 	return sm.GetRuntimeVersion(r, nil, res)
 }
 
-// SubscribeStorage isn't implemented properly yet.
-func (sm *StateModule) SubscribeStorage(r *http.Request, req *StateStorageQueryRangeRequest, res *StorageChangeSetResponse) {
-	// TODO implement change storage trie so that block hash parameter works (See issue #834)
+// SubscribeStorage Storage subscription. If storage keys are specified, it creates a message for each block which
+//  changes the specified storage keys. If none are specified, then it creates a message for every block.
+//  This endpoint communicates over the Websockt protocol, but this func should remain here so it's added to rpc_methods list
+func (sm *StateModule) SubscribeStorage(r *http.Request, req *StateStorageQueryRangeRequest, res *StorageChangeSetResponse) error {
+	return nil
 }
 
 func convertAPIs(in []*runtime.API_Item) []interface{} {

dot/rpc/websocket.go

@@ -25,7 +25,7 @@ import (
 	"strings"
 
 	"github.com/ChainSafe/gossamer/dot/rpc/modules"
-
+	"github.com/ChainSafe/gossamer/lib/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/gorilla/websocket"
 )
@@ -128,15 +128,16 @@ func (h *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			switch method {
 			case "chain_subscribeNewHeads", "chain_subscribeNewHead":
 				subType = SUB_NEW_HEAD
-			case "chain_subscribeStorage":
+			case "state_subscribeStorage":
 				subType = SUB_STORAGE
 			case "chain_subscribeFinalizedHeads":
 				subType = SUB_FINALIZED_HEAD
 			}
-
+			params := msg["params"]
 			var e1 error
-			_, e1 = h.registerSubscription(ws, mid, subType)
+			_, e1 = h.registerSubscription(ws, mid, subType, params)
 			if e1 != nil {
+				// todo send error message to client
 				log.Error("[rpc] failed to register subscription", "error", err)
 			}
 			continue
@@ -191,12 +192,18 @@ func (h *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 }
 
-func (h *HTTPServer) registerSubscription(conn *websocket.Conn, reqID float64, subscriptionType int) (uint32, error) {
+func (h *HTTPServer) registerSubscription(conn *websocket.Conn, reqID float64, subscriptionType int, params interface{}) (uint32, error) {
 	wssub := h.serverConfig.WSSubscriptions
 	sub := uint32(len(wssub)) + 1
+	pA := params.([]interface{})
+	filter := make(map[string]bool)
+	for _, param := range pA {
+		filter[param.(string)] = true
+	}
 	wss := &WebSocketSubscription{
 		WSConnection:     conn,
 		SubscriptionType: subscriptionType,
+		Filter:           filter,
 	}
 	wssub[sub] = wss
 	h.serverConfig.WSSubscriptions = wssub
@@ -233,3 +240,36 @@ func (h *HTTPServer) blockReceivedListener() {
 		}
 	}
 }
+
+func (h *HTTPServer) storageChangeListener() {
+	if h.serverConfig.StorageAPI == nil {
+		return
+	}
+
+	for change := range h.storageChan {
+
+		if change != nil {
+			for i, sub := range h.serverConfig.WSSubscriptions {
+				if sub.SubscriptionType == SUB_STORAGE {
+					// check it change key is in subscription filter
+					cKey := common.BytesToHex(change.Key)
+					if len(sub.Filter) > 0 && !sub.Filter[cKey] {
+						continue
+					}
+
+					changeM := make(map[string]interface{})
+					changeM["result"] = []string{cKey, common.BytesToHex(change.Value)}
+					res := newSubcriptionBaseResponseJSON(i)
+					res.Method = "state_storage"
+					res.Params = changeM
+					if sub.WSConnection != nil {
+						err := sub.WSConnection.WriteJSON(res)
+						if err != nil {
+							log.Error("[rpc] error writing response", "error", err)
+						}
+					}
+				}
+			}
+		}
+	}
+}

dot/rpc/websocket_test.go

@@ -20,9 +20,10 @@ var testCalls = []struct {
 	expected []byte
 }{
 	{[]byte(`{"jsonrpc":"2.0","method":"system_name","params":[],"id":1}`), []byte(`{"id":1,"jsonrpc":"2.0","result":"gossamer"}` + "\n")},                                                            // working request
-	{[]byte(`{"jsonrpc":"2.0","method":"unknown","params":[],"id":1}`), []byte(`{"error":{"code":-32000,"data":null,"message":"rpc error method unknown not found"},"id":1,"jsonrpc":"2.0"}` + "\n")}, // unknown method
+	{[]byte(`{"jsonrpc":"2.0","method":"unknown","params":[],"id":2}`), []byte(`{"error":{"code":-32000,"data":null,"message":"rpc error method unknown not found"},"id":2,"jsonrpc":"2.0"}` + "\n")}, // unknown method
 	{[]byte{}, []byte(`{"jsonrpc":"2.0","error":{"code":-32600,"message":"Invalid request"},"id":null}` + "\n")},                                                                                      // empty request
-	{[]byte(`{"jsonrpc":"2.0","method":"chain_subscribeNewHeads","params":[],"id":1}`), []byte(`{"jsonrpc":"2.0","result":1,"id":1}` + "\n")},
+	{[]byte(`{"jsonrpc":"2.0","method":"chain_subscribeNewHeads","params":[],"id":3}`), []byte(`{"jsonrpc":"2.0","result":1,"id":3}` + "\n")},
+	{[]byte(`{"jsonrpc":"2.0","method":"state_subscribeStorage","params":[],"id":4}`), []byte(`{"jsonrpc":"2.0","result":2,"id":4}` + "\n")},
 }
 
 func TestNewWebSocketServer(t *testing.T) {

dot/state/storage.go

@@ -52,6 +52,10 @@ type StorageState struct {
 	trie *trie.Trie
 	db   *StorageDB
 	lock sync.RWMutex
+
+	// change notifiers
+	changed     map[byte]chan<- *KeyValue
+	changedLock sync.RWMutex
 }
 
 // NewStorageDB instantiates badgerDB instance for storing trie structure
@@ -72,8 +76,9 @@ func NewStorageState(db chaindb.Database, t *trie.Trie) (*StorageState, error) {
 	}
 
 	return &StorageState{
-		trie: t,
-		db:   NewStorageDB(db),
+		trie:    t,
+		db:      NewStorageDB(db),
+		changed: make(map[byte]chan<- *KeyValue),
 	}, nil
 }
 
@@ -120,7 +125,18 @@ func (s *StorageState) EnumeratedTrieRoot(values [][]byte) {
 func (s *StorageState) SetStorage(key []byte, value []byte) error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
-	return s.trie.Put(key, value)
+	kv := &KeyValue{
+		Key:   key,
+		Value: value,
+	}
+	//fmt.Printf("SetStorage Key: %x Value: %x\n", key, value)
+	// TODO ed add channel notify here
+	err := s.trie.Put(key, value)
+	if err != nil {
+		return err
+	}
+	s.notifyChanged(kv)
+	return nil
 }
 
 // ClearPrefix not implemented

dot/state/storage_notify.go

@@ -0,0 +1,76 @@
+// Copyright 2020 ChainSafe Systems (ON) Corp.
+// This file is part of gossamer.
+//
+// The gossamer 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 gossamer 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 gossamer library. If not, see <http://www.gnu.org/licenses/>.
+package state
+
+import (
+	"errors"
+)
+
+// KeyValue struct to hold key value pairs
+type KeyValue struct {
+	Key   []byte
+	Value []byte
+}
+
+// RegisterStorageChangeChannel function to register storage change channels
+func (s *StorageState) RegisterStorageChangeChannel(ch chan<- *KeyValue) (byte, error) {
+	s.changedLock.RLock()
+
+	if len(s.changed) == 256 {
+		return 0, errors.New("channel limit reached")
+	}
+
+	var id byte
+	for {
+		id = generateID()
+		if s.changed[id] == nil {
+			break
+		}
+	}
+
+	s.changedLock.RUnlock()
+
+	s.changedLock.Lock()
+	s.changed[id] = ch
+	s.changedLock.Unlock()
+	return id, nil
+}
+
+// UnregisterStorageChangeChannel removes the storage change notification channel with the given ID.
+// A channel must be unregistered before closing it.
+func (s *StorageState) UnregisterStorageChangeChannel(id byte) {
+	s.changedLock.Lock()
+	defer s.changedLock.Unlock()
+
+	delete(s.changed, id)
+}
+
+func (s *StorageState) notifyChanged(change *KeyValue) {
+	s.changedLock.RLock()
+	defer s.changedLock.RUnlock()
+
+	if len(s.changed) == 0 {
+		return
+	}
+
+	logger.Trace("notifying changed storage chans...", "chans", s.changed)
+
+	for _, ch := range s.changed {
+		go func(ch chan<- *KeyValue) {
+			ch <- change
+		}(ch)
+	}
+}