[FABG-761] Added prefer-org and prefer-peer resolvers

Implemented two new peer resolvers:

Prefer-Org:
- Determines which peers are suitable based on block height, although
  will prefer the peers in the current org (as long as their block height
  is above a configured threshold). If none of the peers from the current
  org are suitable then a peer from another org is chosen.

Prefer-Peer:
- Determines which peers are suitable based on block height, although
  will prefer the peers in the provided list (as long as their block
  height is above a configured threshold). If none of the peers in the
  provided list are suitable then an attempt is made to select a peer from
  the current org will be selected. If none of the peers from the current
  org are suitable then a peer from another org is chosen.

Change-Id: Ieffee027ba03550cf83399742a7ac9f44e38c96f
Signed-off-by: Bob Stasyszyn <Bob.Stasyszyn@securekey.com>

Author: bstasyszyn

pkg/fab/events/client/client_test.go

@@ -15,16 +15,18 @@ import (
 	"testing"
 	"time"
 
-	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/lbp"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/peerresolver/preferorg"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/options"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/api"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/dispatcher"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/lbp"
 	clientmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/mocks"
 	mockconn "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/mocks"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/peerresolver/minblockheight"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/peerresolver/preferpeer"
 	esdispatcher "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service/dispatcher"
 	servicemocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service/mocks"
 	fabmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
@@ -1449,8 +1451,6 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx4", pb.TxValidationCode_VALID))
 	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx5", pb.TxValidationCode_VALID))
 
-	time.Sleep(time.Second)
-
 	// Set the block height of another peer to be greater than the disconnect threshold
 	// so that the event client can reconnect to another peer
 	p2.SetBlockHeight(9)
@@ -1463,6 +1463,152 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 	}
 }
 
+// TestPreferLocalOrgConnection tests the scenario where an org wishes to connect to it's own peers
+// if they are above the block height lag threshold but, if they fall below the threshold, the
+// connection should be made to another org's peer. Once the local org's peers have caught up in
+// block height, the connection to the local peer should be re-established.
+func TestPreferLocalOrgConnection(t *testing.T) {
+	channelID := "testchannel"
+	org1MSP := "Org1MSP"
+	org2MSP := "Org2MSP"
+	blockHeightLagThreshold := 2
+
+	p1O1 := clientmocks.NewMockStatefulPeer("p1_o1", "peer1.org1.com:7051", clientmocks.WithMSP(org1MSP), clientmocks.WithBlockHeight(4))
+	p2O1 := clientmocks.NewMockStatefulPeer("p2_o1", "peer2.org1.com:7051", clientmocks.WithMSP(org1MSP), clientmocks.WithBlockHeight(3))
+	p1O2 := clientmocks.NewMockStatefulPeer("p1_o2", "peer1.org2.com:7051", clientmocks.WithMSP(org2MSP), clientmocks.WithBlockHeight(10))
+	p2O2 := clientmocks.NewMockStatefulPeer("p2_o2", "peer1.org2.com:7051", clientmocks.WithMSP(org2MSP), clientmocks.WithBlockHeight(11))
+
+	connectch := make(chan *dispatcher.ConnectionEvent)
+
+	conn := clientmocks.NewMockConnection(
+		clientmocks.WithLedger(servicemocks.NewMockLedger(servicemocks.FilteredBlockEventFactory, sourceURL)),
+	)
+	connectionProvider := clientmocks.NewProviderFactory().Provider(conn)
+
+	eventClient, _, err := newClientWithMockConnAndOpts(
+		fabmocks.NewMockContext(
+			mspmocks.NewMockSigningIdentity("user1", "Org1MSP"),
+		),
+		fabmocks.NewMockChannelCfg(channelID),
+		clientmocks.NewDiscoveryService(p1O1, p2O1, p1O2, p2O2),
+		connectionProvider, filteredClientProvider,
+		[]options.Opt{
+			esdispatcher.WithEventConsumerTimeout(3 * time.Second),
+			WithMaxConnectAttempts(1),
+			WithTimeBetweenConnectAttempts(time.Millisecond),
+			WithConnectionEvent(connectch),
+			WithResponseTimeout(2 * time.Second),
+			dispatcher.WithPeerResolver(preferorg.NewResolver()),
+			dispatcher.WithLoadBalancePolicy(lbp.NewRoundRobin()),
+			dispatcher.WithPeerMonitorPeriod(250 * time.Millisecond),
+			minblockheight.WithBlockHeightLagThreshold(blockHeightLagThreshold),
+			minblockheight.WithReconnectBlockHeightThreshold(3),
+		},
+	)
+	require.NoErrorf(t, err, "error creating channel event client")
+	err = eventClient.Connect()
+	require.NoErrorf(t, err, "errorconnecting channel event client")
+	defer eventClient.Close()
+
+	connectedPeer := eventClient.Dispatcher().(*dispatcher.Dispatcher).ConnectedPeer()
+	assert.Equal(t, org2MSP, connectedPeer.MSPID())
+
+	outcomech := make(chan mockconn.Outcome)
+	go listenConnection(connectch, outcomech)
+
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx1", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx2", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx3", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx4", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx5", pb.TxValidationCode_VALID))
+
+	// Set the block height of the local peer to be greater than the disconnect threshold
+	// so that the event client can reconnect to the local peer
+	p2O1.SetBlockHeight(9)
+
+	select {
+	case outcome := <-outcomech:
+		assert.Equal(t, mockconn.ReconnectedOutcome, outcome)
+		connectedPeer := eventClient.Dispatcher().(*dispatcher.Dispatcher).ConnectedPeer()
+		assert.Equal(t, org1MSP, connectedPeer.MSPID())
+	case <-time.After(3 * time.Second):
+		t.Fatal("Timed out waiting for reconnect")
+	}
+}
+
+// TestPreferLocalPeersConnection tests the scenario where an org wishes to connect to one of a list of preferred peers
+// if they are above the block height lag threshold but, if they fall below the threshold, the
+// connection should be made to another peer. Once the preferred peers have caught up in
+// block height, the connection to one of the preferred peers should be re-established.
+func TestPreferLocalPeersConnection(t *testing.T) {
+	channelID := "testchannel"
+	org1MSP := "Org1MSP"
+	org2MSP := "Org2MSP"
+	blockHeightLagThreshold := 2
+
+	p1O1 := clientmocks.NewMockStatefulPeer("p1_o1", "peer1.org1.com:7051", clientmocks.WithMSP(org1MSP), clientmocks.WithBlockHeight(4))
+	p2O1 := clientmocks.NewMockStatefulPeer("p2_o1", "peer2.org1.com:7051", clientmocks.WithMSP(org1MSP), clientmocks.WithBlockHeight(3))
+	p1O2 := clientmocks.NewMockStatefulPeer("p1_o2", "peer1.org2.com:7051", clientmocks.WithMSP(org2MSP), clientmocks.WithBlockHeight(10))
+	p2O2 := clientmocks.NewMockStatefulPeer("p2_o2", "peer1.org2.com:7051", clientmocks.WithMSP(org2MSP), clientmocks.WithBlockHeight(11))
+
+	connectch := make(chan *dispatcher.ConnectionEvent)
+
+	conn := clientmocks.NewMockConnection(
+		clientmocks.WithLedger(servicemocks.NewMockLedger(servicemocks.FilteredBlockEventFactory, sourceURL)),
+	)
+	connectionProvider := clientmocks.NewProviderFactory().Provider(conn)
+
+	eventClient, _, err := newClientWithMockConnAndOpts(
+		fabmocks.NewMockContext(
+			mspmocks.NewMockSigningIdentity("user1", "Org1MSP"),
+		),
+		fabmocks.NewMockChannelCfg(channelID),
+		clientmocks.NewDiscoveryService(p1O1, p2O1, p1O2, p2O2),
+		connectionProvider, filteredClientProvider,
+		[]options.Opt{
+			esdispatcher.WithEventConsumerTimeout(3 * time.Second),
+			WithMaxConnectAttempts(1),
+			WithTimeBetweenConnectAttempts(time.Millisecond),
+			WithConnectionEvent(connectch),
+			WithResponseTimeout(2 * time.Second),
+			dispatcher.WithPeerResolver(preferpeer.NewResolver(p1O1.URL(), p2O1.URL())),
+			dispatcher.WithLoadBalancePolicy(lbp.NewRoundRobin()),
+			dispatcher.WithPeerMonitorPeriod(250 * time.Millisecond),
+			minblockheight.WithBlockHeightLagThreshold(blockHeightLagThreshold),
+			minblockheight.WithReconnectBlockHeightThreshold(3),
+		},
+	)
+	require.NoErrorf(t, err, "error creating channel event client")
+	err = eventClient.Connect()
+	require.NoErrorf(t, err, "errorconnecting channel event client")
+	defer eventClient.Close()
+
+	connectedPeer := eventClient.Dispatcher().(*dispatcher.Dispatcher).ConnectedPeer()
+	assert.Equal(t, org2MSP, connectedPeer.MSPID())
+
+	outcomech := make(chan mockconn.Outcome)
+	go listenConnection(connectch, outcomech)
+
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx1", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx2", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx3", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx4", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx5", pb.TxValidationCode_VALID))
+
+	// Set the block height of the local peer to be greater than the disconnect threshold
+	// so that the event client can reconnect to the local peer
+	p2O1.SetBlockHeight(9)
+
+	select {
+	case outcome := <-outcomech:
+		assert.Equal(t, mockconn.ReconnectedOutcome, outcome)
+		connectedPeer := eventClient.Dispatcher().(*dispatcher.Dispatcher).ConnectedPeer()
+		assert.Equal(t, org1MSP, connectedPeer.MSPID())
+	case <-time.After(3 * time.Second):
+		t.Fatal("Timed out waiting for reconnect")
+	}
+}
+
 func TestTransferRegistrations(t *testing.T) {
 	// Tests the scenario where all event registrations are transferred to another event client.
 	t.Run("Transfer", func(t *testing.T) {

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

@@ -242,7 +242,7 @@ func (ed *Dispatcher) clearConnectionRegistration() {
 }
 
 func (ed *Dispatcher) monitorPeer(done chan struct{}) {
-	logger.Infof("Starting peer monitor on channel [%s]", ed.chConfig.ID())
+	logger.Debugf("Starting peer monitor on channel [%s]", ed.chConfig.ID())
 
 	ticker := time.NewTicker(ed.peerMonitorPeriod)
 	defer ticker.Stop()

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

@@ -14,6 +14,7 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/lbp"
 	clientmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/mocks"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/peerresolver/minblockheight"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/peerresolver/preferorg"
 	esdispatcher "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service/dispatcher"
 	servicemocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service/mocks"
 	fabmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
@@ -293,7 +294,6 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
 	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
 
-	time.Sleep(time.Second)
 	p2.SetBlockHeight(15)
 
 	select {
@@ -304,6 +304,98 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 	}
 }
 
+// TestPreferLocalOrgConnection tests the scenario where an org wishes to connect to it's own peers
+// if they are above the block height lag threshold but, if they fall below the threshold, the
+// connection should be made to another org's peer. Once the local org's peers have caught up in
+// block height, the connection to the other org's peer should be terminated.
+func TestPreferLocalOrgConnection(t *testing.T) {
+	channelID := "testchannel"
+	org1MSP := "Org1MSP"
+	org2MSP := "Org2MSP"
+
+	p1O1 := clientmocks.NewMockStatefulPeer("p1_o1", "peer1.org1.com:7051", clientmocks.WithMSP(org1MSP), clientmocks.WithBlockHeight(4))
+	p2O1 := clientmocks.NewMockStatefulPeer("p2_o1", "peer2.org1.com:7051", clientmocks.WithMSP(org1MSP), clientmocks.WithBlockHeight(3))
+	p1O2 := clientmocks.NewMockStatefulPeer("p1_o2", "peer1.org2.com:7051", clientmocks.WithMSP(org2MSP), clientmocks.WithBlockHeight(10))
+	p2O2 := clientmocks.NewMockStatefulPeer("p2_o2", "peer1.org2.com:7051", clientmocks.WithMSP(org2MSP), clientmocks.WithBlockHeight(12))
+
+	dispatcher := New(
+		fabmocks.NewMockContext(
+			mspmocks.NewMockSigningIdentity("user1", org1MSP),
+		),
+		fabmocks.NewMockChannelCfg(channelID),
+		clientmocks.NewDiscoveryService(p1O1, p2O1, p1O2, p2O2),
+		clientmocks.NewProviderFactory().Provider(
+			clientmocks.NewMockConnection(
+				clientmocks.WithLedger(
+					servicemocks.NewMockLedger(servicemocks.FilteredBlockEventFactory, sourceURL),
+				),
+			),
+		),
+		WithPeerResolver(preferorg.NewResolver()),
+		WithPeerMonitorPeriod(250*time.Millisecond),
+		minblockheight.WithBlockHeightLagThreshold(2),
+		minblockheight.WithReconnectBlockHeightThreshold(3),
+		WithLoadBalancePolicy(lbp.NewRandom()),
+	)
+
+	if err := dispatcher.Start(); err != nil {
+		t.Fatalf("Error starting dispatcher: %s", err)
+	}
+
+	dispatcherEventch, err := dispatcher.EventCh()
+	if err != nil {
+		t.Fatalf("Error getting event channel from dispatcher: %s", err)
+	}
+
+	// Register for connection events
+	regerrch := make(chan error)
+	regch := make(chan fab.Registration)
+	connch := make(chan *ConnectionEvent, 10)
+	dispatcherEventch <- NewRegisterConnectionEvent(connch, regch, regerrch)
+
+	select {
+	case <-regch:
+		// No need get the registration to unregister since we're relying on the
+		// connch channel being closed when the dispatcher is stopped.
+	case err := <-regerrch:
+		t.Fatalf("Error registering for connection events: %s", err)
+	}
+
+	// Connect
+	errch := make(chan error)
+	dispatcherEventch <- NewConnectEvent(errch)
+	err = <-errch
+	if err != nil {
+		t.Fatalf("Error connecting: %s", err)
+	}
+
+	dispatcherEventch <- NewConnectedEvent()
+
+	select {
+	case e := <-connch:
+		assert.Truef(t, e.Connected, "expecting connected event")
+	case <-time.After(time.Second):
+		t.Fatal("Expecting connected event but got none")
+	}
+
+	// The initial connection should have been to an Org2 peer since their block heights are higher than Org1
+	blockProducer := servicemocks.NewBlockProducer()
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+
+	p2O1.SetBlockHeight(15)
+
+	select {
+	case e := <-connch:
+		assert.Falsef(t, e.Connected, "expecting disconnected event")
+	case <-time.After(time.Second):
+		t.Fatal("Expecting disconnected event but got none")
+	}
+}
+
 func checkEvent(connch chan *ConnectionEvent, errch chan error, state, expectedDisconnectErr string) {
 	for {
 		select {

pkg/fab/events/client/mocks/mockpeer.go

@@ -9,21 +9,49 @@ package mocks
 import (
 	"sync"
 
-	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	fabmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
 )
 
 // MockPeer contains mock PeerState
 type MockPeer struct {
-	fab.Peer
+	*fabmocks.MockPeer
 	blockHeight uint64
 	lock        sync.RWMutex
 }
 
+// MockPeerOpt is a mock peer option
+type MockPeerOpt func(*MockPeer)
+
+// WithMSP sets the MSP ID of the mock peer
+func WithMSP(mspID string) MockPeerOpt {
+	return func(p *MockPeer) {
+		p.SetMSPID(mspID)
+	}
+}
+
+// WithBlockHeight sets the block height of the mock peer
+func WithBlockHeight(blockHeight uint64) MockPeerOpt {
+	return func(p *MockPeer) {
+		p.blockHeight = blockHeight
+	}
+}
+
+// NewMockStatefulPeer returns a new MockPeer with the given options
+func NewMockStatefulPeer(name, url string, opts ...MockPeerOpt) *MockPeer {
+	p := &MockPeer{
+		MockPeer: fabmocks.NewMockPeer(name, url),
+	}
+	for _, opt := range opts {
+		opt(p)
+	}
+	return p
+}
+
 // NewMockPeer returns a new MockPeer
+// Deprecated: This function will be deprecated in the future. Use NewMockStatefulPeer instead.
 func NewMockPeer(name, url string, blockHeight uint64) *MockPeer {
 	return &MockPeer{
-		Peer:        fabmocks.NewMockPeer(name, url),
+		MockPeer:    fabmocks.NewMockPeer(name, url),
 		blockHeight: blockHeight,
 	}
 }

pkg/fab/events/client/peerresolver/preferorg/opts.go

@@ -0,0 +1,27 @@
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package preferorg
+
+import (
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/lbp"
+)
+
+type params struct {
+	loadBalancePolicy lbp.LoadBalancePolicy
+}
+
+func defaultParams(context context.Client) *params {
+	return &params{
+		loadBalancePolicy: lbp.NewRandom(),
+	}
+}
+
+func (p *params) SetLoadBalancePolicy(value lbp.LoadBalancePolicy) {
+	logger.Debugf("LoadBalancePolicy: %#v", value)
+	p.loadBalancePolicy = value
+}