func DeleteForwardedHeader(ctx thrift.Context) bool {
	_, ok := ctx.Headers()[ForwardedHeaderName]
	if ok {
		delete(ctx.Headers(), ForwardedHeaderName)
	}
	return ok
}
DeleteForwardedHeader takes the headers that came in via TChannel and looks
for the precense of a specific ringpop header to see if ringpop already
forwarded the message. If the header is present it will delete the header
from the context. The return value indicates if the header was present and
deleted
package forward
import (
	"encoding/json"
	"sync"
	"time"
	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/ringpop-go/util"
	"github.com/uber/tchannel-go"
	"github.com/uber/tchannel-go/thrift"
)
type Sender interface {
	WhoAmI() (string, error)
	Lookup(string) (string, error)
}
type Options struct {
	Ctx            thrift.Context
	MaxRetries     int
	RerouteRetries bool
	RetrySchedule  []time.Duration
	Timeout        time.Duration
	Headers        []byte
}
func (f *Forwarder) defaultOptions() *Options {
	return &Options{
		MaxRetries:    3,
		RetrySchedule: []time.Duration{3 * time.Second, 6 * time.Second, 12 * time.Second},
		Timeout:       3 * time.Second,
	}
}
func (f *Forwarder) mergeDefaultOptions(opts *Options) *Options {
	def := f.defaultOptions()
	if opts == nil {
		return def
	}
	var merged Options
	merged.MaxRetries = util.SelectInt(opts.MaxRetries, def.MaxRetries)
	merged.Timeout = util.SelectDuration(opts.Timeout, def.Timeout)
	merged.RerouteRetries = opts.RerouteRetries
	merged.RetrySchedule = opts.RetrySchedule
	if opts.RetrySchedule == nil {
		merged.RetrySchedule = def.RetrySchedule
	}
	merged.Headers = opts.Headers
	merged.Ctx = opts.Ctx
	return &merged
}
type Forwarder struct {
	events.AsyncEventEmitter
	sender  Sender
	channel shared.SubChannel
	logger  log.Logger
	inflightLock sync.Mutex
	inflight     int64
}
func NewForwarder(s Sender, ch shared.SubChannel) *Forwarder {
	logger := logging.Logger("forwarder")
	if address, err := s.WhoAmI(); err == nil {
		logger = logger.WithField("local", address)
	}
	return &Forwarder{
		sender:  s,
		channel: ch,
		logger:  logger,
	}
}
func (f *Forwarder) incrementInflight() {
	f.inflightLock.Lock()
	f.inflight++
	inflight := f.inflight
	f.inflightLock.Unlock()
	f.EmitEvent(InflightRequestsChangedEvent{inflight})
}
func (f *Forwarder) decrementInflight() {
	f.inflightLock.Lock()
	pre := f.inflight
	f.inflight--
	if f.inflight < 0 {
		f.inflight = 0
		f.EmitEvent(InflightRequestsMiscountEvent{InflightDecrement})
	}
	inflight := f.inflight
	f.inflightLock.Unlock()
	if pre != inflight {
		f.EmitEvent(InflightRequestsChangedEvent{inflight})
	}
}
func (f *Forwarder) ForwardRequest(request []byte, destination, service, endpoint string,
	keys []string, format tchannel.Format, opts *Options) ([]byte, error) {
	f.EmitEvent(RequestForwardedEvent{})
	f.incrementInflight()
	opts = f.mergeDefaultOptions(opts)
	rs := newRequestSender(f.sender, f, f.channel, request, keys, destination, service, endpoint, format, opts)
	b, err := rs.Send()
	f.decrementInflight()
	if err != nil {
		f.EmitEvent(FailedEvent{})
	} else {
		f.EmitEvent(SuccessEvent{})
	}
	return b, err
}
var (
	ForwardedHeaderName = "ringpop-forward-keys"
)
func SetForwardedHeader(ctx thrift.Context, keys []string) thrift.Context {
	headers := make(map[string]string, len(ctx.Headers())+1)
	for key, value := range ctx.Headers() {
		headers[key] = value
	}
	if keys == nil {
		keys = []string{}
	}
	keysBytes, _ := json.Marshal(keys)
	keysString := string(keysBytes)
	headers[ForwardedHeaderName] = keysString
	return thrift.WithHeaders(ctx, headers)
}