Added generic HandleResponse, updated providers

Author: UlyanaAndrukhiv

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

@@ -58,7 +58,14 @@ func NewBlockDigestsDataProvider(
 //
 // No errors are expected during normal operations.
 func (p *BlockDigestsDataProvider) Run() error {
-	return subscription.HandleSubscription(p.subscription, p.handleResponse(p.send))
+	return subscription.HandleSubscription(
+		p.subscription,
+		subscription.HandleResponse(p.send, func(block *flow.BlockDigest) (interface{}, error) {
+			return &models.BlockDigestMessageResponse{
+				Block: block,
+			}, nil
+		}),
+	)
 }
 
 // createSubscription creates a new subscription using the specified input arguments.
@@ -73,16 +80,3 @@ func (p *BlockDigestsDataProvider) createSubscription(ctx context.Context, args
 
 	return p.api.SubscribeBlockDigestsFromLatest(ctx, args.BlockStatus)
 }
-
-// handleResponse processes a block digest and sends the formatted response.
-//
-// No errors are expected during normal operations.
-func (p *BlockDigestsDataProvider) handleResponse(send chan<- interface{}) func(block *flow.BlockDigest) error {
-	return func(block *flow.BlockDigest) error {
-		send <- &models.BlockDigestMessageResponse{
-			Block: block,
-		}
-
-		return nil
-	}
-}

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

@@ -58,7 +58,14 @@ func NewBlockHeadersDataProvider(
 //
 // No errors are expected during normal operations.
 func (p *BlockHeadersDataProvider) Run() error {
-	return subscription.HandleSubscription(p.subscription, p.handleResponse(p.send))
+	return subscription.HandleSubscription(
+		p.subscription,
+		subscription.HandleResponse(p.send, func(header *flow.Header) (interface{}, error) {
+			return &models.BlockHeaderMessageResponse{
+				Header: header,
+			}, nil
+		}),
+	)
 }
 
 // createSubscription creates a new subscription using the specified input arguments.
@@ -73,16 +80,3 @@ func (p *BlockHeadersDataProvider) createSubscription(ctx context.Context, args
 
 	return p.api.SubscribeBlockHeadersFromLatest(ctx, args.BlockStatus)
 }
-
-// handleResponse processes a block header and sends the formatted response.
-//
-// No errors are expected during normal operations.
-func (p *BlockHeadersDataProvider) handleResponse(send chan<- interface{}) func(header *flow.Header) error {
-	return func(header *flow.Header) error {
-		send <- &models.BlockHeaderMessageResponse{
-			Header: header,
-		}
-
-		return nil
-	}
-}

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

@@ -67,7 +67,14 @@ func NewBlocksDataProvider(
 //
 // No errors are expected during normal operations.
 func (p *BlocksDataProvider) Run() error {
-	return subscription.HandleSubscription(p.subscription, p.handleResponse(p.send))
+	return subscription.HandleSubscription(
+		p.subscription,
+		subscription.HandleResponse(p.send, func(block *flow.Block) (interface{}, error) {
+			return &models.BlockMessageResponse{
+				Block: block,
+			}, nil
+		}),
+	)
 }
 
 // createSubscription creates a new subscription using the specified input arguments.
@@ -83,19 +90,6 @@ func (p *BlocksDataProvider) createSubscription(ctx context.Context, args Blocks
 	return p.api.SubscribeBlocksFromLatest(ctx, args.BlockStatus)
 }
 
-// handleResponse processes a block and sends the formatted response.
-//
-// No errors are expected during normal operations.
-func (p *BlocksDataProvider) handleResponse(send chan<- interface{}) func(*flow.Block) error {
-	return func(block *flow.Block) error {
-		send <- &models.BlockMessageResponse{
-			Block: block,
-		}
-
-		return nil
-	}
-}
-
 // ParseBlocksArguments validates and initializes the blocks arguments.
 func ParseBlocksArguments(arguments models.Arguments) (BlocksArguments, error) {
 	var args BlocksArguments

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

@@ -217,7 +217,7 @@ func (s *BlocksProviderSuite) requireBlock(v interface{}, expectedBlock *flow.Bl
 // Arguments:
 // - topic: The topic associated with the data provider.
 // - tests: A slice of test cases to run, each specifying setup and validation logic.
-// - sendData: A function to simulate emitting data into the data channel.
+// - sendData: A function to simulate emitting data into the subscription's data channel.
 // - requireFn: A function to validate the output received in the send channel.
 func (s *BlocksProviderSuite) testHappyPath(
 	topic string,

engine/access/subscription/util.go

@@ -54,3 +54,25 @@ func HandleRPCSubscription[T any](sub Subscription, handleResponse func(resp T)
 
 	return nil
 }
+
+// HandleResponse processes a generic response of type and sends it to the provided channel.
+//
+// Parameters:
+// - send: The channel to which the processed response is sent.
+// - transform: A function to transform the response into the expected interface{} type.
+//
+// No errors are expected during normal operations.
+func HandleResponse[T any](send chan<- interface{}, transform func(resp T) (interface{}, error)) func(resp T) error {
+	return func(response T) error {
+		// Transform the response
+		resp, err := transform(response)
+		if err != nil {
+			return fmt.Errorf("failed to transform response: %w", err)
+		}
+
+		// send to the channel
+		send <- resp
+
+		return nil
+	}
+}