Added support for fromBlock when generating eventservice cache keys (#184)

* Added support for seek type and fromBlock parameters when generating event service cache keys

Signed-off-by: Jim Zhang <[email protected]>

* Add chaincode ID to the cache key calculation

Signed-off-by: Jim Zhang <[email protected]>

* Fix lint errors

Signed-off-by: Jim Zhang <[email protected]>

* revert local temporary changes checked in by accident

Signed-off-by: Jim Zhang <[email protected]>

* Unit test coverage

Signed-off-by: Jim Zhang <[email protected]>

* Add unit test to cachekey.go

Signed-off-by: Jim Zhang <[email protected]>

Author: jimthematrix

pkg/client/channel/chclientrun_std.go

@@ -1,3 +1,4 @@
+//go:build !pprof
 // +build !pprof
 
 /*

pkg/client/event/event.go

@@ -33,11 +33,13 @@ type Client struct {
 	permitBlockEvents    bool
 	fromBlock            uint64
 	seekType             seek.Type
+	chaincodeID          string
 	eventConsumerTimeout *time.Duration
 }
 
 // New returns a Client instance. Client receives events such as block, filtered block,
 // chaincode, and transaction status events.
+// nolint: gocyclo
 func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client, error) {
 
 	channelContext, err := channelProvider()
@@ -68,6 +70,9 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 				opts = append(opts, deliverclient.WithBlockNum(eventClient.fromBlock))
 			}
 		}
+		if eventClient.chaincodeID != "" {
+			opts = append(opts, deliverclient.WithChaincodeID(eventClient.chaincodeID))
+		}
 		if eventClient.eventConsumerTimeout != nil {
 			opts = append(opts, dispatcher.WithEventConsumerTimeout(*eventClient.eventConsumerTimeout))
 		}

pkg/client/event/event_test.go

@@ -20,11 +20,11 @@ import (
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 
+	pb "github.com/hyperledger/fabric-protos-go/peer"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/seek"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service/dispatcher"
 	servicemocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service/mocks"
-	pb "github.com/hyperledger/fabric-protos-go/peer"
 )
 
 var (
@@ -43,7 +43,7 @@ func TestNewEventClient(t *testing.T) {
 		t.Fatalf("Failed to create new event client: %s", err)
 	}
 
-	_, err = New(ctx, WithBlockEvents(), WithSeekType(seek.Newest), WithBlockNum(math.MaxUint64), WithEventConsumerTimeout(500 * time.Millisecond))
+	_, err = New(ctx, WithBlockEvents(), WithSeekType(seek.Newest), WithBlockNum(math.MaxUint64), WithEventConsumerTimeout(500*time.Millisecond), WithChaincodeID("testChaincode"))
 	if err != nil {
 		t.Fatalf("Failed to create new event client: %s", err)
 	}
@@ -55,6 +55,22 @@ func TestNewEventClient(t *testing.T) {
 	}
 }
 
+func TestNewEventClientWithFromBlock(t *testing.T) {
+
+	fabCtx := setupCustomTestContext(t, nil)
+	ctx := createChannelContext(fabCtx, channelID)
+
+	_, err := New(ctx)
+	if err != nil {
+		t.Fatalf("Failed to create new event client: %s", err)
+	}
+
+	_, err = New(ctx, WithBlockEvents(), WithSeekType(seek.FromBlock), WithBlockNum(100), WithChaincodeID("testChaincode"))
+	if err != nil {
+		t.Fatalf("Failed to create new event client: %s", err)
+	}
+}
+
 func TestBlockEvents(t *testing.T) {
 
 	eventService, eventProducer, err := newServiceWithMockProducer(defaultOpts, withBlockLedger(sourceURL))

pkg/client/event/example_test.go

@@ -77,6 +77,25 @@ func ExampleClient_RegisterChaincodeEvent() {
 
 }
 
+func ExampleClient_RegisterChaincodeEvent_NewService() {
+
+	ec, err := New(mockChannelProvider("mychannel"), WithChaincodeID("examplecc"))
+	if err != nil {
+		fmt.Println("failed to create client")
+	}
+
+	registration, _, err := ec.RegisterChaincodeEvent("examplecc", "event123")
+	if err != nil {
+		fmt.Println("failed to register chaincode event")
+	}
+	defer ec.Unregister(registration)
+
+	fmt.Println("chaincode event registered successfully")
+
+	// Output: chaincode event registered successfully
+
+}
+
 func ExampleClient_RegisterChaincodeEvent_withPayload() {
 
 	// If you require payload for chaincode events you have to use WithBlockEvents() option

pkg/client/event/opts.go

@@ -42,6 +42,15 @@ func WithSeekType(seek seek.Type) ClientOption {
 	}
 }
 
+// WithChaincodeID indicates the target chaincode
+// Only deliverclient supports this
+func WithChaincodeID(id string) ClientOption {
+	return func(c *Client) error {
+		c.chaincodeID = id
+		return nil
+	}
+}
+
 // WithEventConsumerTimeout is the timeout when sending events to a registered consumer.
 // If < 0, if buffer full, unblocks immediately and does not send.
 // If 0, if buffer full, will block and guarantee the event will be sent out.

pkg/fab/events/deliverclient/deliverclient_test.go

@@ -72,6 +72,7 @@ func TestClientConnect(t *testing.T) {
 		),
 		WithSeekType(seek.FromBlock),
 		WithBlockNum(0),
+		WithChaincodeID("testChaincode"),
 		client.WithResponseTimeout(3*time.Second),
 	)
 	if err != nil {

pkg/fab/events/deliverclient/opts.go

@@ -19,6 +19,7 @@ type params struct {
 	connProvider api.ConnectionProvider
 	seekType     seek.Type
 	fromBlock    uint64
+	chaincodeID  string
 	respTimeout  time.Duration
 }
 
@@ -48,6 +49,15 @@ func WithBlockNum(value uint64) options.Opt {
 	}
 }
 
+// WithChaincodeID specifies the chaincode from which events are to be received.
+func WithChaincodeID(value string) options.Opt {
+	return func(p options.Params) {
+		if setter, ok := p.(chaincodeIDSetter); ok {
+			setter.SetChaincodeID(value)
+		}
+	}
+}
+
 type seekTypeSetter interface {
 	SetSeekType(value seek.Type)
 }
@@ -56,6 +66,10 @@ type fromBlockSetter interface {
 	SetFromBlock(value uint64)
 }
 
+type chaincodeIDSetter interface {
+	SetChaincodeID(value string)
+}
+
 func (p *params) PermitBlockEvents() {
 	logger.Debug("PermitBlockEvents")
 	p.connProvider = deliverProvider
@@ -79,6 +93,11 @@ func (p *params) SetSeekType(value seek.Type) {
 	}
 }
 
+func (p *params) SetChaincodeID(value string) {
+	logger.Debugf("ChaincodId: %d", value)
+	p.chaincodeID = value
+}
+
 func (p *params) SetResponseTimeout(value time.Duration) {
 	logger.Debugf("ResponseTimeout: %s", value)
 	p.respTimeout = value

pkg/fabsdk/fabsdk_std.go

@@ -1,3 +1,4 @@
+//go:build !pprof
 // +build !pprof
 
 /*

pkg/fabsdk/provider/chpvdr/cachekey.go

@@ -8,10 +8,11 @@ package chpvdr
 
 import (
 	"crypto/sha256"
-	"strconv"
+	"fmt"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/options"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/seek"
 )
 
 // ctxtCacheKey is a lazy cache key for the context cache
@@ -99,6 +100,9 @@ func (k *eventCacheKey) String() string {
 
 type params struct {
 	permitBlockEvents bool
+	seekType          seek.Type
+	fromBlock         uint64
+	chaincodeID       string
 }
 
 func defaultParams() *params {
@@ -109,8 +113,24 @@ func (p *params) PermitBlockEvents() {
 	p.permitBlockEvents = true
 }
 
+func (p *params) SetFromBlock(value uint64) {
+	p.fromBlock = value
+}
+
+func (p *params) SetSeekType(value seek.Type) {
+	if value != "" {
+		p.seekType = value
+	}
+}
+
+func (p *params) SetChaincodeID(value string) {
+	if value != "" {
+		p.chaincodeID = value
+	}
+}
+
 func (p *params) getOptKey() string {
 	//	Construct opts portion
-	optKey := "blockEvents:" + strconv.FormatBool(p.permitBlockEvents)
+	optKey := fmt.Sprintf("blockEvents:%t,seekType:%s,fromBlock:%d,chaincodeId:%s", p.permitBlockEvents, p.seekType, p.fromBlock, p.chaincodeID)
 	return optKey
 }

pkg/fabsdk/provider/chpvdr/chprovider_test.go

@@ -1,3 +1,4 @@
+//go:build testing
 // +build testing
 
 /*
@@ -24,6 +25,8 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/chconfig"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/discovery"
 	discmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/discovery/mocks"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
 	mspmocks "github.com/hyperledger/fabric-sdk-go/pkg/msp/test/mockmsp"
 	"github.com/pkg/errors"
@@ -83,7 +86,7 @@ func TestBasicValidChannel(t *testing.T) {
 	assert.NotNil(t, channelConfig)
 	assert.NotEmptyf(t, channelConfig.ID(), "Got empty channel ID from channel config")
 
-	eventService, err := channelService.EventService()
+	eventService, err := channelService.EventService(client.WithBlockEvents(), deliverclient.WithSeekType("from"), deliverclient.WithBlockNum(10), deliverclient.WithChaincodeID("testChaincode"))
 	require.NoError(t, err)
 	require.NotNil(t, eventService)