observer refactoring to generics

block/block.proto

@@ -17,7 +17,6 @@ import "hyperledger/fabric-protos/peer/proposal_response.proto";
 import "hyperledger/fabric-protos/peer/transaction.proto";
 
 message Block {
-  string channel = 4;
   common.BlockHeader header = 1;
   BlockData data = 2;
   BlockMetadata metadata = 3;

client/peer.go

@@ -80,7 +80,6 @@ func NewPeer(dialCtx context.Context, c config.ConnectionConfig, identity msp.Si
 
 // NewFromGRPC allows initializing peer from existing GRPC connection
 func NewFromGRPC(conn *grpc.ClientConn, identity msp.SigningIdentity, tlsCertHash []byte, logger *zap.Logger, endorseDefaultTimeout time.Duration) (api.Peer, error) {
-
 	if conn == nil {
 		return nil, errors.New(`empty connection`)
 	}
@@ -95,9 +94,29 @@ func NewFromGRPC(conn *grpc.ClientConn, identity msp.SigningIdentity, tlsCertHas
 		identity:              identity,
 		tlsCertHash:           tlsCertHash,
 		endorseDefaultTimeout: endorseDefaultTimeout,
+		configBlocks:          make(map[string]*common.Block),
 		logger:                logger.Named(`peer`),
 	}
 
+	qsccService := qscc.NewQSCC(p)
+	ctx := context.Background()
+
+	channels, err := p.GetChannels(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("get all channels: %w", err)
+	}
+
+	for _, ch := range channels.GetChannels() {
+		configBlock, err := qsccService.GetBlockByNumber(ctx, &qscc.GetBlockByNumberRequest{ChannelName: ch.ChannelId, BlockNumber: 0})
+		if err != nil {
+			return nil, fmt.Errorf("get block by number from channel %s: %w", ch.ChannelId, err)
+		}
+
+		if configBlock != nil {
+			p.configBlocks[ch.ChannelId] = configBlock
+		}
+	}
+
 	return p, nil
 }
 
@@ -198,12 +217,7 @@ func (p *peer) ParsedBlocks(ctx context.Context, channel string, identity msp.Si
 				}
 
 				p.mu.Lock()
-				var configBlock *common.Block
-				if b.Header.Number == 0 {
-					p.configBlocks[channel] = configBlock
-				} else {
-					configBlock = p.configBlocks[channel]
-				}
+				configBlock := p.configBlocks[channel]
 				p.mu.Unlock()
 
 				parsedBlock, err := block.ParseBlock(b, block.WithConfigBlock(configBlock))

observer/all_channels_blocks.go

@@ -189,7 +189,7 @@ func (acb *allChannelsBlocks[T]) startNotObservedChannels(ctx context.Context, n
 	for _, notObservedChannel := range notObservedChannels {
 		chBlocks := notObservedChannel
 
-		if _, err := chBlocks.observe(ctx); err != nil {
+		if _, err := chBlocks.Observe(ctx); err != nil {
 			acb.logger.Warn(`init channel observer`, zap.String("channel", notObservedChannel.channel), zap.Error(err))
 		}
 

observer/all_channels_blocks_concurrently.go

@@ -67,7 +67,7 @@ func (acb *allChannelsBlocks[T]) startNotObservedChannelsConcurrently(
 	for _, notObservedChannel := range notObservedChannels {
 		chBlocks := notObservedChannel
 
-		if _, err := chBlocks.observe(ctx); err != nil {
+		if _, err := chBlocks.Observe(ctx); err != nil {
 			acb.logger.Warn(`init channel observer concurrently`, zap.String("channel", notObservedChannel.channel), zap.Error(err))
 		}
 

observer/channel_blocks.go

@@ -96,7 +96,7 @@ func (c *channelBlocks[T]) Stop() error {
 	return err
 }
 
-func (c *channelBlocks[T]) observe(ctx context.Context) (<-chan *Block[T], error) {
+func (c *channelBlocks[T]) Observe(ctx context.Context) (<-chan *Block[T], error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()