[FAB-9445] Endpoint Options: cc query, endorsing peer

Change-Id: If4a6f960b794bfb68f10636628fa0285726da074
Signed-off-by: Sandra Vrtikapa <sandra.vrtikapa@securekey.com>

Author: sandrask

pkg/client/channel/chclient.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/channel/invoke"
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/discovery/greylist"
+	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/filter"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/retry"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/status"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
@@ -79,22 +80,39 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 
 // Query chaincode using request and optional options provided
 func (cc *Client) Query(request Request, options ...RequestOption) (Response, error) {
-	optsWithTimeout, err := cc.addDefaultTimeout(cc.context, fab.Query, options...)
-	if err != nil {
-		return Response{}, errors.WithMessage(err, "option failed")
-	}
 
-	return cc.InvokeHandler(invoke.NewQueryHandler(), request, optsWithTimeout...)
+	options = append(options, addDefaultTimeout(fab.Query))
+	options = append(options, addDefaultTargetFilter(cc.context, filter.ChaincodeQuery))
+
+	return cc.InvokeHandler(invoke.NewQueryHandler(), request, options...)
 }
 
 // Execute prepares and executes transaction using request and optional options provided
 func (cc *Client) Execute(request Request, options ...RequestOption) (Response, error) {
-	optsWithTimeout, err := cc.addDefaultTimeout(cc.context, fab.Execute, options...)
-	if err != nil {
-		return Response{}, errors.WithMessage(err, "option failed")
+	options = append(options, addDefaultTimeout(fab.Execute))
+	options = append(options, addDefaultTargetFilter(cc.context, filter.EndorsingPeer))
+
+	return cc.InvokeHandler(invoke.NewExecuteHandler(), request, options...)
+}
+
+// addDefaultTargetFilter adds default target filter if target filter is not specified
+func addDefaultTargetFilter(chCtx context.Channel, ft filter.EndpointType) RequestOption {
+	return func(ctx context.Client, o *requestOptions) error {
+		if len(o.Targets) == 0 && o.TargetFilter == nil {
+			return WithTargetFilter(filter.NewEndpointFilter(chCtx, ft))(ctx, o)
+		}
+		return nil
 	}
+}
 
-	return cc.InvokeHandler(invoke.NewExecuteHandler(), request, optsWithTimeout...)
+// addDefaultTimeout adds default timeout if timeout is not specified
+func addDefaultTimeout(tt fab.TimeoutType) RequestOption {
+	return func(ctx context.Client, o *requestOptions) error {
+		if o.Timeouts[tt] == 0 {
+			return WithTimeout(tt, ctx.EndpointConfig().Timeout(tt))(ctx, o)
+		}
+		return nil
+	}
 }
 
 //InvokeHandler invokes handler using request and options provided
@@ -224,23 +242,6 @@ func (cc *Client) prepareOptsFromOptions(ctx context.Client, options ...RequestO
 	return txnOpts, nil
 }
 
-//addDefaultTimeout adds given default timeout if it is missing in options
-func (cc *Client) addDefaultTimeout(ctx context.Client, timeOutType fab.TimeoutType, options ...RequestOption) ([]RequestOption, error) {
-	txnOpts := requestOptions{}
-	for _, option := range options {
-		err := option(ctx, &txnOpts)
-		if err != nil {
-			return nil, errors.WithMessage(err, "option failed")
-		}
-	}
-
-	if txnOpts.Timeouts[timeOutType] == 0 {
-		//InvokeHandler relies on Execute timeout
-		return append(options, WithTimeout(fab.Execute, cc.context.EndpointConfig().Timeout(timeOutType))), nil
-	}
-	return options, nil
-}
-
 // RegisterChaincodeEvent registers chain code event
 // @param {chan bool} channel which receives event details when the event is complete
 // @returns {object} object handle that should be used to unregister

pkg/client/common/filter/endpoint.go

@@ -0,0 +1,82 @@
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+// Package filter provides common filters (e.g. Endpoint)
+package filter
+
+import (
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
+)
+
+// EndpointType represents endpoint type
+type EndpointType int32
+
+// Endpoint types
+const (
+	ChaincodeQuery EndpointType = iota
+	EndorsingPeer
+	LedgerQuery
+	EventSource
+)
+
+// NewEndpointFilter creates a new endpoint filter that is based on configuration.
+// If channel peer is not configured it will be selected by default.
+func NewEndpointFilter(ctx context.Channel, et EndpointType) *EndpointFilter {
+
+	// Retrieve channel peers
+	chPeers, err := ctx.EndpointConfig().ChannelPeers(ctx.ChannelID())
+	if err != nil {
+		// Setting channel peers to empty due to config error, should not happen
+		chPeers = []fab.ChannelPeer{}
+	}
+
+	return &EndpointFilter{endpointType: et, ctx: ctx, chPeers: chPeers}
+
+}
+
+// EndpointFilter filters based on endpoint config options
+type EndpointFilter struct {
+	endpointType EndpointType
+	ctx          context.Channel
+	chPeers      []fab.ChannelPeer // configured channel peers
+}
+
+// Accept returns false if this peer is to be excluded from the target list
+func (f *EndpointFilter) Accept(peer fab.Peer) bool {
+
+	peerConfig, err := f.ctx.EndpointConfig().PeerConfigByURL(peer.URL())
+	if err != nil || peerConfig == nil {
+		return true
+	}
+
+	chPeer := f.getChannelPeer(peerConfig)
+	if chPeer == nil {
+		return true
+	}
+
+	switch t := f.endpointType; t {
+	case ChaincodeQuery:
+		return chPeer.ChaincodeQuery
+	case EndorsingPeer:
+		return chPeer.EndorsingPeer
+	case LedgerQuery:
+		return chPeer.LedgerQuery
+	case EventSource:
+		return chPeer.EventSource
+	}
+
+	return true
+}
+
+func (f *EndpointFilter) getChannelPeer(peerConfig *fab.PeerConfig) *fab.ChannelPeer {
+	for _, chpeer := range f.chPeers {
+		if chpeer.URL == peerConfig.URL {
+			return &chpeer
+		}
+	}
+	return nil
+}

pkg/client/common/filter/endpoint_test.go

@@ -0,0 +1,117 @@
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+package filter
+
+import (
+	"testing"
+
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
+)
+
+const channelID = "mychannel"
+
+func TestInvalidOpt(t *testing.T) {
+
+	channel, err := mocks.NewMockChannel(channelID)
+	if err != nil {
+		t.Fatalf("Failed to create mock channel: %s", err)
+	}
+
+	ef := NewEndpointFilter(channel, 10)
+
+	peer := mocks.NewMockPeer("Peer1", "example.com")
+	if !ef.Accept(peer) {
+		t.Fatalf("Should have accepted peer")
+	}
+
+}
+
+func TestChaincodeQueryFilter(t *testing.T) {
+
+	channel, err := mocks.NewMockChannel(channelID)
+	if err != nil {
+		t.Fatalf("Failed to create mock channel: %s", err)
+	}
+
+	ef := NewEndpointFilter(channel, ChaincodeQuery)
+
+	if !ef.Accept(mocks.NewMockPeer("Peer1", "non-configured.com")) {
+		t.Fatalf("Should have accepted peer that is not configured")
+	}
+
+	// Configured peer
+	peer := mocks.NewMockPeer("Peer1", "example.com")
+	if !ef.Accept(peer) {
+		t.Fatalf("Should have accepted peer")
+	}
+
+	channel, err = mocks.NewMockChannel("noEndpoints")
+	ef = NewEndpointFilter(channel, ChaincodeQuery)
+	if err != nil {
+		t.Fatalf("Failed to create mock channel: %s", err)
+	}
+	if ef.Accept(peer) {
+		t.Fatalf("Should NOT have accepted peer since peers chaincode query option is configured to false")
+	}
+
+	channel, err = mocks.NewMockChannel("noChannelPeers")
+	ef = NewEndpointFilter(channel, ChaincodeQuery)
+	if err != nil {
+		t.Fatalf("Failed to create mock channel: %s", err)
+	}
+	if !ef.Accept(peer) {
+		t.Fatalf("Should have accepted peer since no peers configured")
+	}
+
+}
+
+func TestLedgerQueryFilter(t *testing.T) {
+
+	channel, err := mocks.NewMockChannel(channelID)
+	if err != nil {
+		t.Fatalf("Failed to create mock channel: %s", err)
+	}
+
+	ef := NewEndpointFilter(channel, LedgerQuery)
+
+	peer := mocks.NewMockPeer("Peer1", "example.com")
+	if !ef.Accept(peer) {
+		t.Fatalf("Should have accepted peer")
+	}
+
+}
+
+func TestEndorsingPeerFilter(t *testing.T) {
+
+	channel, err := mocks.NewMockChannel(channelID)
+	if err != nil {
+		t.Fatalf("Failed to create mock channel: %s", err)
+	}
+
+	ef := NewEndpointFilter(channel, EndorsingPeer)
+
+	peer := mocks.NewMockPeer("Peer1", "example.com")
+	if !ef.Accept(peer) {
+		t.Fatalf("Should have accepted peer")
+	}
+
+}
+
+func TestEventSourceFilter(t *testing.T) {
+
+	channel, err := mocks.NewMockChannel(channelID)
+	if err != nil {
+		t.Fatalf("Failed to create mock channel: %s", err)
+	}
+
+	ef := NewEndpointFilter(channel, EventSource)
+
+	peer := mocks.NewMockPeer("Peer1", "example.com")
+	if !ef.Accept(peer) {
+		t.Fatalf("Should have accepted peer")
+	}
+
+}

pkg/client/ledger/ledger.go

@@ -15,6 +15,7 @@ import (
 	"github.com/golang/protobuf/proto"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/discovery"
+	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/filter"
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/verifier"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/status"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
@@ -74,12 +75,7 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 		return nil, err
 	}
 
-	chpeers, err := channelContext.EndpointConfig().ChannelPeers(channelContext.ChannelID())
-	if err != nil {
-		return nil, status.New(status.ClientStatus, status.NoPeersFound.ToInt32(), err.Error(), nil)
-	}
-
-	ledgerFilter := &ledgerFilter{ctx: channelContext, chPeers: chpeers}
+	ledgerFilter := filter.NewEndpointFilter(channelContext, filter.LedgerQuery)
 
 	// Apply filter to discovery service
 	discovery := discovery.NewDiscoveryFilterService(channelContext.DiscoveryService(), ledgerFilter)
@@ -420,34 +416,3 @@ func shuffle(a []fab.Peer) {
 		a[i], a[j] = a[j], a[i]
 	}
 }
-
-// ledgerFilter filters based on ledgerQuery config option
-type ledgerFilter struct {
-	ctx     context.Channel
-	chPeers []fab.ChannelPeer // configured channel peers
-}
-
-// Accept returns false if this peer is to be excluded from the target list
-func (f *ledgerFilter) Accept(peer fab.Peer) bool {
-
-	peerConfig, err := f.ctx.EndpointConfig().PeerConfigByURL(peer.URL())
-	if err != nil || peerConfig == nil {
-		return true
-	}
-
-	chPeer := f.getChannelPeer(peerConfig)
-	if chPeer == nil {
-		return true
-	}
-
-	return chPeer.LedgerQuery
-}
-
-func (f *ledgerFilter) getChannelPeer(peerConfig *fab.PeerConfig) *fab.ChannelPeer {
-	for _, chpeer := range f.chPeers {
-		if chpeer.URL == peerConfig.URL {
-			return &chpeer
-		}
-	}
-	return nil
-}

pkg/fab/mocks/mockconfig.go

@@ -274,7 +274,18 @@ func (c *MockConfig) ChannelConfig(name string) (*fab.ChannelNetworkConfig, erro
 
 // ChannelPeers returns the channel peers configuration
 func (c *MockConfig) ChannelPeers(name string) ([]fab.ChannelPeer, error) {
-	return nil, nil
+
+	if name == "noChannelPeers" {
+		return nil, nil
+	}
+
+	peerChCfg := fab.PeerChannelConfig{EndorsingPeer: true, ChaincodeQuery: true, LedgerQuery: true, EventSource: true}
+	if name == "noEndpoints" {
+		peerChCfg = fab.PeerChannelConfig{EndorsingPeer: false, ChaincodeQuery: false, LedgerQuery: false, EventSource: false}
+	}
+
+	mockPeer := fab.ChannelPeer{PeerChannelConfig: peerChCfg, NetworkPeer: fab.NetworkPeer{PeerConfig: fab.PeerConfig{URL: "example.com"}}}
+	return []fab.ChannelPeer{mockPeer}, nil
 }
 
 // ChannelOrderers returns a list of channel orderers

test/fixtures/config/config_test_endpoints.yaml

@@ -134,7 +134,7 @@ channels:
         # [Optional]. will this peer be the target of the SDK's listener registration? All peers can
         # produce events but the app typically only needs to connect to one to listen to events.
         # Default: true
-        eventSource: false
+        eventSource: true
 
 #
 # list of participating organizations in this network