[FAB-8578] Connection Caching

This change introduces the CachingConnector. This component
provides the ability to cache GRPC connections. It provides a
GRPC compatible Context Dialer interface via the "DialContext" method.

Change-Id: Idd2f61e52a3e078cf81042853808c59c9a37b47b
Signed-off-by: Troy Ronda <troy@troyronda.com>

Author: troyronda

pkg/context/api/core/provider.go

@@ -76,6 +76,10 @@ const (
 	OrdererResponse
 	// DiscoveryGreylistExpiry discovery Greylist expiration period
 	DiscoveryGreylistExpiry
+	// ConnectionIdle is the timeout for closing idle connections
+	ConnectionIdle
+	// CacheSweepInterval is the duration between cache sweeps
+	CacheSweepInterval
 )
 
 // Providers represents the SDK configured core providers context.

pkg/context/api/fab/provider.go

@@ -19,6 +19,7 @@ type InfraProvider interface {
 	CreateEventHub(ic IdentityContext, name string) (EventHub, error)
 	CreatePeerFromConfig(peerCfg *core.NetworkPeer) (Peer, error)
 	CreateOrdererFromConfig(cfg *core.OrdererConfig) (Orderer, error)
+	Close()
 }
 
 // SelectionProvider is used to select peers for endorsement

pkg/core/config/config.go

@@ -34,9 +34,11 @@ import (
 var logger = logging.NewLogger(logModule)
 
 const (
-	cmdRoot        = "FABRIC_SDK"
-	logModule      = "fabric_sdk_go"
-	defaultTimeout = time.Second * 5
+	cmdRoot                   = "FABRIC_SDK"
+	logModule                 = "fabric_sdk_go"
+	defaultTimeout            = time.Second * 5
+	defaultConnIdleTimeout    = time.Second * 30
+	defaultCacheSweepInterval = time.Second * 15
 )
 
 // Config represents the configuration for the client
@@ -458,7 +460,16 @@ func (c *Config) TimeoutOrDefault(conn core.TimeoutType) time.Duration {
 		timeout = c.configViper.GetDuration("client.orderer.timeout.connection")
 	case core.OrdererResponse:
 		timeout = c.configViper.GetDuration("client.orderer.timeout.response")
-
+	case core.CacheSweepInterval: // EXPERIMENTAL - do we need this to be configurable?
+		timeout = c.configViper.GetDuration("client.cache.interval.sweep")
+		if timeout == 0 {
+			timeout = defaultCacheSweepInterval
+		}
+	case core.ConnectionIdle:
+		timeout = c.configViper.GetDuration("client.cache.timeout.connectionIdle")
+		if timeout == 0 {
+			timeout = defaultConnIdleTimeout
+		}
 	}
 	if timeout == 0 {
 		timeout = defaultTimeout

pkg/fab/comm/comm_test.go

@@ -0,0 +1,86 @@
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package comm
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	eventmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/mocks"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
+	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
+	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+)
+
+const (
+	peerAddress     = "localhost:9999"
+	endorserAddress = "127.0.0.1:0"
+	peerURL         = "grpc://" + peerAddress
+)
+
+func TestMain(m *testing.M) {
+	var opts []grpc.ServerOption
+	grpcServer := grpc.NewServer(opts...)
+
+	lis, err := net.Listen("tcp", peerAddress)
+	if err != nil {
+		panic(fmt.Sprintf("Error starting events listener %s", err))
+	}
+
+	testServer = eventmocks.NewMockEventhubServer()
+
+	pb.RegisterEventsServer(grpcServer, testServer)
+
+	go grpcServer.Serve(lis)
+
+	srvs, addrs, err := startEndorsers(2, endorserAddress)
+	if err != nil {
+		panic(fmt.Sprintf("Error starting endorser %s", err))
+	}
+	for _, srv := range srvs {
+		defer srv.Stop()
+	}
+	endorserAddr = addrs
+
+	time.Sleep(2 * time.Second)
+	os.Exit(m.Run())
+}
+
+func startEndorsers(count int, address string) ([]*grpc.Server, []string, error) {
+	srvs := make([]*grpc.Server, 0, count)
+	addrs := make([]string, 0, count)
+
+	for i := 0; i < count; i++ {
+		srv := grpc.NewServer()
+		_, addr, ok := startEndorserServer(srv, address)
+		if !ok {
+			return nil, nil, errors.New("unable to start GRPC server")
+		}
+		srvs = append(srvs, srv)
+		addrs = append(addrs, addr)
+	}
+	return srvs, addrs, nil
+}
+
+func startEndorserServer(grpcServer *grpc.Server, address string) (*mocks.MockEndorserServer, string, bool) {
+	lis, err := net.Listen("tcp", address)
+	if err != nil {
+		fmt.Printf("Error starting test server %s", err)
+		return nil, "", false
+	}
+	addr := lis.Addr().String()
+
+	endorserServer := &mocks.MockEndorserServer{}
+	pb.RegisterEndorserServer(grpcServer, endorserServer)
+	fmt.Printf("Starting test server on %s", addr)
+	go grpcServer.Serve(lis)
+	return endorserServer, addr, true
+}

pkg/fab/comm/connection_test.go

@@ -8,9 +8,6 @@ package comm
 
 import (
 	"context"
-	"fmt"
-	"net"
-	"os"
 	"testing"
 	"time"
 
@@ -26,11 +23,6 @@ import (
 	"google.golang.org/grpc"
 )
 
-const (
-	peerAddress = "localhost:9999"
-	peerURL     = "grpc://" + peerAddress
-)
-
 var testStream = func(grpcconn *grpc.ClientConn) (grpc.ClientStream, error) {
 	return pb.NewDeliverClient(grpcconn).Deliver(context.Background())
 }
@@ -93,25 +85,7 @@ func TestConnection(t *testing.T) {
 
 // Use the Deliver server for testing
 var testServer *eventmocks.MockEventhubServer
-
-func TestMain(m *testing.M) {
-	var opts []grpc.ServerOption
-	grpcServer := grpc.NewServer(opts...)
-
-	lis, err := net.Listen("tcp", peerAddress)
-	if err != nil {
-		panic(fmt.Sprintf("Error starting events listener %s", err))
-	}
-
-	testServer = eventmocks.NewMockEventhubServer()
-
-	pb.RegisterEventsServer(grpcServer, testServer)
-
-	go grpcServer.Serve(lis)
-
-	time.Sleep(2 * time.Second)
-	os.Exit(m.Run())
-}
+var endorserAddr []string
 
 func newPeerConfig(peerURL string) *core.PeerConfig {
 	return &core.PeerConfig{