[FAB-8995] Event client should read channel config

The event endpoint discovery service now reads the
channel peer config. If a channel peer config is
present for a given peer URL and eventSource==false
then the peer will be excluded.

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

Author: bstasyszyn

pkg/fab/events/deliverclient/connection/connection.go

@@ -59,6 +59,7 @@ var (
 
 // New returns a new Deliver Server connection
 func New(ctx fabcontext.Client, chConfig fab.ChannelCfg, streamProvider StreamProvider, url string, opts ...options.Opt) (*DeliverConnection, error) {
+	logger.Debugf("Connecting to %s...", url)
 	connect, err := comm.NewConnection(
 		ctx, chConfig,
 		func(grpcconn *grpc.ClientConn) (grpc.ClientStream, error) {

pkg/fab/events/endpoint/endpoint_test.go

@@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0
 package endpoint
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -16,6 +17,23 @@ import (
 	mspmocks "github.com/hyperledger/fabric-sdk-go/pkg/msp/test/mockmsp"
 )
 
+const (
+	url1 = "p1.test.com:9051"
+	url2 = "p2.test.com:9051"
+	url3 = "p3.test.com:9051"
+)
+
+var p1 = fabmocks.NewMockPeer("p1", url1)
+var p2 = fabmocks.NewMockPeer("p2", url2)
+var p3 = fabmocks.NewMockPeer("p3", url3)
+
+var pc1 = core.PeerConfig{URL: url1}
+var pc2 = core.PeerConfig{URL: url2}
+var pc3 = core.PeerConfig{URL: url3}
+
+var peers = []fab.Peer{p1, p2, p3}
+var peerConfigs = []core.PeerConfig{pc1, pc2, pc3}
+
 func TestEndpoint(t *testing.T) {
 	expectedEventURL := "localhost:7053"
 	expectedAllowInsecure := true
@@ -69,71 +87,119 @@ func TestEndpoint(t *testing.T) {
 
 func TestDiscoveryProvider(t *testing.T) {
 	ctx := newMockContext()
+
+	expectedNumPeers := len(peers)
+
 	discoveryProvider := NewDiscoveryProvider(ctx)
 
 	discoveryService, err := discoveryProvider.CreateDiscoveryService("testchannel")
 	if err != nil {
 		t.Fatalf("error creating discovery service: %s", err)
 	}
-	_, err = discoveryService.GetPeers()
+	peers, err = discoveryService.GetPeers()
 	if err != nil {
 		t.Fatalf("error getting peers: %s", err)
 	}
-
+	if len(peers) != expectedNumPeers {
+		t.Fatalf("expecting %d peers but got %d", expectedNumPeers, len(peers))
+	}
 }
 
 func TestDiscoveryProviderWithTargetFilter(t *testing.T) {
 	ctx := newMockContext()
 
-	var numTimesCalled int
-	expectedNumTimesCalled := 1
+	expectedNumPeers := len(peers) - 1
+
+	discoveryProvider := NewDiscoveryProvider(ctx, WithTargetFilter(newMockFilter(p3)))
+
+	discoveryService, err := discoveryProvider.CreateDiscoveryService("testchannel")
+	if err != nil {
+		t.Fatalf("error creating discovery service: %s", err)
+	}
+	peers, err = discoveryService.GetPeers()
+	if err != nil {
+		t.Fatalf("error getting peers: %s", err)
+	}
+	if len(peers) != expectedNumPeers {
+		t.Fatalf("expecting %d peers but got %d", expectedNumPeers, len(peers))
+	}
+}
+
+func TestDiscoveryProviderWithEventSource(t *testing.T) {
+	ctx := newMockContext()
+
+	chPeer2 := core.ChannelPeer{}
+	chPeer2.URL = p2.URL()
+	chPeer2.EventSource = false
+	ctx.SetConfig(newMockConfig(chPeer2))
 
-	discoveryProvider := NewDiscoveryProvider(ctx, WithTargetFilter(newMockFilter(&numTimesCalled)))
+	expectedNumPeers := len(peers) - 1
+
+	discoveryProvider := NewDiscoveryProvider(ctx)
 
 	discoveryService, err := discoveryProvider.CreateDiscoveryService("testchannel")
 	if err != nil {
 		t.Fatalf("error creating discovery service: %s", err)
 	}
-	_, err = discoveryService.GetPeers()
+	peers, err = discoveryService.GetPeers()
 	if err != nil {
 		t.Fatalf("error getting peers: %s", err)
 	}
-	if numTimesCalled != expectedNumTimesCalled {
-		t.Fatalf("expecting target filter to be called %d time(s) but was called %d time(s)", expectedNumTimesCalled, numTimesCalled)
+	if len(peers) != expectedNumPeers {
+		t.Fatalf("expecting %d peers but got %d", expectedNumPeers, len(peers))
 	}
 }
 
 type mockConfig struct {
 	core.Config
+	channelPeers []core.ChannelPeer
 }
 
-func newMockConfig() *mockConfig {
+func newMockConfig(channelPeers ...core.ChannelPeer) *mockConfig {
 	return &mockConfig{
-		Config: fabmocks.NewMockConfig(),
+		Config:       fabmocks.NewMockConfig(),
+		channelPeers: channelPeers,
 	}
 }
 
 func (c *mockConfig) PeerConfigByURL(url string) (*core.PeerConfig, error) {
-	return &core.PeerConfig{}, nil
+	for _, pc := range peerConfigs {
+		if pc.URL == url {
+			return &pc, nil
+		}
+	}
+	return nil, nil
+}
+
+func (c *mockConfig) ChannelPeers(name string) ([]core.ChannelPeer, error) {
+	fmt.Printf("mockConfig.ChannelPeers - returning %#v", c.channelPeers)
+	return c.channelPeers, nil
 }
 
 func newMockContext() *fabmocks.MockContext {
-	ctx := fabmocks.NewMockContext(
+	discoveryProvider, _ := fabmocks.NewMockDiscoveryProvider(nil, peers)
+
+	ctx := fabmocks.NewMockContextWithCustomDiscovery(
 		mspmocks.NewMockSigningIdentity("user1", "Org1MSP"),
+		discoveryProvider,
 	)
 	ctx.SetConfig(newMockConfig())
 	return ctx
 }
 
 type mockFilter struct {
-	numTimesCalled *int
+	excludePeers []fab.Peer
 }
 
-func newMockFilter(numTimesCalled *int) *mockFilter {
-	return &mockFilter{numTimesCalled: numTimesCalled}
+func newMockFilter(excludePeers ...fab.Peer) *mockFilter {
+	return &mockFilter{excludePeers: excludePeers}
 }
 
 func (f *mockFilter) Accept(peer fab.Peer) bool {
-	*f.numTimesCalled++
+	for _, p := range f.excludePeers {
+		if p.URL() == peer.URL() {
+			return false
+		}
+	}
 	return true
 }

pkg/fab/events/endpoint/endpointdiscovery.go

@@ -8,11 +8,15 @@ package endpoint
 
 import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/discovery"
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/logging"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/core"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	"github.com/pkg/errors"
 )
 
+var logger = logging.NewLogger("fabsdk/fab")
+
 // DiscoveryProvider is a wrapper around a discovery provider that
 // converts each peer into an EventEndpoint. The EventEndpoint
 // provides additional connection options.
@@ -57,15 +61,24 @@ func (p *DiscoveryProvider) CreateDiscoveryService(channelID string) (fab.Discov
 		target = discovery.NewDiscoveryFilterService(target, p.filter)
 	}
 
+	chpeers, err := p.ctx.Config().ChannelPeers(channelID)
+	if err != nil {
+		return nil, errors.Wrapf(err, "unable to get channel peers for channel [%s]", channelID)
+	}
+
 	return &discoveryService{
 		DiscoveryService: target,
 		ctx:              p.ctx,
+		channelID:        channelID,
+		chPeers:          chpeers,
 	}, nil
 }
 
 type discoveryService struct {
 	fab.DiscoveryService
-	ctx context.Client
+	ctx       context.Client
+	channelID string
+	chPeers   []core.ChannelPeer
 }
 
 func (s *discoveryService) GetPeers() ([]fab.Peer, error) {
@@ -85,6 +98,15 @@ func (s *discoveryService) GetPeers() ([]fab.Peer, error) {
 			return nil, errors.Errorf("unable to get peer config from [%s]", peer.URL())
 		}
 
+		chPeer := s.getChannelPeer(peerConfig)
+
+		logger.Debugf("Channel peer config for [%s]: %#v", peer.URL(), chPeer)
+
+		if chPeer != nil && !chPeer.EventSource {
+			logger.Debugf("Excluding peer [%s] since it is not configured as an event source", peer.URL())
+			continue
+		}
+
 		eventEndpoint, err := FromPeerConfig(s.ctx.Config(), peer, peerConfig)
 		if err != nil {
 			return nil, errors.Wrapf(err, "unable to create event endpoint for [%s]", peer.URL())
@@ -94,3 +116,12 @@ func (s *discoveryService) GetPeers() ([]fab.Peer, error) {
 
 	return eventEndpoints, nil
 }
+
+func (s *discoveryService) getChannelPeer(peerConfig *core.PeerConfig) *core.ChannelPeer {
+	for _, chpeer := range s.chPeers {
+		if chpeer.URL == peerConfig.URL {
+			return &chpeer
+		}
+	}
+	return nil
+}

test/fixtures/config/config_revoke_test.yaml

@@ -170,7 +170,9 @@ channels:
         endorsingPeer: true
         chaincodeQuery: true
         ledgerQuery: true
-        eventSource: true
+        # Don't use revoked peer as event source or
+        # else the test will time out waiting for event
+        eventSource: false
 
 
 #