fix (dot/rpc, dot/state): state_subscribeStorage to only notify for value changes

dot/rpc/http.go

@@ -177,9 +177,8 @@ func (h *HTTPServer) Stop() error {
 		for _, conn := range h.wsConns {
 			for _, sub := range conn.Subscriptions {
 				switch v := sub.(type) {
-				case *subscription.StorageChangeListener:
-					h.serverConfig.StorageAPI.UnregisterStorageChangeChannel(v.ChanID)
-					close(v.Channel)
+				case *subscription.StorageObserver:
+					h.serverConfig.StorageAPI.UnregisterStorageObserver(v)
 				case *subscription.BlockListener:
 					h.serverConfig.BlockAPI.UnregisterImportedChannel(v.ChanID)
 					close(v.Channel)

dot/rpc/modules/api.go

@@ -16,10 +16,10 @@ type StorageAPI interface {
 	GetStorage(root *common.Hash, key []byte) ([]byte, error)
 	GetStorageByBlockHash(bhash common.Hash, key []byte) ([]byte, error)
 	Entries(root *common.Hash) (map[string][]byte, error)
-	RegisterStorageChangeChannel(sub state.StorageSubscription) (byte, error)
-	UnregisterStorageChangeChannel(id byte)
 	GetStateRootFromBlock(bhash *common.Hash) (*common.Hash, error)
 	GetKeysWithPrefix(root *common.Hash, prefix []byte) ([][]byte, error)
+	RegisterStorageObserver(observer state.Observer)
+	UnregisterStorageObserver(observer state.Observer)
 }
 
 // BlockAPI is the interface for the block state

dot/rpc/subscription/listeners.go

@@ -30,121 +30,61 @@ type Listener interface {
 	Listen()
 }
 
-func (c *WSConn) startListener(lid int) {
-	go c.Subscriptions[lid].Listen()
+// WSConnAPI interface defining methors a WSConn should have
+type WSConnAPI interface {
+	safeSend(interface{})
 }
 
-func (c *WSConn) initStorageChangeListener(reqID float64, params interface{}) (int, error) {
-	scl := &StorageChangeListener{
-		Channel: make(chan *state.SubscriptionResult),
-		wsconn:  c,
-	}
-	sub := &state.StorageSubscription{
-		Filter:   make(map[string]bool),
-		Listener: scl.Channel,
-	}
-
-	pA := params.([]interface{})
-	for _, param := range pA {
-		switch p := param.(type) {
-		case []interface{}:
-			for _, pp := range param.([]interface{}) {
-				sub.Filter[pp.(string)] = true
-			}
-		case string:
-			sub.Filter[p] = true
-		default:
-			return 0, fmt.Errorf("unknow parameter type")
-		}
-	}
+// StorageObserver struct to hold data for observer (Observer Design Pattern)
+type StorageObserver struct {
+	id     int
+	filter map[string][]byte
+	wsconn WSConnAPI
+}
 
-	if c.StorageAPI == nil {
-		c.safeSendError(reqID, nil, "error StorageAPI not set")
-		return 0, fmt.Errorf("error StorageAPI not set")
+// Update is called to notify observer of new value
+func (s *StorageObserver) Update(change *state.SubscriptionResult) {
+	if change == nil {
+		return
 	}
 
-	chanID, err := c.StorageAPI.RegisterStorageChangeChannel(*sub)
-	if err != nil {
-		return 0, err
+	result := make(map[string]interface{})
+	result["block"] = change.Hash.String()
+	changes := make([][]string, 0, len(change.Changes))
+	for _, v := range change.Changes {
+		kv := []string{common.BytesToHex(v.Key), common.BytesToHex(v.Value)}
+		changes = append(changes, kv)
 	}
-	scl.ChanID = chanID
+	result["changes"] = changes
 
-	c.qtyListeners++
-	scl.subID = c.qtyListeners
-	c.Subscriptions[scl.subID] = scl
-	c.StorageSubChannels[scl.subID] = chanID
-
-	initRes := newSubscriptionResponseJSON(scl.subID, reqID)
-	c.safeSend(initRes)
-
-	return scl.subID, nil
+	res := newSubcriptionBaseResponseJSON()
+	res.Method = "state_storage"
+	res.Params.Result = result
+	res.Params.SubscriptionID = s.GetID()
+	s.wsconn.safeSend(res)
 }
 
-// StorageChangeListener for listening to state change channels
-type StorageChangeListener struct {
-	Channel chan *state.SubscriptionResult
-	wsconn  *WSConn
-	ChanID  byte
-	subID   int
+// GetID the id for the Observer
+func (s *StorageObserver) GetID() int {
+	return s.id
 }
 
-// Listen implementation of Listen interface to listen for importedChan changes
-func (l *StorageChangeListener) Listen() {
-	for change := range l.Channel {
-		if change == nil {
-			continue
-		}
-
-		result := make(map[string]interface{})
-		result["block"] = change.Hash.String()
-		changes := [][]string{}
-		for _, v := range change.Changes {
-			kv := []string{common.BytesToHex(v.Key), common.BytesToHex(v.Value)}
-			changes = append(changes, kv)
-		}
-		result["changes"] = changes
-
-		res := newSubcriptionBaseResponseJSON()
-		res.Method = "state_storage"
-		res.Params.Result = result
-		res.Params.SubscriptionID = l.subID
-		l.wsconn.safeSend(res)
-	}
+// GetFilter returns the filter the Observer is using
+func (s *StorageObserver) GetFilter() map[string][]byte {
+	return s.filter
 }
 
+// Listen to satisfy Listener interface (but is no longer used by StorageObserver)
+func (s *StorageObserver) Listen() {}
+
 // BlockListener to handle listening for blocks importedChan
 type BlockListener struct {
 	Channel chan *types.Block
-	wsconn  *WSConn
+	wsconn  WSConnAPI
 	ChanID  byte
 	subID   int
 }
 
-func (c *WSConn) initBlockListener(reqID float64) (int, error) {
-	bl := &BlockListener{
-		Channel: make(chan *types.Block),
-		wsconn:  c,
-	}
-
-	if c.BlockAPI == nil {
-		c.safeSendError(reqID, nil, "error BlockAPI not set")
-		return 0, fmt.Errorf("error BlockAPI not set")
-	}
-	chanID, err := c.BlockAPI.RegisterImportedChannel(bl.Channel)
-	if err != nil {
-		return 0, err
-	}
-	bl.ChanID = chanID
-	c.qtyListeners++
-	bl.subID = c.qtyListeners
-	c.Subscriptions[bl.subID] = bl
-	c.BlockSubChannels[bl.subID] = chanID
-	initRes := newSubscriptionResponseJSON(bl.subID, reqID)
-	c.safeSend(initRes)
-
-	return bl.subID, nil
-}
-
 // Listen implementation of Listen interface to listen for importedChan changes
 func (l *BlockListener) Listen() {
 	for block := range l.Channel {
@@ -167,36 +107,11 @@ func (l *BlockListener) Listen() {
 // BlockFinalizedListener to handle listening for finalized blocks
 type BlockFinalizedListener struct {
 	channel chan *types.Header
-	wsconn  *WSConn
+	wsconn  WSConnAPI
 	chanID  byte
 	subID   int
 }
 
-func (c *WSConn) initBlockFinalizedListener(reqID float64) (int, error) {
-	bfl := &BlockFinalizedListener{
-		channel: make(chan *types.Header),
-		wsconn:  c,
-	}
-
-	if c.BlockAPI == nil {
-		c.safeSendError(reqID, nil, "error BlockAPI not set")
-		return 0, fmt.Errorf("error BlockAPI not set")
-	}
-	chanID, err := c.BlockAPI.RegisterFinalizedChannel(bfl.channel)
-	if err != nil {
-		return 0, err
-	}
-	bfl.chanID = chanID
-	c.qtyListeners++
-	bfl.subID = c.qtyListeners
-	c.Subscriptions[bfl.subID] = bfl
-	c.BlockSubChannels[bfl.subID] = chanID
-	initRes := newSubscriptionResponseJSON(bfl.subID, reqID)
-	c.safeSend(initRes)
-
-	return bfl.subID, nil
-}
-
 // Listen implementation of Listen interface to listen for importedChan changes
 func (l *BlockFinalizedListener) Listen() {
 	for header := range l.channel {
@@ -217,7 +132,7 @@ func (l *BlockFinalizedListener) Listen() {
 
 // ExtrinsicSubmitListener to handle listening for extrinsic events
 type ExtrinsicSubmitListener struct {
-	wsconn    *WSConn
+	wsconn    WSConnAPI
 	subID     int
 	extrinsic types.Extrinsic
 
@@ -231,55 +146,6 @@ type ExtrinsicSubmitListener struct {
 // AuthorExtrinsicUpdates method name
 const AuthorExtrinsicUpdates = "author_extrinsicUpdate"
 
-func (c *WSConn) initExtrinsicWatch(reqID float64, params interface{}) (int, error) {
-	pA := params.([]interface{})
-	extBytes, err := common.HexToBytes(pA[0].(string))
-	if err != nil {
-		return 0, err
-	}
-
-	// listen for built blocks
-	esl := &ExtrinsicSubmitListener{
-		importedChan:  make(chan *types.Block),
-		wsconn:        c,
-		extrinsic:     types.Extrinsic(extBytes),
-		finalizedChan: make(chan *types.Header),
-	}
-
-	if c.BlockAPI == nil {
-		return 0, fmt.Errorf("error BlockAPI not set")
-	}
-	esl.importedChanID, err = c.BlockAPI.RegisterImportedChannel(esl.importedChan)
-	if err != nil {
-		return 0, err
-	}
-
-	esl.finalizedChanID, err = c.BlockAPI.RegisterFinalizedChannel(esl.finalizedChan)
-	if err != nil {
-		return 0, err
-	}
-
-	c.qtyListeners++
-	esl.subID = c.qtyListeners
-	c.Subscriptions[esl.subID] = esl
-	c.BlockSubChannels[esl.subID] = esl.importedChanID
-
-	err = c.CoreAPI.HandleSubmittedExtrinsic(extBytes)
-	if err != nil {
-		return 0, err
-	}
-	c.safeSend(newSubscriptionResponseJSON(esl.subID, reqID))
-
-	// TODO (ed) since HandleSubmittedExtrinsic has been called we assume the extrinsic is in the tx queue
-	//  should we add a channel to tx queue so we're notified when it's in the queue
-	if c.CoreAPI.IsBlockProducer() {
-		c.safeSend(newSubscriptionResponse(AuthorExtrinsicUpdates, esl.subID, "ready"))
-	}
-
-	// todo (ed) determine which peer extrinsic has been broadcast to, and set status
-	return esl.subID, err
-}
-
 // Listen implementation of Listen interface to listen for importedChan changes
 func (l *ExtrinsicSubmitListener) Listen() {
 	// listen for imported blocks with extrinsic
@@ -322,25 +188,10 @@ type RuntimeVersionListener struct {
 	subID  int
 }
 
-func (c *WSConn) initRuntimeVersionListener(reqID float64) (int, error) {
-	rvl := &RuntimeVersionListener{
-		wsconn: c,
-	}
-	if c.CoreAPI == nil {
-		c.safeSendError(reqID, nil, "error CoreAPI not set")
-		return 0, fmt.Errorf("error CoreAPI not set")
-	}
-	c.qtyListeners++
-	rvl.subID = c.qtyListeners
-	c.Subscriptions[rvl.subID] = rvl
-	initRes := newSubscriptionResponseJSON(rvl.subID, reqID)
-	c.safeSend(initRes)
-
-	return rvl.subID, nil
-}
-
 // Listen implementation of Listen interface to listen for runtime version changes
 func (l *RuntimeVersionListener) Listen() {
+	// This sends current runtime version once when subscription is created
+	// TODO (ed) add logic to send updates when runtime version changes
 	rtVersion, err := l.wsconn.CoreAPI.GetRuntimeVersion(nil)
 	if err != nil {
 		return