Removed storing arguments in the data providers

Author: UlyanaAndrukhiv

engine/access/rest/websockets/controller.go

@@ -164,6 +164,7 @@ func (c *Controller) handleSubscribe(ctx context.Context, msg models.SubscribeMe
 	dp, err := c.dataProviderFactory.NewDataProvider(ctx, msg.Topic, msg.Arguments, c.communicationChannel)
 	if err != nil {
 		// TODO: handle error here
+		c.logger.Error().Err(err).Msgf("error while creating data provider for topic: %s", msg.Topic)
 	}
 
 	c.dataProviders.Add(dp.ID(), dp)

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

@@ -18,7 +18,6 @@ type BlockDigestsDataProvider struct {
 	*BaseDataProviderImpl
 
 	logger zerolog.Logger
-	args   BlocksArguments
 	api    access.API
 }
 
@@ -38,20 +37,18 @@ func NewBlockDigestsDataProvider(
 		api:    api,
 	}
 
-	// Initialize arguments passed to the provider.
-	var err error
-	p.args, err = ParseBlocksArguments(arguments)
+	// Parse arguments passed to the provider.
+	blockArgs, err := ParseBlocksArguments(arguments)
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments: %w", err)
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
-
+	subCtx, cancel := context.WithCancel(ctx)
 	p.BaseDataProviderImpl = NewBaseDataProviderImpl(
 		topic,
 		cancel,
 		send,
-		p.createSubscription(ctx), // Set up a subscription to block digests based on arguments.
+		p.createSubscription(subCtx, blockArgs), // Set up a subscription to block digests based on arguments.
 	)
 
 	return p, nil
@@ -65,16 +62,16 @@ func (p *BlockDigestsDataProvider) Run() error {
 }
 
 // createSubscription creates a new subscription using the specified input arguments.
-func (p *BlockDigestsDataProvider) createSubscription(ctx context.Context) subscription.Subscription {
-	if p.args.StartBlockID != flow.ZeroID {
-		return p.api.SubscribeBlockDigestsFromStartBlockID(ctx, p.args.StartBlockID, p.args.BlockStatus)
+func (p *BlockDigestsDataProvider) createSubscription(ctx context.Context, args BlocksArguments) subscription.Subscription {
+	if args.StartBlockID != flow.ZeroID {
+		return p.api.SubscribeBlockDigestsFromStartBlockID(ctx, args.StartBlockID, args.BlockStatus)
 	}
 
-	if p.args.StartBlockHeight != request.EmptyHeight {
-		return p.api.SubscribeBlockDigestsFromStartHeight(ctx, p.args.StartBlockHeight, p.args.BlockStatus)
+	if args.StartBlockHeight != request.EmptyHeight {
+		return p.api.SubscribeBlockDigestsFromStartHeight(ctx, args.StartBlockHeight, args.BlockStatus)
 	}
 
-	return p.api.SubscribeBlockDigestsFromLatest(ctx, p.args.BlockStatus)
+	return p.api.SubscribeBlockDigestsFromLatest(ctx, args.BlockStatus)
 }
 
 // handleResponse processes a block digest and sends the formatted response.

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

@@ -18,7 +18,6 @@ type BlockHeadersDataProvider struct {
 	*BaseDataProviderImpl
 
 	logger zerolog.Logger
-	args   BlocksArguments
 	api    access.API
 }
 
@@ -38,20 +37,18 @@ func NewBlockHeadersDataProvider(
 		api:    api,
 	}
 
-	// Initialize arguments passed to the provider.
-	var err error
-	p.args, err = ParseBlocksArguments(arguments)
+	// Parse arguments passed to the provider.
+	blockArgs, err := ParseBlocksArguments(arguments)
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments: %w", err)
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
-
+	subCtx, cancel := context.WithCancel(ctx)
 	p.BaseDataProviderImpl = NewBaseDataProviderImpl(
 		topic,
 		cancel,
 		send,
-		p.createSubscription(ctx), // Set up a subscription to block headers based on arguments.
+		p.createSubscription(subCtx, blockArgs), // Set up a subscription to block headers based on arguments.
 	)
 
 	return p, nil
@@ -65,16 +62,16 @@ func (p *BlockHeadersDataProvider) Run() error {
 }
 
 // createSubscription creates a new subscription using the specified input arguments.
-func (p *BlockHeadersDataProvider) createSubscription(ctx context.Context) subscription.Subscription {
-	if p.args.StartBlockID != flow.ZeroID {
-		return p.api.SubscribeBlockHeadersFromStartBlockID(ctx, p.args.StartBlockID, p.args.BlockStatus)
+func (p *BlockHeadersDataProvider) createSubscription(ctx context.Context, args BlocksArguments) subscription.Subscription {
+	if args.StartBlockID != flow.ZeroID {
+		return p.api.SubscribeBlockHeadersFromStartBlockID(ctx, args.StartBlockID, args.BlockStatus)
 	}
 
-	if p.args.StartBlockHeight != request.EmptyHeight {
-		return p.api.SubscribeBlockHeadersFromStartHeight(ctx, p.args.StartBlockHeight, p.args.BlockStatus)
+	if args.StartBlockHeight != request.EmptyHeight {
+		return p.api.SubscribeBlockHeadersFromStartHeight(ctx, args.StartBlockHeight, args.BlockStatus)
 	}
 
-	return p.api.SubscribeBlockHeadersFromLatest(ctx, p.args.BlockStatus)
+	return p.api.SubscribeBlockHeadersFromLatest(ctx, args.BlockStatus)
 }
 
 // handleResponse processes a block header and sends the formatted response.

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

@@ -27,7 +27,6 @@ type BlocksDataProvider struct {
 	*BaseDataProviderImpl
 
 	logger zerolog.Logger
-	args   BlocksArguments
 	api    access.API
 }
 
@@ -47,20 +46,18 @@ func NewBlocksDataProvider(
 		api:    api,
 	}
 
-	// Initialize arguments passed to the provider.
-	var err error
-	p.args, err = ParseBlocksArguments(arguments)
+	// Parse arguments passed to the provider.
+	blockArgs, err := ParseBlocksArguments(arguments)
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments: %w", err)
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
-
+	subCtx, cancel := context.WithCancel(ctx)
 	p.BaseDataProviderImpl = NewBaseDataProviderImpl(
 		topic,
 		cancel,
 		send,
-		p.createSubscription(ctx), // Set up a subscription to blocks based on arguments.
+		p.createSubscription(subCtx, blockArgs), // Set up a subscription to blocks based on arguments.
 	)
 
 	return p, nil
@@ -74,16 +71,16 @@ func (p *BlocksDataProvider) Run() error {
 }
 
 // createSubscription creates a new subscription using the specified input arguments.
-func (p *BlocksDataProvider) createSubscription(ctx context.Context) subscription.Subscription {
-	if p.args.StartBlockID != flow.ZeroID {
-		return p.api.SubscribeBlocksFromStartBlockID(ctx, p.args.StartBlockID, p.args.BlockStatus)
+func (p *BlocksDataProvider) createSubscription(ctx context.Context, args BlocksArguments) subscription.Subscription {
+	if args.StartBlockID != flow.ZeroID {
+		return p.api.SubscribeBlocksFromStartBlockID(ctx, args.StartBlockID, args.BlockStatus)
 	}
 
-	if p.args.StartBlockHeight != request.EmptyHeight {
-		return p.api.SubscribeBlocksFromStartHeight(ctx, p.args.StartBlockHeight, p.args.BlockStatus)
+	if args.StartBlockHeight != request.EmptyHeight {
+		return p.api.SubscribeBlocksFromStartHeight(ctx, args.StartBlockHeight, args.BlockStatus)
 	}
 
-	return p.api.SubscribeBlocksFromLatest(ctx, p.args.BlockStatus)
+	return p.api.SubscribeBlocksFromLatest(ctx, args.BlockStatus)
 }
 
 // handleResponse processes a block and sends the formatted response.