[FAB-8571] request context by higherlevel clients

- higher level clients to create request context for
lower level clients
- higher level clients to create parent request context
for  multiple lower level operations
- added timeout options in higher level ledger client
 functions QueryInfo,QueryBlockByHash,QueryBlock,
QueryTransaction,QueryConfig
- InfraProvider CreateChannelTransactor function signature changed
- Transactor() removed from channel service
- ChannelConfig Query() function signature changed
- new timeout types added peer.timeout.response and
global.timeout.resmgmt

TODO:
- overall and low level timeout override options
for high level operation.



Change-Id: I1e0284c7e8fc7e2bc855ca603d5b2bc10de65410
Signed-off-by: Sudesh Shetty <sudesh.shetty@securekey.com>

Author: sudeshrshetty

pkg/client/channel/chclient.go

@@ -8,12 +8,14 @@ SPDX-License-Identifier: Apache-2.0
 package channel
 
 import (
+	reqContext "context"
 	"time"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/channel/invoke"
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/discovery"
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/discovery/greylist"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/context"
+	contextImpl "github.com/hyperledger/fabric-sdk-go/pkg/context"
 	"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/multi"
@@ -37,7 +39,6 @@ const (
 type Client struct {
 	context         context.Channel
 	membership      fab.ChannelMembership
-	transactor      fab.Transactor
 	eventService    fab.EventService
 	greylist        *greylist.Filter
 	discoveryFilter fab.TargetFilter
@@ -82,19 +83,13 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 		return nil, errors.WithMessage(err, "event service creation failed")
 	}
 
-	transactor, err := channelContext.ChannelService().Transactor()
-	if err != nil {
-		return nil, errors.WithMessage(err, "transactor creation failed")
-	}
-
 	membership, err := channelContext.ChannelService().Membership()
 	if err != nil {
 		return nil, errors.WithMessage(err, "membership creation failed")
 	}
 
 	channelClient := Client{
 		membership:   membership,
-		transactor:   transactor,
 		eventService: eventService,
 		greylist:     greylistProvider,
 	}
@@ -133,8 +128,11 @@ func (cc *Client) InvokeHandler(handler invoke.Handler, request Request, options
 		return Response{}, err
 	}
 
+	reqCtx, cancel := cc.createReqContext(&txnOpts)
+	defer cancel()
+
 	//Prepare context objects for handler
-	requestContext, clientContext, err := cc.prepareHandlerContexts(request, txnOpts)
+	requestContext, clientContext, err := cc.prepareHandlerContexts(reqCtx, request, txnOpts)
 	if err != nil {
 		return Response{}, err
 	}
@@ -180,18 +178,39 @@ func (cc *Client) resolveRetry(ctx *invoke.RequestContext, o requestOptions) boo
 	return false
 }
 
+//createReqContext creates req context for invoke handler
+func (cc *Client) createReqContext(txnOpts *requestOptions) (reqContext.Context, reqContext.CancelFunc) {
+
+	//Setting default timeouts when not provided
+	if txnOpts.Timeout == 0 {
+		txnOpts.Timeout = cc.context.Config().Timeout(core.Execute)
+		if txnOpts.Timeout == 0 {
+			//If still zero, then set default handler timeout
+			txnOpts.Timeout = defaultHandlerTimeout
+		}
+	}
+
+	return contextImpl.NewRequest(cc.context, contextImpl.WithTimeout(txnOpts.Timeout))
+}
+
 //prepareHandlerContexts prepares context objects for handlers
-func (cc *Client) prepareHandlerContexts(request Request, o requestOptions) (*invoke.RequestContext, *invoke.ClientContext, error) {
+func (cc *Client) prepareHandlerContexts(reqCtx reqContext.Context, request Request, o requestOptions) (*invoke.RequestContext, *invoke.ClientContext, error) {
 
 	if request.ChaincodeID == "" || request.Fcn == "" {
 		return nil, nil, errors.New("ChaincodeID and Fcn are required")
 	}
 
+	chConfig := cc.context.ChannelService().ChannelConfig()
+	transactor, err := cc.context.InfraProvider().CreateChannelTransactor(reqCtx, chConfig)
+	if err != nil {
+		return nil, nil, errors.WithMessage(err, "failed to create transactor")
+	}
+
 	clientContext := &invoke.ClientContext{
 		Selection:    cc.context.SelectionService(),
 		Discovery:    cc.context.DiscoveryService(),
 		Membership:   cc.membership,
-		Transactor:   cc.transactor,
+		Transactor:   transactor,
 		EventService: cc.eventService,
 	}
 
@@ -202,15 +221,6 @@ func (cc *Client) prepareHandlerContexts(request Request, o requestOptions) (*in
 		RetryHandler: retry.New(o.Retry),
 	}
 
-	if requestContext.Opts.Timeout == 0 {
-		to := cc.context.Config().Timeout(core.Execute)
-		if to == 0 {
-			requestContext.Opts.Timeout = defaultHandlerTimeout
-		} else {
-			requestContext.Opts.Timeout = to
-		}
-	}
-
 	return requestContext, clientContext, nil
 }
 

pkg/client/channel/chclient_test.go

@@ -503,13 +503,6 @@ func setupTestChannelService(ctx context.Client, orderers []fab.Orderer) (fab.Ch
 		return nil, errors.WithMessage(err, "mock channel service creation failed")
 	}
 
-	transactor := txnmocks.MockTransactor{
-		Ctx:       ctx,
-		ChannelID: channelID,
-		Orderers:  orderers,
-	}
-	chService.(*fcmocks.MockChannelService).SetTransactor(&transactor)
-
 	return chService, nil
 }
 
@@ -528,6 +521,14 @@ func setupCustomTestContext(t *testing.T, selectionService fab.SelectionService,
 		orderers = []fab.Orderer{orderer}
 	}
 
+	transactor := txnmocks.MockTransactor{
+		Ctx:       ctx,
+		ChannelID: channelID,
+		Orderers:  orderers,
+	}
+
+	ctx.InfraProvider().(*fcmocks.MockInfraProvider).SetCustomTransactor(&transactor)
+
 	testChannelSvc, err := setupTestChannelService(ctx, orderers)
 	assert.Nil(t, err, "Got error %s", err)
 

pkg/client/common/mocks/mocktransactor.go

@@ -7,7 +7,10 @@ SPDX-License-Identifier: Apache-2.0
 package mocks
 
 import (
+	"time"
+
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/context"
+	contextImpl "github.com/hyperledger/fabric-sdk-go/pkg/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/txn"
 	"github.com/pkg/errors"
@@ -32,7 +35,9 @@ func (t *MockTransactor) CreateTransactionHeader() (fab.TransactionHeader, error
 
 // SendTransactionProposal sends a TransactionProposal to the target peers.
 func (t *MockTransactor) SendTransactionProposal(proposal *fab.TransactionProposal, targets []fab.ProposalProcessor) ([]*fab.TransactionProposalResponse, error) {
-	return txn.SendProposal(t.Ctx, proposal, targets)
+	rqtx, cancel := contextImpl.NewRequest(t.Ctx, contextImpl.WithTimeout(10*time.Second))
+	defer cancel()
+	return txn.SendProposal(rqtx, proposal, targets)
 }
 
 // CreateTransaction create a transaction with proposal response.
@@ -42,5 +47,7 @@ func (t *MockTransactor) CreateTransaction(request fab.TransactionRequest) (*fab
 
 // SendTransaction send a transaction to the chain’s orderer service (one or more orderer endpoints) for consensus and committing to the ledger.
 func (t *MockTransactor) SendTransaction(tx *fab.Transaction) (*fab.TransactionResponse, error) {
-	return txn.Send(t.Ctx, tx, t.Orderers)
+	rqtx, cancel := contextImpl.NewRequest(t.Ctx, contextImpl.WithTimeout(10*time.Second))
+	defer cancel()
+	return txn.Send(rqtx, tx, t.Orderers)
 }

pkg/client/ledger/ledger.go

@@ -8,6 +8,7 @@ SPDX-License-Identifier: Apache-2.0
 package ledger
 
 import (
+	reqContext "context"
 	"math/rand"
 	"time"
 
@@ -18,6 +19,8 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/common"
 	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
 
+	contextImpl "github.com/hyperledger/fabric-sdk-go/pkg/context"
+	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/core"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/channel"
 	"github.com/hyperledger/fabric-sdk-go/pkg/logging"
 	"github.com/pkg/errors"
@@ -35,9 +38,9 @@ const (
 // An application that requires interaction with multiple channels should create a separate
 // instance of the ledger client for each channel. Ledger client supports specific queries only.
 type Client struct {
-	context context.Channel
-	filter  TargetFilter
-	ledger  *channel.Ledger
+	ctx    context.Channel
+	filter TargetFilter
+	ledger *channel.Ledger
 }
 
 // MSPFilter is default filter
@@ -58,14 +61,14 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 		return nil, err
 	}
 
-	ledger, err := channel.NewLedger(channelContext, channelContext.ChannelID())
+	ledger, err := channel.NewLedger(channelContext.ChannelID())
 	if err != nil {
 		return nil, err
 	}
 
 	ledgerClient := Client{
-		context: channelContext,
-		ledger:  ledger,
+		ctx:    channelContext,
+		ledger: ledger,
 	}
 
 	for _, opt := range opts {
@@ -103,7 +106,10 @@ func (c *Client) QueryInfo(options ...RequestOption) (*fab.BlockchainInfoRespons
 		return nil, errors.WithMessage(err, "failed to determine target peers for QueryBlockByHash")
 	}
 
-	responses, err := c.ledger.QueryInfo(peersToTxnProcessors(targets))
+	reqCtx, cancel := c.createRequestContext(opts)
+	defer cancel()
+
+	responses, err := c.ledger.QueryInfo(reqCtx, peersToTxnProcessors(targets))
 	if err != nil && len(responses) == 0 {
 		return nil, errors.WithMessage(err, "Failed to QueryBlockByHash")
 	}
@@ -146,7 +152,10 @@ func (c *Client) QueryBlockByHash(blockHash []byte, options ...RequestOption) (*
 		return nil, errors.WithMessage(err, "failed to determine target peers for QueryBlockByHash")
 	}
 
-	responses, err := c.ledger.QueryBlockByHash(blockHash, peersToTxnProcessors(targets))
+	reqCtx, cancel := c.createRequestContext(opts)
+	defer cancel()
+
+	responses, err := c.ledger.QueryBlockByHash(reqCtx, blockHash, peersToTxnProcessors(targets))
 	if err != nil && len(responses) == 0 {
 		return nil, errors.WithMessage(err, "Failed to QueryBlockByHash")
 	}
@@ -187,7 +196,10 @@ func (c *Client) QueryBlock(blockNumber int, options ...RequestOption) (*common.
 		return nil, errors.WithMessage(err, "failed to determine target peers for QueryBlock")
 	}
 
-	responses, err := c.ledger.QueryBlock(blockNumber, peersToTxnProcessors(targets))
+	reqCtx, cancel := c.createRequestContext(opts)
+	defer cancel()
+
+	responses, err := c.ledger.QueryBlock(reqCtx, blockNumber, peersToTxnProcessors(targets))
 	if err != nil && len(responses) == 0 {
 		return nil, errors.WithMessage(err, "Failed to QueryBlock")
 	}
@@ -229,7 +241,10 @@ func (c *Client) QueryTransaction(transactionID fab.TransactionID, options ...Re
 		return nil, errors.WithMessage(err, "failed to determine target peers for QueryTransaction")
 	}
 
-	responses, err := c.ledger.QueryTransaction(transactionID, peersToTxnProcessors(targets))
+	reqCtx, cancel := c.createRequestContext(opts)
+	defer cancel()
+
+	responses, err := c.ledger.QueryTransaction(reqCtx, transactionID, peersToTxnProcessors(targets))
 	if err != nil && len(responses) == 0 {
 		return nil, errors.WithMessage(err, "Failed to QueryTransaction")
 	}
@@ -269,20 +284,23 @@ func (c *Client) QueryConfig(options ...RequestOption) (fab.ChannelCfg, error) {
 		return nil, errors.WithMessage(err, "failed to determine target peers for QueryConfig")
 	}
 
-	channelConfig, err := chconfig.New(c.context, c.context.ChannelID(), chconfig.WithPeers(targets), chconfig.WithMinResponses(opts.MinTargets))
+	channelConfig, err := chconfig.New(c.ctx.ChannelID(), chconfig.WithPeers(targets), chconfig.WithMinResponses(opts.MinTargets))
 	if err != nil {
 		return nil, errors.WithMessage(err, "QueryConfig failed")
 	}
 
-	return channelConfig.Query()
+	reqCtx, cancel := c.createRequestContext(opts)
+	defer cancel()
+
+	return channelConfig.Query(reqCtx)
 
 }
 
 //prepareRequestOpts Reads Opts from Option array
 func (c *Client) prepareRequestOpts(options ...RequestOption) (requestOptions, error) {
 	opts := requestOptions{}
 	for _, option := range options {
-		err := option(c.context, &opts)
+		err := option(c.ctx, &opts)
 		if err != nil {
 			return opts, errors.WithMessage(err, "Failed to read request opts")
 		}
@@ -318,7 +336,7 @@ func (c *Client) calculateTargets(opts requestOptions) ([]fab.Peer, error) {
 	var err error
 	if targets == nil {
 		// Retrieve targets from discovery
-		targets, err = c.context.DiscoveryService().GetPeers()
+		targets, err = c.ctx.DiscoveryService().GetPeers()
 		if err != nil {
 			return nil, err
 		}
@@ -352,6 +370,17 @@ func (c *Client) calculateTargets(opts requestOptions) ([]fab.Peer, error) {
 	return targets[:numOfTargets], nil
 }
 
+//createRequestContext creates request context for grpc
+func (c *Client) createRequestContext(opts requestOptions) (reqContext.Context, reqContext.CancelFunc) {
+
+	timeout := opts.Timeout
+	if timeout == 0 {
+		timeout = c.ctx.Config().TimeoutOrDefault(core.PeerResponse)
+	}
+
+	return contextImpl.NewRequest(c.ctx, contextImpl.WithTimeout(timeout))
+}
+
 // filterTargets is helper method to filter peers
 func filterTargets(peers []fab.Peer, filter TargetFilter) []fab.Peer {
 

pkg/client/ledger/opts.go

@@ -7,6 +7,8 @@ SPDX-License-Identifier: Apache-2.0
 package ledger
 
 import (
+	"time"
+
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/core/config"
@@ -40,10 +42,11 @@ type TargetFilter interface {
 
 //requestOptions contains options for operations performed by LedgerClient
 type requestOptions struct {
-	Targets      []fab.Peer   // target peers
-	TargetFilter TargetFilter // target filter
-	MaxTargets   int          // maximum number of targets to select
-	MinTargets   int          // min number of targets that have to respond with no error (or agree on result)
+	Targets      []fab.Peer    // target peers
+	TargetFilter TargetFilter  // target filter
+	MaxTargets   int           // maximum number of targets to select
+	MinTargets   int           // min number of targets that have to respond with no error (or agree on result)
+	Timeout      time.Duration //timeout options for QueryInfo,QueryBlockByHash,QueryBlock,QueryTransaction,QueryConfig
 }
 
 //WithTargets encapsulates fab.Peer targets to ledger RequestOption
@@ -104,3 +107,12 @@ func WithMinTargets(minTargets int) RequestOption {
 		return nil
 	}
 }
+
+//WithTimeout encapsulates timeout to ledger RequestOption
+//for QueryInfo,QueryBlockByHash,QueryBlock,QueryTransaction,QueryConfig functions
+func WithTimeout(timeout time.Duration) RequestOption {
+	return func(ctx context.Client, opts *requestOptions) error {
+		opts.Timeout = timeout
+		return nil
+	}
+}

pkg/client/msp/testdata/config_test.yaml

@@ -46,6 +46,7 @@ client:
 # Global configuration for peer, event service and orderer timeouts
   peer:
     timeout:
+      response: 10s
       connection: 3s
       discovery:
         # Expiry period for discovery service greylist filter
@@ -65,6 +66,7 @@ client:
     timeout:
       query: 45s
       execute: 60s
+      resmgmt: 60s
     cache:
       connectionIdle: 30s