[FAB-8813] Pass GRPC options to event client

Ensure that all GRPC options are passed to the
event clients.

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

Author: bstasyszyn

pkg/core/config/config.go

@@ -919,18 +919,28 @@ func (c *Config) PeerConfigByURL(url string) (*core.PeerConfig, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	var matchPeerConfig *core.PeerConfig
 	staticPeers := config.Peers
 	for _, staticPeerConfig := range staticPeers {
 		if strings.EqualFold(staticPeerConfig.URL, url) {
-			return &staticPeerConfig, nil
+			matchPeerConfig = &staticPeerConfig
+			break
 		}
 	}
 
-	// try to match from entity matchers
-	matchPeerConfig, err := c.tryMatchingPeerConfig(url)
-	if err != nil {
-		return nil, errors.WithMessage(err, "No Peer found with the url from config")
+	if matchPeerConfig == nil {
+		// try to match from entity matchers
+		matchPeerConfig, err = c.tryMatchingPeerConfig(url)
+		if err != nil {
+			return nil, errors.WithMessage(err, "No Peer found with the url from config")
+		}
+	}
+
+	if matchPeerConfig != nil && matchPeerConfig.TLSCACerts.Path != "" {
+		matchPeerConfig.TLSCACerts.Path = SubstPathVars(matchPeerConfig.TLSCACerts.Path)
 	}
+
 	return matchPeerConfig, nil
 }
 

pkg/fab/events/api/endpoint.go

@@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0
 package api
 
 import (
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/options"
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/fab"
 )
 
@@ -17,4 +18,7 @@ type EventEndpoint interface {
 
 	// EventURL returns the event URL
 	EventURL() string
+
+	// Opts returns additional options for the connection
+	Opts() []options.Opt
 }

pkg/fab/events/deliverclient/deliverclient.go

@@ -19,6 +19,7 @@ import (
 	deliverconn "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/connection"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/dispatcher"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/seek"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/endpoint"
 	"github.com/hyperledger/fabric-sdk-go/pkg/logging"
 	"github.com/pkg/errors"
 )
@@ -27,12 +28,20 @@ var logger = logging.NewLogger("fabsdk/fab")
 
 // deliverProvider is the connection provider used for connecting to the Deliver service
 var deliverProvider = func(context fabcontext.Client, chConfig fab.ChannelCfg, peer fab.Peer) (api.Connection, error) {
-	return deliverconn.New(context, chConfig, deliverconn.Deliver, peer.URL())
+	eventEndpoint, ok := peer.(api.EventEndpoint)
+	if !ok {
+		panic("peer is not an EventEndpoint")
+	}
+	return deliverconn.New(context, chConfig, deliverconn.Deliver, peer.URL(), eventEndpoint.Opts()...)
 }
 
 // deliverFilteredProvider is the connection provider used for connecting to the DeliverFiltered service
 var deliverFilteredProvider = func(context fabcontext.Client, chConfig fab.ChannelCfg, peer fab.Peer) (api.Connection, error) {
-	return deliverconn.New(context, chConfig, deliverconn.DeliverFiltered, peer.URL())
+	eventEndpoint, ok := peer.(api.EventEndpoint)
+	if !ok {
+		panic("peer is not an EventEndpoint")
+	}
+	return deliverconn.New(context, chConfig, deliverconn.DeliverFiltered, peer.URL(), eventEndpoint.Opts()...)
 }
 
 // Client connects to a peer and receives channel events, such as bock, filtered block, chaincode, and transaction status events.
@@ -46,10 +55,14 @@ func New(context fabcontext.Client, chConfig fab.ChannelCfg, opts ...options.Opt
 	params := defaultParams()
 	options.Apply(params, opts)
 
+	// Use a context that returns a custom Discovery Provider which
+	// produces event endpoints containing additional GRPC options.
+	deliverCtx := newDeliverContext(context)
+
 	client := &Client{
 		Client: *client.New(
 			params.permitBlockEvents,
-			dispatcher.New(context, chConfig, params.connProvider, opts...),
+			dispatcher.New(deliverCtx, chConfig, params.connProvider, opts...),
 			opts...,
 		),
 		params: *params,
@@ -121,3 +134,20 @@ func (c *Client) seekInfo() (*ab.SeekInfo, error) {
 		return nil, errors.Errorf("unsupported seek type:[%s]", c.seekType)
 	}
 }
+
+// deliverContext overrides the DiscoveryProvider
+type deliverContext struct {
+	fabcontext.Client
+}
+
+func newDeliverContext(ctx fabcontext.Client) fabcontext.Client {
+	return &deliverContext{
+		Client: ctx,
+	}
+}
+
+// DiscoveryProvider returns a custom discovery provider which produces
+// event endpoints with additional GRPC options
+func (ctx *deliverContext) DiscoveryProvider() fab.DiscoveryProvider {
+	return endpoint.NewDiscoveryProvider(ctx.Client)
+}

pkg/fab/events/deliverclient/deliverclient_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/options"
+	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/core"
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client"
 	clientdisp "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client/dispatcher"
@@ -33,17 +34,14 @@ const (
 var (
 	defaultOpts = []options.Opt{}
 
-	peer1 = fabclientmocks.NewMockPeer("peer1", "grpcs://peer1.example.com:7051")
-	peer2 = fabclientmocks.NewMockPeer("peer2", "grpcs://peer2.example.com:7051")
+	peer1 = fabclientmocks.NewMockPeer("peer1", "peer1.example.com:7051")
+	peer2 = fabclientmocks.NewMockPeer("peer2", "peer2.example.com:7051")
 )
 
 func TestOptionsInNewClient(t *testing.T) {
 	channelID := "mychannel"
 	client, err := New(
-		fabmocks.NewMockContextWithCustomDiscovery(
-			fabmocks.NewMockUser("user1"),
-			clientmocks.NewDiscoveryProvider(peer1, peer2),
-		),
+		newMockContext(),
 		fabmocks.NewMockChannelCfg(channelID),
 		WithBlockEvents(),
 	)
@@ -56,10 +54,7 @@ func TestOptionsInNewClient(t *testing.T) {
 func TestClientConnect(t *testing.T) {
 	channelID := "mychannel"
 	eventClient, err := New(
-		fabmocks.NewMockContextWithCustomDiscovery(
-			fabmocks.NewMockUser("user1"),
-			clientmocks.NewDiscoveryProvider(peer1, peer2),
-		),
+		newMockContext(),
 		fabmocks.NewMockChannelCfg(channelID),
 		withConnectionProvider(
 			clientmocks.NewProviderFactory().Provider(
@@ -187,10 +182,7 @@ func testConnect(t *testing.T, maxConnectAttempts uint, expectedOutcome clientmo
 
 	channelID := "mychannel"
 	eventClient, err := New(
-		fabmocks.NewMockContextWithCustomDiscovery(
-			fabmocks.NewMockUser("user1"),
-			clientmocks.NewDiscoveryProvider(peer1, peer2),
-		),
+		newMockContext(),
 		fabmocks.NewMockChannelCfg(channelID),
 		withConnectionProvider(
 			cp.FlakeyProvider(
@@ -230,10 +222,7 @@ func testReconnect(t *testing.T, reconnect bool, maxReconnectAttempts uint, expe
 
 	channelID := "mychannel"
 	eventClient, err := New(
-		fabmocks.NewMockContextWithCustomDiscovery(
-			fabmocks.NewMockUser("user1"),
-			clientmocks.NewDiscoveryProvider(peer1, peer2),
-		),
+		newMockContext(),
 		fabmocks.NewMockChannelCfg(channelID),
 		withConnectionProvider(
 			cp.FlakeyProvider(
@@ -303,10 +292,7 @@ func testReconnectRegistration(t *testing.T, connectResults clientmocks.ConnectA
 	cp := clientmocks.NewProviderFactory()
 
 	eventClient, err := New(
-		fabmocks.NewMockContextWithCustomDiscovery(
-			fabmocks.NewMockUser("user1"),
-			clientmocks.NewDiscoveryProvider(peer1, peer2),
-		),
+		newMockContext(),
 		fabmocks.NewMockChannelCfg(channelID),
 		withConnectionProvider(
 			cp.FlakeyProvider(
@@ -451,3 +437,26 @@ func listenEvents(blockch <-chan *fab.BlockEvent, ccch <-chan *fab.CCEvent, wait
 		}
 	}
 }
+
+type mockConfig struct {
+	core.Config
+}
+
+func newMockConfig() *mockConfig {
+	return &mockConfig{
+		Config: fabmocks.NewMockConfig(),
+	}
+}
+
+func (c *mockConfig) PeerConfigByURL(url string) (*core.PeerConfig, error) {
+	return &core.PeerConfig{}, nil
+}
+
+func newMockContext() *fabmocks.MockContext {
+	ctx := fabmocks.NewMockContextWithCustomDiscovery(
+		fabmocks.NewMockUser("user1"),
+		clientmocks.NewDiscoveryProvider(peer1, peer2),
+	)
+	ctx.SetConfig(newMockConfig())
+	return ctx
+}

pkg/fab/events/endpoint/endpoint.go

@@ -10,10 +10,12 @@ import (
 	"crypto/x509"
 	"time"
 
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/options"
+	"github.com/hyperledger/fabric-sdk-go/pkg/errors/status"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/comm"
+
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/core"
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/fab"
-	"github.com/hyperledger/fabric-sdk-go/pkg/errors/status"
-	"github.com/hyperledger/fabric-sdk-go/pkg/fab/peer"
 	"github.com/spf13/cast"
 	"google.golang.org/grpc/keepalive"
 )
@@ -29,20 +31,31 @@ type EventEndpoint struct {
 	KeepAliveParams keepalive.ClientParameters
 	FailFast        bool
 	ConnectTimeout  time.Duration
+	AllowInsecure   bool
 }
 
 // EventURL returns the event URL
 func (e *EventEndpoint) EventURL() string {
 	return e.EvtURL
 }
 
-// FromPeerConfig creates a new EventEndpoint from the given config
-func FromPeerConfig(config core.Config, peerCfg core.NetworkPeer) (*EventEndpoint, error) {
-	p, err := peer.New(config, peer.FromPeerConfig(&peerCfg))
-	if err != nil {
-		return nil, err
+// Opts returns additional options for the event connection
+func (e *EventEndpoint) Opts() []options.Opt {
+	opts := []options.Opt{
+		comm.WithHostOverride(e.HostOverride),
+		comm.WithFailFast(e.FailFast),
+		comm.WithKeepAliveParams(e.KeepAliveParams),
+		comm.WithCertificate(e.Certificate),
+		comm.WithConnectTimeout(e.ConnectTimeout),
+	}
+	if e.AllowInsecure {
+		opts = append(opts, comm.WithInsecure())
 	}
+	return opts
+}
 
+// FromPeerConfig creates a new EventEndpoint from the given config
+func FromPeerConfig(config core.Config, peer fab.Peer, peerCfg *core.PeerConfig) (*EventEndpoint, error) {
 	certificate, err := peerCfg.TLSCACerts.TLSCert()
 	if err != nil {
 		//Ignore empty cert errors,
@@ -53,31 +66,32 @@ func FromPeerConfig(config core.Config, peerCfg core.NetworkPeer) (*EventEndpoin
 	}
 
 	return &EventEndpoint{
-		Peer:            p,
+		Peer:            peer,
 		EvtURL:          peerCfg.EventURL,
 		HostOverride:    getServerNameOverride(peerCfg),
 		Certificate:     certificate,
 		KeepAliveParams: getKeepAliveOptions(peerCfg),
 		FailFast:        getFailFast(peerCfg),
 		ConnectTimeout:  config.TimeoutOrDefault(core.EventHubConnection),
+		AllowInsecure:   isInsecureAllowed(peerCfg),
 	}, nil
 }
 
-func getServerNameOverride(peerCfg core.NetworkPeer) string {
+func getServerNameOverride(peerCfg *core.PeerConfig) string {
 	if str, ok := peerCfg.GRPCOptions["ssl-target-name-override"].(string); ok {
 		return str
 	}
 	return ""
 }
 
-func getFailFast(peerCfg core.NetworkPeer) bool {
+func getFailFast(peerCfg *core.PeerConfig) bool {
 	if ff, ok := peerCfg.GRPCOptions["fail-fast"].(bool); ok {
 		return cast.ToBool(ff)
 	}
 	return false
 }
 
-func getKeepAliveOptions(peerCfg core.NetworkPeer) keepalive.ClientParameters {
+func getKeepAliveOptions(peerCfg *core.PeerConfig) keepalive.ClientParameters {
 	var kap keepalive.ClientParameters
 	if kaTime, ok := peerCfg.GRPCOptions["keep-alive-time"]; ok {
 		kap.Time = cast.ToDuration(kaTime)
@@ -90,3 +104,11 @@ func getKeepAliveOptions(peerCfg core.NetworkPeer) keepalive.ClientParameters {
 	}
 	return kap
 }
+
+func isInsecureAllowed(peerCfg *core.PeerConfig) bool {
+	allowInsecure, ok := peerCfg.GRPCOptions["allow-insecure"].(bool)
+	if ok {
+		return allowInsecure
+	}
+	return false
+}