Remove WebSocket subscription of which closed connection (#487)

* Remove websocket filter if websocket has been closed without uninstall

* Corrects test comment

* Fix log level

* Rename filtersLock to lock

* simplify loop for flushing in FilterManager

* Use defer for unlock in FilterManager

* Rename LogFilter to LogQuery

* Refactor FilterManager

* Fix lint error

* Add comments to FilterManager

* Rename filterOptions to query

* Add new error in FilterManagger:GetChanges for WebSocket filter

* Update BSD License list

jsonrpc/dispatcher.go

@@ -137,11 +137,11 @@ func (d *Dispatcher) handleSubscribe(req Request, conn wsConn) (string, Error) {
 	if subscribeMethod == "newHeads" {
 		filterID = d.filterManager.NewBlockFilter(conn)
 	} else if subscribeMethod == "logs" {
-		logFilter, err := decodeLogFilterFromInterface(params[1])
+		logQuery, err := decodeLogQueryFromInterface(params[1])
 		if err != nil {
 			return "", NewInternalError(err.Error())
 		}
-		filterID = d.filterManager.NewLogFilter(logFilter, conn)
+		filterID = d.filterManager.NewLogFilter(logQuery, conn)
 	} else {
 		return "", NewSubscriptionNotFoundError(subscribeMethod)
 	}
@@ -178,8 +178,7 @@ func (d *Dispatcher) HandleWs(reqBody []byte, conn wsConn) ([]byte, error) {
 	if req.Method == "eth_subscribe" {
 		filterID, err := d.handleSubscribe(req, conn)
 		if err != nil {
-			//nolint
-			NewRPCResponse(req.ID, "2.0", nil, err).Bytes()
+			return NewRPCResponse(req.ID, "2.0", nil, err).Bytes()
 		}
 
 		resp, err := formatFilterResponse(req.ID, filterID)
@@ -209,6 +208,7 @@ func (d *Dispatcher) HandleWs(reqBody []byte, conn wsConn) ([]byte, error) {
 
 		return []byte(resp), nil
 	}
+
 	// its a normal query that we handle with the dispatcher
 	resp, err := d.handleReq(req)
 	if err != nil {

jsonrpc/dispatcher_test.go

@@ -187,7 +187,7 @@ func (m *mockService) BlockPtr(a string, f *BlockNumber) (interface{}, error) {
 	return nil, nil
 }
 
-func (m *mockService) Filter(f LogFilter) (interface{}, error) {
+func (m *mockService) Filter(f LogQuery) (interface{}, error) {
 	m.msgCh <- f
 
 	return nil, nil
@@ -249,9 +249,10 @@ func TestDispatcherFuncDecode(t *testing.T) {
 		{
 			"filter",
 			`[{"fromBlock": "pending", "toBlock": "earliest"}]`,
-			LogFilter{fromBlock: PendingBlockNumber, toBlock: EarliestBlockNumber},
+			LogQuery{fromBlock: PendingBlockNumber, toBlock: EarliestBlockNumber},
 		},
 	}
+
 	for _, c := range cases {
 		res := handleReq(c.typ, c.msg)
 		if !reflect.DeepEqual(res, c.res) {

jsonrpc/eth_blockchain_test.go

@@ -111,36 +111,36 @@ func TestEth_Block_GetLogs(t *testing.T) {
 
 	testTable := []struct {
 		name           string
-		filterOptions  *LogFilter
+		query          *LogQuery
 		shouldFail     bool
 		expectedLength int
 	}{
 		{"Found matching logs, fromBlock < toBlock",
-			&LogFilter{
+			&LogQuery{
 				fromBlock: 1,
 				toBlock:   3,
 				Topics:    topics,
 			},
 			false, 3},
 		{"Found matching logs, fromBlock == toBlock",
-			&LogFilter{
+			&LogQuery{
 				fromBlock: 2,
 				toBlock:   2,
 				Topics:    topics,
 			},
 			false, 1},
 		{"Found matching logs, BlockHash present",
-			&LogFilter{
+			&LogQuery{
 				BlockHash: &blockHash,
 				Topics:    topics,
 			},
 			false, 1},
-		{"No logs found", &LogFilter{
+		{"No logs found", &LogQuery{
 			fromBlock: 4,
 			toBlock:   5,
 			Topics:    topics,
 		}, false, 0},
-		{"Invalid block range", &LogFilter{
+		{"Invalid block range", &LogQuery{
 			fromBlock: 10,
 			toBlock:   5,
 			Topics:    topics,
@@ -176,7 +176,7 @@ func TestEth_Block_GetLogs(t *testing.T) {
 
 	for _, testCase := range testTable {
 		t.Run(testCase.name, func(t *testing.T) {
-			foundLogs, logError := eth.GetLogs(testCase.filterOptions)
+			foundLogs, logError := eth.GetLogs(testCase.query)
 
 			if logError != nil && !testCase.shouldFail {
 				// If there is an error and test isn't expected to fail

jsonrpc/eth_endpoint.go

@@ -665,7 +665,7 @@ func (e *Eth) EstimateGas(arg *txnArgs, rawNum *BlockNumber) (interface{}, error
 }
 
 // GetLogs returns an array of logs matching the filter options
-func (e *Eth) GetLogs(filterOptions *LogFilter) (interface{}, error) {
+func (e *Eth) GetLogs(query *LogQuery) (interface{}, error) {
 	result := make([]*Log, 0)
 	parseReceipts := func(block *types.Block) error {
 		receipts, err := e.store.GetReceiptsByHash(block.Header.Hash)
@@ -675,7 +675,7 @@ func (e *Eth) GetLogs(filterOptions *LogFilter) (interface{}, error) {
 
 		for indx, receipt := range receipts {
 			for logIndx, log := range receipt.Logs {
-				if filterOptions.Match(log) {
+				if query.Match(log) {
 					result = append(result, &Log{
 						Address:     log.Address,
 						Topics:      log.Topics,
@@ -693,8 +693,8 @@ func (e *Eth) GetLogs(filterOptions *LogFilter) (interface{}, error) {
 		return nil
 	}
 
-	if filterOptions.BlockHash != nil {
-		block, ok := e.store.GetBlockByHash(*filterOptions.BlockHash, true)
+	if query.BlockHash != nil {
+		block, ok := e.store.GetBlockByHash(*query.BlockHash, true)
 		if !ok {
 			return nil, fmt.Errorf("not found")
 		}
@@ -729,8 +729,8 @@ func (e *Eth) GetLogs(filterOptions *LogFilter) (interface{}, error) {
 		return uint64(num)
 	}
 
-	from := resolveNum(filterOptions.fromBlock)
-	to := resolveNum(filterOptions.toBlock)
+	from := resolveNum(query.fromBlock)
+	to := resolveNum(query.toBlock)
 
 	if to < from {
 		return nil, fmt.Errorf("incorrect range")
@@ -858,7 +858,7 @@ func (e *Eth) GetCode(address types.Address, filter BlockNumberOrHash) (interfac
 }
 
 // NewFilter creates a filter object, based on filter options, to notify when the state changes (logs).
-func (e *Eth) NewFilter(filter *LogFilter) (interface{}, error) {
+func (e *Eth) NewFilter(filter *LogQuery) (interface{}, error) {
 	return e.filterManager.NewLogFilter(filter, nil), nil
 }