Added names to interface arguments, created Arguments type for websockets

Author: UlyanaAndrukhiv

engine/access/rest/websockets/data_providers/base_provider.go

@@ -16,7 +16,7 @@ type BaseDataProvider interface {
 	// Topic returns the topic associated with the data provider.
 	Topic() string
 	// Close terminates the data provider.
-	Close()
+	Close() error
 }
 
 var _ BaseDataProvider = (*BaseDataProviderImpl)(nil)
@@ -58,6 +58,7 @@ func (b *BaseDataProviderImpl) Topic() string {
 }
 
 // Close terminates the data provider.
-func (b *BaseDataProviderImpl) Close() {
+func (b *BaseDataProviderImpl) Close() error {
 	b.cancel()
+	return nil
 }

engine/access/rest/websockets/data_providers/block_digests_provider.go

@@ -30,7 +30,7 @@ func NewBlockDigestsDataProvider(
 	logger zerolog.Logger,
 	api access.API,
 	topic string,
-	arguments map[string]string,
+	arguments models.Arguments,
 	send chan<- interface{},
 ) (*BlockDigestsDataProvider, error) {
 	p := &BlockDigestsDataProvider{

engine/access/rest/websockets/data_providers/block_headers_provider.go

@@ -30,7 +30,7 @@ func NewBlockHeadersDataProvider(
 	logger zerolog.Logger,
 	api access.API,
 	topic string,
-	arguments map[string]string,
+	arguments models.Arguments,
 	send chan<- interface{},
 ) (*BlockHeadersDataProvider, error) {
 	p := &BlockHeadersDataProvider{

engine/access/rest/websockets/data_providers/blocks_provider.go

@@ -39,7 +39,7 @@ func NewBlocksDataProvider(
 	logger zerolog.Logger,
 	api access.API,
 	topic string,
-	arguments map[string]string,
+	arguments models.Arguments,
 	send chan<- interface{},
 ) (*BlocksDataProvider, error) {
 	p := &BlocksDataProvider{
@@ -100,7 +100,7 @@ func (p *BlocksDataProvider) handleResponse(send chan<- interface{}) func(*flow.
 }
 
 // ParseBlocksArguments validates and initializes the blocks arguments.
-func ParseBlocksArguments(arguments map[string]string) (BlocksArguments, error) {
+func ParseBlocksArguments(arguments models.Arguments) (BlocksArguments, error) {
 	var args BlocksArguments
 
 	// Parse 'block_status'

engine/access/rest/websockets/data_providers/blocks_provider_test.go

@@ -10,6 +10,7 @@ import (
 
 	accessmock "github.com/onflow/flow-go/access/mock"
 	"github.com/onflow/flow-go/engine/access/rest/common/parser"
+	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/utils/unittest"
 )
@@ -18,7 +19,7 @@ const unknownBlockStatus = "unknown_block_status"
 
 type testErrType struct {
 	name             string
-	arguments        map[string]string
+	arguments        models.Arguments
 	expectedErrorMsg string
 }
 
@@ -69,21 +70,21 @@ func (s *BlocksProviderSuite) invalidArgumentsTestCases() []testErrType {
 	return []testErrType{
 		{
 			name: "missing 'block_status' argument",
-			arguments: map[string]string{
+			arguments: models.Arguments{
 				"start_block_id": s.rootBlock.ID().String(),
 			},
 			expectedErrorMsg: "'block_status' must be provided",
 		},
 		{
 			name: "unknown 'block_status' argument",
-			arguments: map[string]string{
+			arguments: models.Arguments{
 				"block_status": unknownBlockStatus,
 			},
 			expectedErrorMsg: fmt.Sprintf("invalid 'block_status', must be '%s' or '%s'", parser.Finalized, parser.Sealed),
 		},
 		{
 			name: "provide both 'start_block_id' and 'start_block_height' arguments",
-			arguments: map[string]string{
+			arguments: models.Arguments{
 				"block_status":       parser.Finalized,
 				"start_block_id":     s.rootBlock.ID().String(),
 				"start_block_height": fmt.Sprintf("%d", s.rootBlock.Header.Height),

engine/access/rest/websockets/data_providers/factory.go

@@ -7,6 +7,7 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/onflow/flow-go/access"
+	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 )
 
@@ -29,7 +30,7 @@ type DataProviderFactory interface {
 	// and configuration parameters.
 	//
 	// No errors are expected during normal operations.
-	NewDataProvider(context.Context, string, map[string]string, chan<- interface{}) (DataProvider, error)
+	NewDataProvider(ctx context.Context, topic string, arguments models.Arguments, ch chan<- interface{}) (DataProvider, error)
 }
 
 var _ DataProviderFactory = (*DataProviderFactoryImpl)(nil)
@@ -76,7 +77,7 @@ func NewDataProviderFactory(
 func (s *DataProviderFactoryImpl) NewDataProvider(
 	ctx context.Context,
 	topic string,
-	arguments map[string]string,
+	arguments models.Arguments,
 	ch chan<- interface{},
 ) (DataProvider, error) {
 	switch topic {

engine/access/rest/websockets/data_providers/factory_test.go

@@ -10,6 +10,7 @@ import (
 
 	accessmock "github.com/onflow/flow-go/access/mock"
 	"github.com/onflow/flow-go/engine/access/rest/common/parser"
+	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
 	statestreammock "github.com/onflow/flow-go/engine/access/state_stream/mock"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/utils/unittest"
@@ -63,14 +64,14 @@ func (s *DataProviderFactorySuite) TestSupportedTopics() {
 	testCases := []struct {
 		name               string
 		topic              string
-		arguments          map[string]string
+		arguments          models.Arguments
 		setupSubscription  func()
 		assertExpectations func()
 	}{
 		{
 			name:      "block topic",
 			topic:     BlocksTopic,
-			arguments: map[string]string{"block_status": parser.Finalized},
+			arguments: models.Arguments{"block_status": parser.Finalized},
 			setupSubscription: func() {
 				s.setupSubscription(s.accessApi.On("SubscribeBlocksFromLatest", mock.Anything, flow.BlockStatusFinalized))
 			},
@@ -81,7 +82,7 @@ func (s *DataProviderFactorySuite) TestSupportedTopics() {
 		{
 			name:      "block headers topic",
 			topic:     BlockHeadersTopic,
-			arguments: map[string]string{"block_status": parser.Finalized},
+			arguments: models.Arguments{"block_status": parser.Finalized},
 			setupSubscription: func() {
 				s.setupSubscription(s.accessApi.On("SubscribeBlockHeadersFromLatest", mock.Anything, flow.BlockStatusFinalized))
 			},
@@ -92,7 +93,7 @@ func (s *DataProviderFactorySuite) TestSupportedTopics() {
 		{
 			name:      "block digests topic",
 			topic:     BlockDigestsTopic,
-			arguments: map[string]string{"block_status": parser.Finalized},
+			arguments: models.Arguments{"block_status": parser.Finalized},
 			setupSubscription: func() {
 				s.setupSubscription(s.accessApi.On("SubscribeBlockDigestsFromLatest", mock.Anything, flow.BlockStatusFinalized))
 			},

engine/access/rest/websockets/data_providers/mock/data_provider.go

@@ -1,10 +1,12 @@
 package models
 
+type Arguments map[string]string
+
 // SubscribeMessageRequest represents a request to subscribe to a topic.
 type SubscribeMessageRequest struct {
 	BaseMessageRequest
-	Topic     string            `json:"topic"`     // Topic to subscribe to
-	Arguments map[string]string `json:"arguments"` // Additional arguments for subscription
+	Topic     string    `json:"topic"`     // Topic to subscribe to
+	Arguments Arguments `json:"arguments"` // Additional arguments for subscription
 }
 
 // SubscribeMessageResponse represents the response to a subscription request.