[FABG-783] Invalid last block received in event client

The service.Dispatcher was being copied into the
deliverclient Dispatcher and therefore there were two
copies of lastBlockNum. This patch uses a reference to
the Dispatcher instead of a copy.

Change-Id: I3fdad90146c46306e0bf3799698a5e025ddf3742
Signed-off-by: Bob Stasyszyn <[email protected]>

pkg/fab/events/client/client.go

@@ -38,7 +38,7 @@ const (
 // chaincode, and transaction status events. Client also monitors the connection to the
 // event server and attempts to reconnect if the connection is closed.
 type Client struct {
-	eventservice.Service
+	*eventservice.Service
 	params
 	sync.RWMutex
 	connEvent       chan *dispatcher.ConnectionEvent
@@ -57,7 +57,7 @@ func New(dispatcher eventservice.Dispatcher, opts ...options.Opt) *Client {
 	options.Apply(params, opts)
 
 	return &Client{
-		Service:         *eventservice.New(dispatcher, opts...),
+		Service:         eventservice.New(dispatcher, opts...),
 		params:          *params,
 		connectionState: int32(Disconnected),
 	}

pkg/fab/events/client/dispatcher/dispatcher.go

@@ -28,7 +28,7 @@ var logger = logging.NewLogger("fabsdk/fab")
 // in order to avoid any race conditions and to ensure that events are processed in the order that they are received.
 // This avoids the need for synchronization.
 type Dispatcher struct {
-	esdispatcher.Dispatcher
+	*esdispatcher.Dispatcher
 	params
 	context                context.Client
 	chConfig               fab.ChannelCfg
@@ -48,7 +48,7 @@ func New(context context.Client, chConfig fab.ChannelCfg, discoveryService fab.D
 	options.Apply(params, opts)
 
 	dispatcher := &Dispatcher{
-		Dispatcher:         *esdispatcher.New(opts...),
+		Dispatcher:         esdispatcher.New(opts...),
 		params:             *params,
 		context:            context,
 		chConfig:           chConfig,

pkg/fab/events/deliverclient/deliverclient.go

@@ -54,7 +54,7 @@ var deliverFilteredProvider = func(context fabcontext.Client, chConfig fab.Chann
 
 // Client connects to a peer and receives channel events, such as bock, filtered block, chaincode, and transaction status events.
 type Client struct {
-	client.Client
+	*client.Client
 	params
 }
 
@@ -81,7 +81,7 @@ func New(context fabcontext.Client, chConfig fab.ChannelCfg, discoveryService fa
 	}
 
 	client := &Client{
-		Client: *client.New(dispatcher, opts...),
+		Client: client.New(dispatcher, opts...),
 		params: *params,
 	}
 

pkg/fab/events/deliverclient/dispatcher/dispatcher.go

@@ -33,13 +33,13 @@ type dsConnection interface {
 // in order to avoid any race conditions and to ensure that events are processed in the order that they are received.
 // This also avoids the need for synchronization.
 type Dispatcher struct {
-	clientdisp.Dispatcher
+	*clientdisp.Dispatcher
 }
 
 // New returns a new deliver dispatcher
 func New(context fabcontext.Client, chConfig fab.ChannelCfg, discoveryService fab.DiscoveryService, connectionProvider api.ConnectionProvider, opts ...options.Opt) *Dispatcher {
 	return &Dispatcher{
-		Dispatcher: *clientdisp.New(context, chConfig, discoveryService, connectionProvider, opts...),
+		Dispatcher: clientdisp.New(context, chConfig, discoveryService, connectionProvider, opts...),
 	}
 }
 

pkg/fab/events/deliverclient/dispatcher/dispatcher_test.go

@@ -21,6 +21,7 @@ import (
 	fabmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
 	mspmocks "github.com/hyperledger/fabric-sdk-go/pkg/msp/test/mockmsp"
 	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
 )
 
 var (
@@ -206,6 +207,16 @@ func TestBlockEvents(t *testing.T) {
 
 	checkBlockEvents(eventch, t)
 
+	lastBlockReceived := dispatcher.LastBlockNum()
+	assert.Equal(t, uint64(0), lastBlockReceived)
+
+	ledger.NewBlock(channelID)
+
+	checkBlockEvents(eventch, t)
+
+	lastBlockReceived = dispatcher.LastBlockNum()
+	assert.Equal(t, uint64(1), lastBlockReceived)
+
 	// Unregister block events
 	dispatcherEventch <- esdispatcher.NewUnregisterEvent(reg)
 

pkg/fab/events/service/dispatcher/dispatcher.go

@@ -157,6 +157,7 @@ func (ed *Dispatcher) updateLastBlockNum(blockNum uint64) error {
 	lastBlockNum := atomic.LoadUint64(&ed.lastBlockNum)
 	if lastBlockNum == math.MaxUint64 || blockNum > lastBlockNum {
 		atomic.StoreUint64(&ed.lastBlockNum, blockNum)
+		logger.Debugf("Updated last block received to %d", blockNum)
 		return nil
 	}
 	return errors.Errorf("Expecting a block number greater than %d but received block number %d", lastBlockNum, blockNum)