Created ScanBlock method to allow for custom and more efficient uses of the delivery service

Signed-off-by: Alexandros Filios <alexandros.filios@ibm.com>

Author: alexandrosfilios

integration/fabric/atsa/chaincode/views/seller.go

@@ -89,7 +89,7 @@ func (a *TransferView) Call(ctx view.Context) (interface{}, error) {
 
 	c, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	assert.Error(ch.Delivery().Scan(
+	assert.Error(ch.Delivery().ScanTx(
 		c,
 		envelope.TxID(),
 		func(tx *fabric.ProcessedTransaction) (bool, error) {
@@ -101,7 +101,7 @@ func (a *TransferView) Call(ctx view.Context) (interface{}, error) {
 
 	c, cancel = context.WithTimeout(context.Background(), 1*time.Minute)
 	defer cancel()
-	assert.NoError(ch.Delivery().Scan(
+	assert.NoError(ch.Delivery().ScanTx(
 		c,
 		envelope.TxID(),
 		func(tx *fabric.ProcessedTransaction) (bool, error) {

platform/fabric/core/generic/delivery/delivery.go

@@ -47,12 +47,6 @@ const (
 	other            messageType       = "other"
 )
 
-// Callback is the callback function prototype to alert the rest of the stack about the availability of a new block.
-// The function returns two argument a boolean to signal if delivery should be stopped, and an error
-// to signal an issue during the processing of the block.
-// In case of an error, the same block is re-processed after a delay.
-type Callback func(context.Context, *common.Block) (bool, error)
-
 // Vault models a key-value store that can be updated by committing rwsets
 type Vault interface {
 	// GetLastTxID returns the last transaction id committed
@@ -73,7 +67,7 @@ type Delivery struct {
 	Services            Services
 	Ledger              driver.Ledger
 	waitForEventTimeout time.Duration
-	callback            Callback
+	callback            driver.BlockCallback
 	vault               Vault
 	client              services.PeerClient
 	tracer              trace.Tracer
@@ -89,7 +83,7 @@ func New(
 	ConfigService driver.ConfigService,
 	PeerManager Services,
 	Ledger driver.Ledger,
-	callback Callback,
+	callback driver.BlockCallback,
 	vault Vault,
 	waitForEventTimeout time.Duration,
 	tracerProvider trace.TracerProvider,

platform/fabric/core/generic/delivery/service.go

@@ -50,7 +50,7 @@ func NewService(
 	waitForEventTimeout time.Duration,
 	txIDStore driver.TXIDStore,
 	transactionManager driver.TransactionManager,
-	callback Callback,
+	callback driver.BlockCallback,
 	tracerProvider trace.TracerProvider,
 	metricsProvider metrics.Provider,
 ) (*Service, error) {
@@ -96,8 +96,7 @@ func (c *Service) Stop() {
 	c.deliveryService.Stop()
 }
 
-func (c *Service) Scan(ctx context.Context, txID string, callback driver.DeliveryCallback) error {
-	vault := &fakeVault{txID: txID}
+func (c *Service) scanBlock(ctx context.Context, vault Vault, callback driver.BlockCallback) error {
 	deliveryService, err := New(
 		c.NetworkName,
 		c.channelConfig,
@@ -106,6 +105,26 @@ func (c *Service) Scan(ctx context.Context, txID string, callback driver.Deliver
 		c.ConfigService,
 		c.PeerManager,
 		c.Ledger,
+		callback,
+		vault,
+		c.channelConfig.CommitterWaitForEventTimeout(),
+		&noop.TracerProvider{},
+		&disabled.Provider{},
+	)
+	if err != nil {
+		return err
+	}
+
+	return deliveryService.Run(ctx)
+}
+
+func (c *Service) ScanBlock(ctx context.Context, callback driver.BlockCallback) error {
+	return c.scanBlock(ctx, &fakeVault{}, callback)
+}
+
+func (c *Service) ScanTx(ctx context.Context, txID string, callback driver.DeliveryCallback) error {
+	vault := &fakeVault{txID: txID}
+	return c.scanBlock(ctx, vault,
 		func(_ context.Context, block *common.Block) (bool, error) {
 			for i, tx := range block.Data.Data {
 				validationCode := ValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])[i]
@@ -140,17 +159,7 @@ func (c *Service) Scan(ctx context.Context, txID string, callback driver.Deliver
 				logger.Debugf("commit transaction [%s] in block [%d]", channelHeader.TxId, block.Header.Number)
 			}
 			return false, nil
-		},
-		vault,
-		c.channelConfig.CommitterWaitForEventTimeout(),
-		&noop.TracerProvider{},
-		&disabled.Provider{},
-	)
-	if err != nil {
-		return err
-	}
-
-	return deliveryService.Run(ctx)
+		})
 }
 
 type fakeVault struct {

platform/fabric/delivery.go

@@ -10,20 +10,29 @@ import (
 	"context"
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/fabric/driver"
+	"github.com/hyperledger/fabric-protos-go/common"
 )
 
 type DeliveryCallback func(tx *ProcessedTransaction) (bool, error)
 
+type BlockCallback func(context.Context, *common.Block) (bool, error)
+
 // Delivery models the Fabric's delivery service
 type Delivery struct {
 	delivery driver.Delivery
 }
 
-// Scan iterates over all transactions in block starting from the block containing the passed transaction id.
+func (d *Delivery) ScanBlock(ctx context.Context, callback BlockCallback) error {
+	return d.delivery.ScanBlock(ctx, func(ctx context.Context, block *common.Block) (bool, error) {
+		return callback(ctx, block)
+	})
+}
+
+// ScanTx iterates over all transactions in block starting from the block containing the passed transaction id.
 // If txID is empty, the iterations starts from the first block.
 // On each transaction, the callback function is invoked.
-func (d *Delivery) Scan(ctx context.Context, txID string, callback DeliveryCallback) error {
-	return d.delivery.Scan(ctx, txID, func(tx driver.ProcessedTransaction) (bool, error) {
+func (d *Delivery) ScanTx(ctx context.Context, txID string, callback DeliveryCallback) error {
+	return d.delivery.ScanTx(ctx, txID, func(tx driver.ProcessedTransaction) (bool, error) {
 		return callback(&ProcessedTransaction{
 			pt: tx,
 		})

platform/fabric/driver/delivery.go

@@ -6,19 +6,33 @@ SPDX-License-Identifier: Apache-2.0
 
 package driver
 
-import "context"
+import (
+	"context"
+
+	"github.com/hyperledger/fabric-protos-go/common"
+)
 
 // DeliveryCallback is a callback function used to process a transaction.
 // Return true, if the scan should finish.
 type DeliveryCallback func(tx ProcessedTransaction) (bool, error)
 
+// BlockCallback is the callback function prototype to alert the rest of the stack about the availability of a new block.
+// The function returns two argument a boolean to signal if delivery should be stopped, and an error
+// to signal an issue during the processing of the block.
+// In case of an error, the same block is re-processed after a delay.
+type BlockCallback func(context.Context, *common.Block) (bool, error)
+
 // Delivery gives access to Fabric channel delivery
 type Delivery interface {
-	// StartDelivery starts the delivery process
+	// Start starts the delivery process
 	Start(ctx context.Context) error
 
-	// Scan iterates over all transactions in block starting from the block containing the passed transaction id.
+	// ScanBlock iterates over all blocks.
+	// On each block, the callback function is invoked.
+	ScanBlock(ctx context.Context, callback BlockCallback) error
+
+	// ScanTx iterates over all transactions in block starting from the block containing the passed transaction id.
 	// If txID is empty, the iterations starts from the first block.
 	// On each transaction, the callback function is invoked.
-	Scan(ctx context.Context, txID string, callback DeliveryCallback) error
+	ScanTx(ctx context.Context, txID string, callback DeliveryCallback) error
 }