func (s *requestSender) LookupKeys(keys []string) []string {
	destSet := make(map[string]struct{})
	for _, key := range keys {
		dest, err := s.sender.Lookup(key)
		if err != nil {
			continue
		}
		destSet[dest] = struct{}{}
	}
	dests := make([]string, 0, len(destSet))
	for dest := range destSet {
		dests = append(dests, dest)
	}
	return dests
}
LookupKeys looks up the destinations of the keys provided. Returns a slice
of destinations. If multiple keys hash to the same destination, they will
be deduped.
package forward
import (
	"encoding/json"
	"errors"
	"time"
	"golang.org/x/net/context"
	log "github.com/uber-common/bark"
	"github.com/uber/ringpop-go/events"
	"github.com/uber/ringpop-go/logging"
	"github.com/uber/ringpop-go/shared"
	"github.com/uber/tchannel-go"
	"github.com/uber/tchannel-go/raw"
	"github.com/uber/tchannel-go/thrift"
)
var errDestinationsDiverged = errors.New("key destinations have diverged")
type requestSender struct {
	sender  Sender
	emitter events.EventEmitter
	channel shared.SubChannel
	request           []byte
	destination       string
	service, endpoint string
	keys              []string
	format            tchannel.Format
	destinations []string  
	timeout             time.Duration
	retries, maxRetries int
	retrySchedule       []time.Duration
	rerouteRetries      bool
	headers []byte
	ctx     thrift.Context
	startTime, retryStartTime time.Time
	logger log.Logger
}
func newRequestSender(sender Sender, emitter events.EventEmitter, channel shared.SubChannel, request []byte, keys []string,
	destination, service, endpoint string, format tchannel.Format, opts *Options) *requestSender {
	logger := logging.Logger("sender")
	if address, err := sender.WhoAmI(); err != nil {
		logger = logger.WithField("local", address)
	}
	return &requestSender{
		sender:         sender,
		emitter:        emitter,
		channel:        channel,
		request:        request,
		keys:           keys,
		destination:    destination,
		service:        service,
		endpoint:       endpoint,
		format:         format,
		timeout:        opts.Timeout,
		maxRetries:     opts.MaxRetries,
		retrySchedule:  opts.RetrySchedule,
		rerouteRetries: opts.RerouteRetries,
		headers:        opts.Headers,
		ctx:            opts.Ctx,
		logger:         logger,
	}
}
func (s *requestSender) Send() (res []byte, err error) {
	var cancel context.CancelFunc
	var ctx thrift.Context
	if s.ctx == nil {
		ctx, cancel = shared.NewTChannelContext(s.timeout)
	} else {
		var c context.Context
		c, cancel = context.WithTimeout(s.ctx, s.timeout)
		ctx = tchannel.Wrap(c)
	}
	defer cancel()
	var forwardError, applicationError error
	select {
	case <-s.MakeCall(ctx, &res, &forwardError, &applicationError):
		if applicationError != nil {
			return nil, applicationError
		}
		if forwardError == nil {
			if s.retries > 0 {
				s.emitter.EmitEvent(RetrySuccessEvent{s.retries})
			}
			return res, nil
		}
		if s.retries < s.maxRetries {
			return s.ScheduleRetry()
		}
		address, _ := s.sender.WhoAmI()
		s.logger.WithFields(log.Fields{
			"local":       address,
			"destination": s.destination,
			"service":     s.service,
			"endpoint":    s.endpoint,
		}).Warn("max retries exceeded for request")
		s.emitter.EmitEvent(MaxRetriesEvent{s.maxRetries})
		return nil, errors.New("max retries exceeded")
	case <-ctx.Done():  
		address, _ := s.sender.WhoAmI()
		s.logger.WithFields(log.Fields{
			"local":       address,
			"destination": s.destination,
			"service":     s.service,
			"endpoint":    s.endpoint,
		}).Warn("request timed out")
		return nil, errors.New("request timed out")
	}
}
func (s *requestSender) MakeCall(ctx context.Context, res *[]byte, fwdError *error, appError *error) <-chan bool {
	done := make(chan bool, 1)
	go func() {
		defer close(done)
		peer := s.channel.Peers().GetOrAdd(s.destination)
		call, err := peer.BeginCall(ctx, s.service, s.endpoint, &tchannel.CallOptions{
			Format: s.format,
		})
		if err != nil {
			*fwdError = err
			done <- true
			return
		}
		var arg3 []byte
		headers := s.headers
		if s.format == tchannel.Thrift {
			if headers == nil {
				headers = []byte{0, 0}
			}
			_, arg3, _, err = raw.WriteArgs(call, headers, s.request)
		} else {
			var resp *tchannel.OutboundCallResponse
			_, arg3, resp, err = raw.WriteArgs(call, headers, s.request)
			if err == nil && resp.ApplicationError() {
				errResp := struct {
					Type    string `json:"type"`
					Message string `json:"message"`
				}{}
				err = json.Unmarshal(arg3, &errResp)
				if err == nil {
					*appError = errors.New(errResp.Message)
					done <- true
					return
				}
			}
		}
		if err != nil {
			*fwdError = err
			done <- true
			return
		}
		*res = arg3
		done <- true
	}()
	return done
}
func (s *requestSender) ScheduleRetry() ([]byte, error) {
	if s.retries == 0 {
		s.retryStartTime = time.Now()
	}
	time.Sleep(s.retrySchedule[s.retries])
	return s.AttemptRetry()
}
func (s *requestSender) AttemptRetry() ([]byte, error) {
	s.retries++
	s.emitter.EmitEvent(RetryAttemptEvent{})
	dests := s.LookupKeys(s.keys)
	if len(dests) != 1 {
		s.emitter.EmitEvent(RetryAbortEvent{errDestinationsDiverged.Error()})
		return nil, errDestinationsDiverged
	}
	if s.rerouteRetries {
		newDest := dests[0]
		if newDest != s.destination {
			return s.RerouteRetry(newDest)
		}
	}
	return s.Send()
}
func (s *requestSender) RerouteRetry(destination string) ([]byte, error) {
	s.emitter.EmitEvent(RerouteEvent{
		s.destination,
		destination,
	})
	s.destination = destination  
	return s.Send()
}