Removed handleResponse func and using generic HandleResponse instead

onflow · Dec 24, 2024 · f71edd5 · f71edd5
1 parent 71bac49
commit f71edd5
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 48 deletions.
diff --git a/engine/access/rest/websockets/data_providers/send_transaction_statuses_provider.go b/engine/access/rest/websockets/data_providers/send_transaction_statuses_provider.go
@@ -68,39 +68,27 @@ func NewSendTransactionStatusesDataProvider(
 //
 // No errors are expected during normal operations.
 func (p *SendTransactionStatusesDataProvider) Run() error {
-	return subscription.HandleSubscription(p.subscription, p.handleResponse())
-}
-
-// createSubscription creates a new subscription using the specified input arguments.
-func (p *SendTransactionStatusesDataProvider) createSubscription(
-	ctx context.Context,
-	args sendTransactionStatusesArguments,
-) subscription.Subscription {
-	return p.api.SendAndSubscribeTransactionStatuses(ctx, &args.Transaction, entities.EventEncodingVersion_JSON_CDC_V0)
-}
-
-// handleResponse processes an account statuses and sends the formatted response.
-//
-// No errors are expected during normal operations.
-func (p *SendTransactionStatusesDataProvider) handleResponse() func(txResults []*access.TransactionResult) error {
-
 	messageIndex := counters.NewMonotonousCounter(0)
 
-	return func(txResults []*access.TransactionResult) error {
-
+	return subscription.HandleSubscription(p.subscription, subscription.HandleResponse(p.send, func(txResults []*access.TransactionResult) (interface{}, error) {
+		index := messageIndex.Value()
 		if ok := messageIndex.Set(messageIndex.Value() + 1); !ok {
-			return status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
+			return nil, status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
 		}
-		index := messageIndex.Value()
 
-		p.send <- &models.TransactionStatusesResponse{
+		return &models.TransactionStatusesResponse{
 			TransactionResults: txResults,
 			MessageIndex:       index,
-		}
-
-		return nil
-	}
+		}, nil
+	}))
+}
 
+// createSubscription creates a new subscription using the specified input arguments.
+func (p *SendTransactionStatusesDataProvider) createSubscription(
+	ctx context.Context,
+	args sendTransactionStatusesArguments,
+) subscription.Subscription {
+	return p.api.SendAndSubscribeTransactionStatuses(ctx, &args.Transaction, entities.EventEncodingVersion_JSON_CDC_V0)
 }
 
 // parseAccountStatusesArguments validates and initializes the account statuses arguments.

diff --git a/engine/access/rest/websockets/data_providers/transaction_statuses_provider.go b/engine/access/rest/websockets/data_providers/transaction_statuses_provider.go
@@ -71,7 +71,19 @@ func NewTransactionStatusesDataProvider(
 //
 // No errors are expected during normal operations.
 func (p *TransactionStatusesDataProvider) Run() error {
-	return subscription.HandleSubscription(p.subscription, p.handleResponse())
+	messageIndex := counters.NewMonotonousCounter(0)
+
+	return subscription.HandleSubscription(p.subscription, subscription.HandleResponse(p.send, func(txResults []*access.TransactionResult) (interface{}, error) {
+		index := messageIndex.Value()
+		if ok := messageIndex.Set(messageIndex.Value() + 1); !ok {
+			return nil, status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
+		}
+
+		return &models.TransactionStatusesResponse{
+			TransactionResults: txResults,
+			MessageIndex:       index,
+		}, nil
+	}))
 }
 
 // createSubscription creates a new subscription using the specified input arguments.
@@ -90,28 +102,6 @@ func (p *TransactionStatusesDataProvider) createSubscription(
 	return p.api.SubscribeTransactionStatusesFromLatest(ctx, args.TxID, entities.EventEncodingVersion_JSON_CDC_V0)
 }
 
-// handleResponse processes an account statuses and sends the formatted response.
-//
-// No errors are expected during normal operations.
-func (p *TransactionStatusesDataProvider) handleResponse() func(txResults []*access.TransactionResult) error {
-	messageIndex := counters.NewMonotonousCounter(0)
-
-	return func(txResults []*access.TransactionResult) error {
-
-		index := messageIndex.Value()
-		if ok := messageIndex.Set(messageIndex.Value() + 1); !ok {
-			return status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
-		}
-
-		p.send <- &models.TransactionStatusesResponse{
-			TransactionResults: txResults,
-			MessageIndex:       index,
-		}
-
-		return nil
-	}
-}
-
 // parseAccountStatusesArguments validates and initializes the account statuses arguments.
 func parseTransactionStatusesArguments(
 	arguments models.Arguments,