parsed-blocks-deliverer (#145)

* 1

* 2

* 3

* observer refactoring to generics

* observer refactoring to generics without Parsed Common

* pre final vers after review

* pre final vers after review #2

* pre final vers after review #3

* fix linters

* fix linters #2

---------

Co-authored-by: Nikita Neznaemov <[email protected]>

.github/workflows/golangci-lint.yml

@@ -20,7 +20,7 @@ jobs:
         uses: golangci/golangci-lint-action@v3
         with:
           # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
-          version: v1.52
+          version: v1.57
 
           # Optional: working directory, useful for monorepos
           # working-directory: somedir

api/client.go

@@ -7,6 +7,8 @@ import (
 	"github.com/hyperledger/fabric-protos-go/common"
 	"github.com/hyperledger/fabric-protos-go/peer"
 	"github.com/hyperledger/fabric/msp"
+
+	hlfproto "github.com/s7techlab/hlf-sdk-go/block"
 )
 
 type CurrentIdentity interface {
@@ -43,6 +45,16 @@ type BlocksDeliverer interface {
 	) (blockChan <-chan *common.Block, closer func() error, err error)
 }
 
+type ParsedBlocksDeliverer interface {
+	// ParsedBlocks the same as BlocksDeliverer.Blocks, but returns a channel with parsed blocks
+	ParsedBlocks(
+		ctx context.Context,
+		channel string,
+		identity msp.SigningIdentity,
+		blockRange ...int64,
+	) (parsedBlockChan <-chan *hlfproto.Block, parsedCloser func() error, err error)
+}
+
 type Querier interface {
 	CurrentIdentity
 	// Query - shortcut for querying chaincodes

api/config/config_yaml.go

@@ -1,14 +1,14 @@
 package config
 
 import (
-	"io/ioutil"
+	"os"
 
 	"github.com/pkg/errors"
 	"gopkg.in/yaml.v2"
 )
 
 func NewYamlConfig(configPath string) (*Config, error) {
-	if configBytes, err := ioutil.ReadFile(configPath); err != nil {
+	if configBytes, err := os.ReadFile(configPath); err != nil {
 		return nil, errors.Wrap(err, `failed to read config file`)
 	} else {
 		var c Config

api/peer.go

@@ -35,12 +35,14 @@ type Peer interface {
 
 	BlocksDeliverer
 
+	ParsedBlocksDeliverer
+
 	EventsDeliverer
 
 	// DeliverClient returns DeliverClient
 	DeliverClient(identity msp.SigningIdentity) (DeliverClient, error)
-	// Uri returns url used for grpc connection
-	Uri() string
+	// URI returns url used for grpc connection
+	URI() string
 	// Conn returns instance of grpc connection
 	Conn() *grpc.ClientConn
 	// Close terminates peer connection

block/block.go

@@ -1,6 +1,7 @@
 package block
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/golang/protobuf/proto"
@@ -14,6 +15,11 @@ import (
 	"github.com/s7techlab/hlf-sdk-go/block/txflags"
 )
 
+var (
+	ErrNilBlock       = errors.New("nil block")
+	ErrNilConfigBlock = errors.New("nil config block")
+)
+
 type (
 	parseBlockOpts struct {
 		configBlock *common.Block
@@ -71,6 +77,10 @@ func ParseBlock(block *common.Block, opts ...ParseBlockOpt) (*Block, error) {
 }
 
 func ParseOrdererIdentity(cb *common.Block) (*msp.SerializedIdentity, error) {
+	if cb == nil {
+		return nil, ErrNilBlock
+	}
+
 	meta, err := protoutil.GetMetadataFromBlock(cb, common.BlockMetadataIndex_SIGNATURES)
 	if err != nil {
 		return nil, fmt.Errorf("get metadata from block: %w", err)
@@ -96,6 +106,14 @@ func ParseOrdererIdentity(cb *common.Block) (*msp.SerializedIdentity, error) {
 }
 
 func ParseBTFOrderersIdentities(block *common.Block, configBlock *common.Block) ([]*OrdererSignature, error) {
+	if block == nil {
+		return nil, ErrNilBlock
+	}
+
+	if configBlock == nil {
+		return nil, ErrNilConfigBlock
+	}
+
 	bftMeta := &bftcommon.BFTMetadata{}
 	if err := proto.Unmarshal(block.Metadata.Metadata[common.BlockMetadataIndex_SIGNATURES], bftMeta); err != nil {
 		return nil, fmt.Errorf("unmarshaling bft block metadata from metadata: %w", err)

observer/transform/action.go → block/transform/action.go

@@ -4,10 +4,9 @@ import (
 	"fmt"
 	"regexp"
 
-	"github.com/mohae/deepcopy"
+	"google.golang.org/protobuf/proto"
 
 	hlfproto "github.com/s7techlab/hlf-sdk-go/block"
-	"github.com/s7techlab/hlf-sdk-go/observer"
 )
 
 type (
@@ -68,77 +67,63 @@ func NewAction(actionMach TxActionMatch, opts ...ActionOpt) *Action {
 	return a
 }
 
-func (s *Action) Transform(block *observer.ParsedBlock) error {
-	if block.Block == nil {
-		return nil
+func (s *Action) Transform(block *hlfproto.Block) (*hlfproto.Block, error) {
+	if block == nil {
+		return nil, hlfproto.ErrNilBlock
 	}
 
-	// if block is transformed, copy of block will be saved to block.BlockOriginal
-	blockCopy := deepcopy.Copy(block.Block).(*hlfproto.Block)
-	blockIsTransformed := false
+	// make block copy not to change original
+	blockCopy := proto.Clone(block).(*hlfproto.Block)
 
-	for _, envelope := range block.Block.Data.Envelopes {
-		if envelope.Payload.Transaction == nil {
+	for _, envelope := range blockCopy.GetData().GetEnvelopes() {
+		if envelope.GetPayload().GetTransaction() == nil {
 			continue
 		}
 
-		for _, txAction := range envelope.Payload.Transaction.Actions {
+		for _, txAction := range envelope.GetPayload().GetTransaction().GetActions() {
 			if !s.match(txAction) {
 				continue
 			}
 
 			for _, argsTransformer := range s.inputArgsTransformers {
-				if err := argsTransformer.Transform(txAction.ChaincodeSpec().Input.Args); err != nil {
-					return fmt.Errorf(`transform input args: %w`, err)
+				if err := argsTransformer.Transform(txAction.ChaincodeSpec().GetInput().GetArgs()); err != nil {
+					return nil, fmt.Errorf(`transform input args: %w`, err)
 				}
 			}
 
 			for _, eventTransformer := range s.eventTransformers {
 				if err := eventTransformer.Transform(txAction.Event()); err != nil {
-					return fmt.Errorf(`transform event: %w`, err)
+					return nil, fmt.Errorf(`transform event: %w`, err)
 				}
 			}
 
 			for _, rwSet := range txAction.NsReadWriteSet() {
-				for _, write := range rwSet.Rwset.Writes {
+				for _, write := range rwSet.GetRwset().GetWrites() {
 					for _, kvWriteTransformer := range s.kvWriteTransformers {
-						origKey := write.Key
 						if err := kvWriteTransformer.Transform(write); err != nil {
-							return fmt.Errorf(`transform KV write with key: %s: %w`, write.Key, err)
-						}
-
-						if origKey != write.Key {
-							blockIsTransformed = true
+							return nil, fmt.Errorf(`transform KV write with key: %s: %w`, write.Key, err)
 						}
 					}
 				}
 
-				for _, read := range rwSet.Rwset.Reads {
+				for _, read := range rwSet.GetRwset().GetReads() {
 					for _, kvReadTransform := range s.kvReadTransformers {
-						origKey := read.Key
 						if err := kvReadTransform.Transform(read); err != nil {
-							return fmt.Errorf(`transform KV read with key: %s: %w`, read.Key, err)
-						}
-						if origKey != read.Key {
-							blockIsTransformed = true
+							return nil, fmt.Errorf(`transform KV read with key: %s: %w`, read.Key, err)
 						}
 					}
 				}
 
 				for _, actionPayloadTransform := range s.actionPayloadTransformers {
 					if err := actionPayloadTransform.Transform(txAction); err != nil {
-						return fmt.Errorf(`transform action payload: %w`, err)
+						return nil, fmt.Errorf(`transform action payload: %w`, err)
 					}
 				}
 			}
 		}
 	}
 
-	if blockIsTransformed {
-		block.BlockOriginal = blockCopy
-	}
-
-	return nil
+	return blockCopy, nil
 }
 
 func TxChaincodeIDMatch(chaincode string) TxActionMatch {

observer/transform/lifecycle.go → block/transform/lifecycle.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/hyperledger/fabric-protos-go/ledger/rwset/kvrwset"
 
-	"github.com/s7techlab/hlf-sdk-go/observer"
+	hlfproto "github.com/s7techlab/hlf-sdk-go/block"
 )
 
 const (
@@ -53,7 +53,7 @@ func keyReplace(key string) string {
 	return key
 }
 
-var LifecycleTransformers = []observer.BlockTransformer{
+var LifecycleTransformers = []hlfproto.Transformer{
 	NewAction(
 		TxChaincodeIDMatch(LifecycleChaincodeName),
 		WithKVWriteTransformer(

block/transformer.go

@@ -0,0 +1,6 @@
+package block
+
+// Transformer transforms parsed observer data. For example decrypt, or transformer protobuf state to json
+type Transformer interface {
+	Transform(*Block) (*Block, error)
+}

client/ca/http/client.go

@@ -4,7 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 
 	"github.com/golang/protobuf/proto"
@@ -92,7 +92,7 @@ func (c *Client) setAuthToken(req *http.Request, body []byte) error {
 
 func (c *Client) processResponse(resp *http.Response, out interface{}, expectedHTTPStatuses ...int) error {
 	defer func() { _ = resp.Body.Close() }()
-	body, err := ioutil.ReadAll(resp.Body)
+	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return errors.Wrap(err, `failed to read response body`)
 	}

client/chaincode/invoke_test.go

@@ -160,8 +160,8 @@ package chaincode_test
 //	return p.deliver, nil
 //}
 //
-//// Uri returns url used for grpc connection
-//func (p *mockPeer) Uri() string {
+//// URI returns url used for grpc connection
+//func (p *mockPeer) URI() string {
 //	return "localhost:7051"
 //}
 //

client/core_opts.go

@@ -2,7 +2,7 @@ package client
 
 import (
 	"fmt"
-	"io/ioutil"
+	"os"
 
 	"github.com/hyperledger/fabric/msp"
 	"github.com/pkg/errors"
@@ -51,7 +51,7 @@ func WithOrderer(orderer api.Orderer) Opt {
 // WithConfigYaml allows passing path to YAML configuration file
 func WithConfigYaml(configPath string) Opt {
 	return func(c *Client) error {
-		configBytes, err := ioutil.ReadFile(configPath)
+		configBytes, err := os.ReadFile(configPath)
 		if err != nil {
 			return errors.Wrap(err, `failed to read config file`)
 		}

client/core_public.go

@@ -10,6 +10,7 @@ import (
 	"github.com/hyperledger/fabric/msp"
 
 	"github.com/s7techlab/hlf-sdk-go/api"
+	"github.com/s7techlab/hlf-sdk-go/block"
 	"github.com/s7techlab/hlf-sdk-go/client/chaincode"
 	"github.com/s7techlab/hlf-sdk-go/client/chaincode/txwaiter"
 	"github.com/s7techlab/hlf-sdk-go/client/tx"
@@ -108,7 +109,7 @@ func (c *Client) Blocks(
 	channel string,
 	identity msp.SigningIdentity,
 	blockRange ...int64,
-) (blocks <-chan *common.Block, closer func() error, _ error) {
+) (<-chan *common.Block, func() error, error) {
 	if identity == nil {
 		identity = c.CurrentIdentity()
 	}
@@ -120,3 +121,21 @@ func (c *Client) Blocks(
 
 	return peer.Blocks(ctx, channel, identity, blockRange...)
 }
+
+func (c *Client) ParsedBlocks(
+	ctx context.Context,
+	channel string,
+	identity msp.SigningIdentity,
+	blockRange ...int64,
+) (<-chan *block.Block, func() error, error) {
+	if identity == nil {
+		identity = c.CurrentIdentity()
+	}
+
+	peer, err := c.PeerPool().FirstReadyPeer(identity.GetMSPIdentifier())
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return peer.ParsedBlocks(ctx, channel, identity, blockRange...)
+}