[FAB-9554] Dynamic Discovery Provider

Added a Dynamic Discovery Provider that
can be plugged into the SDK in order to
retrieve peers of a channel from Fabric's
discovery service.

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

Author: bstasyszyn

pkg/client/common/discovery/dynamicdiscovery/mocks/mockdiscoveryclient.go

@@ -0,0 +1,157 @@
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package mocks
+
+import (
+	reqcontext "context"
+	"sync"
+
+	discclient "github.com/hyperledger/fabric-sdk-go/internal/github.com/hyperledger/fabric/discovery/client"
+	"github.com/hyperledger/fabric-sdk-go/internal/github.com/hyperledger/fabric/protos/discovery"
+	"github.com/hyperledger/fabric-sdk-go/internal/github.com/hyperledger/fabric/protos/gossip"
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
+	fabdiscovery "github.com/hyperledger/fabric-sdk-go/pkg/fab/discovery"
+	discmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/discovery/mocks"
+)
+
+// MockDiscoveryClient implements a mock Discover service
+type MockDiscoveryClient struct {
+	resp []fabdiscovery.Response
+	lock sync.RWMutex
+}
+
+// MockDiscoverEndpointResponse contains a mock response for the discover client
+type MockDiscoverEndpointResponse struct {
+	Target        string
+	PeerEndpoints []*discmocks.MockDiscoveryPeerEndpoint
+	Error         error
+}
+
+// NewMockDiscoveryClient returns a new mock Discover service
+func NewMockDiscoveryClient() *MockDiscoveryClient {
+	return &MockDiscoveryClient{}
+}
+
+// Send sends a Discovery request
+func (m *MockDiscoveryClient) Send(ctx reqcontext.Context, req *discclient.Request, targets ...fab.PeerConfig) ([]fabdiscovery.Response, error) {
+	return m.responses(), nil
+}
+
+// SetResponses sets the responses that the mock client should return from the Send function
+func (m *MockDiscoveryClient) SetResponses(responses ...*MockDiscoverEndpointResponse) {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	m.resp = nil
+
+	for _, resp := range responses {
+		var peers []*discclient.Peer
+		for _, endpoint := range resp.PeerEndpoints {
+			peer := &discclient.Peer{
+				MSPID:            endpoint.MSPID,
+				AliveMessage:     newAliveMessage(endpoint),
+				StateInfoMessage: newStateInfoMessage(endpoint),
+			}
+			peers = append(peers, peer)
+		}
+		m.resp = append(m.resp, &mockDiscoverResponse{
+			Response: &response{peers: peers}, target: resp.Target, err: resp.Error,
+		})
+	}
+}
+
+func (m *MockDiscoveryClient) responses() []fabdiscovery.Response {
+	m.lock.RLock()
+	defer m.lock.RUnlock()
+	return m.resp
+}
+
+type mockDiscoverResponse struct {
+	discclient.Response
+	target string
+	err    error
+}
+
+func (r *mockDiscoverResponse) Target() string {
+	return r.target
+}
+
+func (r *mockDiscoverResponse) Error() error {
+	return r.err
+}
+
+type response struct {
+	peers []*discclient.Peer
+}
+
+func (r *response) ForChannel(string) discclient.ChannelResponse {
+	return &channelResponse{
+		peers: r.peers,
+	}
+}
+
+func (r *response) ForLocal() discclient.LocalResponse {
+	return &localResponse{
+		peers: r.peers,
+	}
+}
+
+type channelResponse struct {
+	peers []*discclient.Peer
+}
+
+// Config returns a response for a config query, or error if something went wrong
+func (cr *channelResponse) Config() (*discovery.ConfigResult, error) {
+	panic("not implemented")
+}
+
+// Peers returns a response for a peer membership query, or error if something went wrong
+func (cr *channelResponse) Peers() ([]*discclient.Peer, error) {
+	return cr.peers, nil
+}
+
+// Endorsers returns the response for an endorser query
+func (cr *channelResponse) Endorsers(cc string, ps discclient.PrioritySelector, ef discclient.ExclusionFilter) (discclient.Endorsers, error) {
+	panic("not implemented")
+}
+
+type localResponse struct {
+	peers []*discclient.Peer
+}
+
+// Peers returns a response for a peer membership query, or error if something went wrong
+func (cr *localResponse) Peers() ([]*discclient.Peer, error) {
+	return cr.peers, nil
+}
+
+func newAliveMessage(endpoint *discmocks.MockDiscoveryPeerEndpoint) *gossip.SignedGossipMessage {
+	return &gossip.SignedGossipMessage{
+		GossipMessage: &gossip.GossipMessage{
+			Content: &gossip.GossipMessage_AliveMsg{
+				AliveMsg: &gossip.AliveMessage{
+					Membership: &gossip.Member{
+						Endpoint: endpoint.Endpoint,
+					},
+				},
+			},
+		},
+	}
+}
+
+func newStateInfoMessage(endpoint *discmocks.MockDiscoveryPeerEndpoint) *gossip.SignedGossipMessage {
+	return &gossip.SignedGossipMessage{
+		GossipMessage: &gossip.GossipMessage{
+			Content: &gossip.GossipMessage_StateInfo{
+				StateInfo: &gossip.StateInfo{
+					Properties: &gossip.Properties{
+						LedgerHeight: endpoint.LedgerHeight,
+					},
+				},
+			},
+		},
+	}
+}

pkg/client/common/discovery/dynamicdiscovery/provider.go

@@ -0,0 +1,83 @@
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package dynamicdiscovery
+
+import (
+	"time"
+
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/logging"
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
+	"github.com/hyperledger/fabric-sdk-go/pkg/util/concurrent/lazycache"
+	"github.com/pkg/errors"
+)
+
+var logger = logging.NewLogger("fabsdk/client")
+
+// Provider implements a dynamic Discovery Provider that queries
+// Fabric's Discovery service for information about the peers that
+// are currently joined to the given channel.
+type Provider struct {
+	cache *lazycache.Cache
+}
+
+// Opt is a provider option
+type Opt func(o *options)
+
+// WithRefreshInterval sets the interval in which the
+// peer cache is refreshed
+func WithRefreshInterval(value time.Duration) Opt {
+	return func(o *options) {
+		o.refreshInterval = value
+	}
+}
+
+// WithResponseTimeout sets the Discover service response timeout
+func WithResponseTimeout(value time.Duration) Opt {
+	return func(o *options) {
+		o.responseTimeout = value
+	}
+}
+
+type options struct {
+	refreshInterval time.Duration
+	responseTimeout time.Duration
+}
+
+// New creates a new dynamic discovery provider
+func New(config fab.EndpointConfig, opts ...Opt) *Provider {
+	options := options{}
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	if options.refreshInterval == 0 {
+		options.refreshInterval = config.Timeout(fab.DiscoveryServiceRefresh)
+	}
+	if options.responseTimeout == 0 {
+		options.responseTimeout = config.Timeout(fab.DiscoveryResponse)
+	}
+
+	return &Provider{
+		cache: lazycache.New("Discovery_Service_Cache", func(key lazycache.Key) (interface{}, error) {
+			return newService(options), nil
+		}),
+	}
+}
+
+// CreateDiscoveryService will create a new membership service
+func (p *Provider) CreateDiscoveryService(channelID string) (fab.DiscoveryService, error) {
+	ref, err := p.cache.Get(lazycache.NewStringKey(channelID))
+	if err != nil {
+		return nil, errors.WithMessage(err, "failed to get discovery service from cache")
+	}
+	return ref.(fab.DiscoveryService), nil
+}
+
+// Close will close the cache and all services contained by the cache.
+func (p *Provider) Close() {
+	p.cache.Close()
+}

pkg/client/common/discovery/dynamicdiscovery/provider_test.go

@@ -0,0 +1,63 @@
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package dynamicdiscovery
+
+import (
+	"testing"
+	"time"
+
+	pfab "github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
+	mspmocks "github.com/hyperledger/fabric-sdk-go/pkg/msp/test/mockmsp"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	ch = "orgchannel"
+
+	mspID1 = "Org1MSP"
+	mspID2 = "Org2MSP"
+
+	peer1MSP1 = "peer1.org1.com:9999"
+)
+
+func TestDiscoveryProvider(t *testing.T) {
+	ctx := mocks.NewMockContext(mspmocks.NewMockSigningIdentity("test", mspID1))
+	config := &config{
+		EndpointConfig: mocks.NewMockEndpointConfig(),
+		peers: []pfab.ChannelPeer{
+			{
+				NetworkPeer: pfab.NetworkPeer{
+					PeerConfig: pfab.PeerConfig{
+						URL: peer1MSP1,
+					},
+				},
+			},
+		},
+	}
+	ctx.SetEndpointConfig(config)
+
+	p := New(config, WithRefreshInterval(30*time.Second), WithResponseTimeout(10*time.Second))
+	defer p.Close()
+
+	service, err := p.CreateDiscoveryService(ch)
+	assert.NoError(t, err)
+
+	chCtx := mocks.NewMockChannelContext(ctx, ch)
+
+	err = service.(*Service).Initialize(chCtx)
+	assert.NoError(t, err)
+}
+
+type config struct {
+	pfab.EndpointConfig
+	peers []pfab.ChannelPeer
+}
+
+func (c *config) ChannelPeers(name string) ([]pfab.ChannelPeer, error) {
+	return c.peers, nil
+}