func (h *InputHost) getExtentsInfoForDestination(ctx thrift.Context, destUUID string) ([]*extentInfo, error) {
	listResp, err := h.listExtentsByStatus(ctx, destUUID, common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN))
	if err != nil {
		return nil, err
	}
	var extents = make([]*extentInfo, 0, 2)
	for _, item := range listResp.ExtentStatsList {
		extent := item.Extent
		info := h.createExtentInfo(extent.GetExtentUUID(), extent.StoreUUIDs)
		if info == nil {
			continue
		}
		extents = append(extents, info)
	}
	if len(extents) < 1 {
		h.logger.WithField(common.TagDst, common.FmtDst(destUUID)).
			WithField("nExtents", len(listResp.GetExtentStatsList())).
			Error("Can't find extents for destination, no healthy extent found")
		return nil, fmt.Errorf("No healthy extent found")
	}
	return extents, nil
}
getExtentsInfoForDestination returns all the active writable extents
on this inputhost for the given destination.
package inputhost
import (
	"fmt"
	"net/http"
	"os"
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
	ccommon "github.com/uber/cherami-client-go/common"
	"github.com/uber/cherami-server/common"
	dconfig "github.com/uber/cherami-server/common/dconfigclient"
	mm "github.com/uber/cherami-server/common/metadata"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/inputhost/load"
	"github.com/uber/cherami-server/stream"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/controller"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
)
const (
	defaultIdleTimeout = 10 * time.Minute
	defaultBufferSize = 1000
	hostLoadReportingInterval = 2 * time.Second
	minimumAllowedMessageDelaySeconds = 60
	minimumAllowedMessageDelaySecondsTest = 3
	extentRolloverSeqnumMin, extentRolloverSeqnumMax = 10000000, 20000000
	defaultDrainTimeout = 1 * time.Minute
	connWGTimeout = 5 * time.Second
	drainAllUUID = "D3A9C5DC-AE62-4465-9898-7FE71BD1FCA"
)
var (
	batchMsgAckTimeout = 1 * time.Minute  
)
type (
	InputHost struct {
		logger                 bark.Logger
		loadShutdownRef        int32
		pathMutex              sync.RWMutex
		pathCache              map[string]*inPathCache
		pathCacheByDestPath    map[string]string
		shutdownWG             sync.WaitGroup
		shutdown               chan struct{}
		cacheTimeout           time.Duration
		mClient                metadata.TChanMetadataService
		hostIDHeartbeater      common.HostIDHeartbeater
		loadReporter           common.LoadReporterDaemon
		m3Client               metrics.Client
		dConfigClient          dconfig.Client
		tokenBucketValue       atomic.Value  
		hostConnLimit          int32
		hostConnLimitPerSecond int32
		maxConnLimit           int32
		extMsgsLimitPerSecond  int32
		connMsgsLimitPerSecond int32
		hostMetrics            *load.HostMetrics
		lastLoadReportedTime   int64         
		nodeStatus             atomic.Value  
		testShortExtentsByPath string        
		common.SCommon
	}
	InOptions struct {
		CacheIdleTimeout time.Duration
	}
	extentInfo struct {
		uuid     string
		replicas []string
	}
	inReconfigInfo struct {
		req        *admin.DestinationUpdatedNotification
		updateUUID string
	}
)
var _ cherami.TChanBIn = &InputHost{}
var _ admin.TChanInputHostAdmin = &InputHost{}
var _ common.WSService = &InputHost{}
var ErrHostShutdown = &cherami.InternalServiceError{Message: "InputHost already shutdown"}
var ErrThrottled = &cherami.InternalServiceError{Message: "InputHost throttling publisher cconnection"}
var ErrDstNotLoaded = &cherami.InternalServiceError{Message: "Destination no longer served by this input host"}
var ErrDrainTimedout = &cherami.InternalServiceError{Message: "Draining of Extents timedout"}
func (h *InputHost) isDestinationWritable(destDesc *shared.DestinationDescription) bool {
	status := destDesc.GetStatus()
	if status != shared.DestinationStatus_ENABLED && status != shared.DestinationStatus_SENDONLY {
		return false
	}
	return true
}
func (h *InputHost) listExtentsByStatus(ctx thrift.Context, destUUID string, status *shared.ExtentStatus) (*metadata.ListInputHostExtentsStatsResult_, error) {
	listReq := metadata.ListInputHostExtentsStatsRequest{
		InputHostUUID:   common.StringPtr(h.GetHostUUID()),
		DestinationUUID: common.StringPtr(destUUID),
		Status:          status,
	}
	listResp, err := h.mClient.ListInputHostExtentsStats(ctx, &listReq)
	if err != nil {
		return nil, fmt.Errorf("ListInputHostExtentStats() returned error dst=%v, err=%v",
			destUUID, err)
	}
	if listResp == nil || len(listResp.GetExtentStatsList()) == 0 {
		return nil, fmt.Errorf("getExtentsInfoForDestination - Found zero extents for destination, dst=%v",
			destUUID)
	}
	return listResp, nil
}
func (h *InputHost) createExtentInfo(extentUUID string, storeIDs []string) *extentInfo {
	if extentUUID == "" || storeIDs == nil || len(storeIDs) == 0 {
		return nil
	}
	var rpm = h.GetRingpopMonitor()
	var info = &extentInfo{
		uuid:     extentUUID,
		replicas: make([]string, 0, 3),
	}
	for _, storeID := range storeIDs {
		addr, err := rpm.ResolveUUID(common.StoreServiceName, storeID)
		if err != nil {
			h.logger.WithField(`storeID`, storeID).
				WithField(common.TagErr, err).
				Error("Cannot resolve UUID for store")
			return nil
		}
		info.replicas = append(info.replicas, addr)
	}
	return info
}