func (s *InputHostSuite) TestInputHostPutMessageBatchTimeout() {
	destinationPath := "foo"
	ctx, cancel := utilGetThriftContextWithPath(destinationPath)
	defer cancel()
	batchMsgAckTimeout = 1 * time.Millisecond
	inputHost, _ := NewInputHost("inputhost-test", s.mockService, s.mockMeta, nil)
	aMsg := store.NewAppendMessageAck()
	msg := cherami.NewPutMessage()
	appendTicker := time.NewTicker(5 * time.Second)
	defer appendTicker.Stop()
	pubTicker := time.NewTicker(5 * time.Second)
	defer pubTicker.Stop()
	s.mockAppend.On("Write", mock.Anything).Return(nil).WaitUntil(appendTicker.C)
	s.mockPub.On("Write", mock.Anything).Return(nil).WaitUntil(pubTicker.C)
	s.mockAppend.On("Read").Return(aMsg, io.EOF).WaitUntil(appendTicker.C)
	s.mockPub.On("Read").Return(msg, io.EOF).WaitUntil(pubTicker.C)
	s.mockStore.On("OpenAppendStream", mock.Anything).Return(s.mockAppend, nil)
	putMessageRequest := &cherami.PutMessageBatchRequest{DestinationPath: &destinationPath}
	msg.ID = common.StringPtr(strconv.Itoa(1))
	msg.Data = []byte(fmt.Sprintf("hello-%d", 1))
	putMessageRequest.Messages = append(putMessageRequest.Messages, msg)
	putMessageAcks, err := inputHost.PutMessageBatch(ctx, putMessageRequest)
	s.NoError(err)
	s.NotNil(putMessageAcks)
	s.Len(putMessageAcks.GetSuccessMessages(), 0)
	s.Len(putMessageAcks.GetFailedMessages(), 1)
	s.Equal(cherami.Status_TIMEDOUT, putMessageAcks.GetFailedMessages()[0].GetStatus())
	inputHost.Shutdown()
}
TestInputHostPutMessageBatchTimeout publishes a batch of messages and make sure
we timeout appropriately
package inputhost
import (
	"fmt"
	"io"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
	"github.com/pborman/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/uber-common/bark"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/configure"
	dconfig "github.com/uber/cherami-server/common/dconfigclient"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/inputhost/load"
	mockcommon "github.com/uber/cherami-server/test/mocks/common"
	mockin "github.com/uber/cherami-server/test/mocks/inputhost"
	mockmeta "github.com/uber/cherami-server/test/mocks/metadata"
	mockstore "github.com/uber/cherami-server/test/mocks/storehost"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	"github.com/uber/cherami-thrift/.generated/go/store"
	"github.com/uber/tchannel-go"
	"github.com/uber/tchannel-go/thrift"
	"golang.org/x/net/context"
)
type InputHostSuite struct {
	*require.Assertions  
	suite.Suite
	cfg                   configure.CommonAppConfig
	mockPub               *mockin.MockBInOpenPublisherStreamInCall
	mockStore             *mockstore.MockStoreHost
	mockAppend            *mockstore.MockBStoreOpenAppendStreamOutCall
	mockService           *common.MockService
	mockMeta              *mockmeta.TChanMetadataService
	mockClientFactory     *mockcommon.MockClientFactory
	mockWSConnector       *mockcommon.MockWSConnector
	mockRpm               *common.MockRingpopMonitor
	desinationDescription *shared.DestinationDescription
}
type mockTimeSource struct {
	lk       sync.RWMutex
	currTime time.Time
}
func (ts *mockTimeSource) Now() time.Time {
	ts.lk.RLock()
	defer ts.lk.RUnlock()
	return ts.currTime
}
func (ts *mockTimeSource) advance(d time.Duration) {
	ts.lk.Lock()
	defer ts.lk.Unlock()
	ts.currTime = ts.currTime.Add(d)
}
func TestInputHostSuite(t *testing.T) {
	s := new(InputHostSuite)
	s.cfg = common.SetupServerConfig(configure.NewCommonConfigure())
	suite.Run(t, s)
}
func (s *InputHostSuite) SetupCommonMock() {
	s.mockPub = new(mockin.MockBInOpenPublisherStreamInCall)
	s.mockStore = new(mockstore.MockStoreHost)
	s.mockAppend = new(mockstore.MockBStoreOpenAppendStreamOutCall)
	s.mockService = new(common.MockService)
	s.mockClientFactory = new(mockcommon.MockClientFactory)
	s.mockMeta = new(mockmeta.TChanMetadataService)
	s.mockWSConnector = new(mockcommon.MockWSConnector)
	ch, _ := tchannel.NewChannel("cherami-test-in", nil)
	s.mockAppend.On("Done").Return(nil)
	s.mockAppend.On("Flush").Return(nil)
	s.mockPub.On("Done").Return(nil)
	s.mockPub.On("Flush").Return(nil)
	s.mockService.On("GetConfig").Return(s.cfg.GetServiceConfig(common.InputServiceName))
	s.mockService.On("GetTChannel").Return(ch)
	s.mockService.On("GetHostPort").Return("inputhost:port")
	s.mockService.On("IsLimitsEnabled").Return(true)
	s.mockService.On("GetHostUUID").Return("99999999-9999-9999-9999-999999999999")
	s.mockService.On("GetMetricsReporter").Return(common.NewMetricReporterWithHostname(configure.NewCommonServiceConfig()))
	s.mockService.On("GetDConfigClient").Return(dconfig.NewDconfigClient(configure.NewCommonServiceConfig(), common.InputServiceName))
	s.mockClientFactory.On("GetThriftStoreClient", mock.Anything, mock.Anything).Return(store.TChanBStore(s.mockStore), nil)
	s.mockService.On("GetWSConnector").Return(s.mockWSConnector, nil)
	s.mockClientFactory.On("ReleaseThriftStoreClient", mock.Anything).Return(nil)
	s.mockClientFactory.On("GetAdminControllerClient").Return(nil, fmt.Errorf("no real controller"))
	s.mockWSConnector.On("OpenAppendStream", mock.Anything, mock.Anything).Return(s.mockAppend, nil)
	s.desinationDescription = shared.NewDestinationDescription()
	s.desinationDescription.DestinationUUID = common.StringPtr(uuid.New())
	mExt := shared.NewExtent()
	mExt.ExtentUUID = common.StringPtr(uuid.New())
	mExt.InputHostUUID = common.StringPtr(uuid.New())
	mExt.StoreUUIDs = make([]string, 0, 4)
	s.mockStore.On("OpenAppendStream", mock.Anything).Return(s.mockAppend, nil).Once()
	mExtStats := shared.NewExtentStats()
	mExtStats.Extent = mExt
	mExtStats.Status = common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN)
	rpm := common.NewMockRingpopMonitor()
	rpm.Add(common.InputServiceName, mExt.GetInputHostUUID(), "127.0.0.1")
	for i := 1; i < 2; i++ {
		uuid := uuid.New()
		rpm.Add(common.StoreServiceName, uuid, "127.0.2."+strconv.Itoa(i))
		mExt.StoreUUIDs = append(mExt.StoreUUIDs, uuid)
	}
	s.mockRpm = rpm
	mListExtStats := metadata.NewListInputHostExtentsStatsResult_()
	mListExtStats.ExtentStatsList = []*shared.ExtentStats{mExtStats}
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(s.desinationDescription, nil)
	s.mockMeta.On("ListInputHostExtentsStats", mock.Anything, mock.Anything).Return(mListExtStats, nil)
	s.mockService.On("GetRingpopMonitor").Return(s.mockRpm)
	s.mockService.On("GetClientFactory").Return(s.mockClientFactory)
	mockLoadReporterDaemonFactory := setupMockLoadReporterDaemonFactory()
	s.mockService.On("GetLoadReporterDaemonFactory").Return(mockLoadReporterDaemonFactory)
}
func (s *InputHostSuite) SetupTest() {
	s.Assertions = require.New(s.T())  
	s.SetupCommonMock()
}
func (s *InputHostSuite) TearDownTest() {
}
func utilGetThriftContext() (thrift.Context, context.CancelFunc) {
	return thrift.NewContext(20 * time.Second)
}
func utilGetThriftContextWithPath(path string) (thrift.Context, context.CancelFunc) {
	ctx, cancel := utilGetThriftContext()
	headers := make(map[string]string)
	headers["path"] = "foo"
	ctx = thrift.WithHeaders(ctx, headers)
	return ctx, cancel
}
func (s *InputHostSuite) TestInputHostRejectNoPath() {
	sName := "chermai-test-in"
	inputHost, _ := NewInputHost(sName, s.mockService, nil, nil)
	ctx, _ := utilGetThriftContext()
	err := inputHost.OpenPublisherStream(ctx, s.mockPub)
	s.Error(err)
	assert.IsType(s.T(), &cherami.BadRequestError{}, err)
}
func (s *InputHostSuite) TestInputHostNoReplicas() {
	sName := "chermai-test-in"
	mExt := shared.Extent{
		ExtentUUID:    common.StringPtr(uuid.New()),
		InputHostUUID: common.StringPtr(uuid.New()),
		StoreUUIDs:    make([]string, 0, 3),
	}
	mExtStats := shared.ExtentStats{
		Extent: &mExt,
		Status: common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN),
	}
	mListExtStats := metadata.NewListInputHostExtentsStatsResult_()
	mListExtStats.ExtentStatsList = []*shared.ExtentStats{&mExtStats}
	mockMeta := new(mockmeta.TChanMetadataService)
	mockMeta.On("ListInputHostExtentsStats", mock.Anything, mock.Anything).Return(mListExtStats, nil)
	destDesc := shared.DestinationDescription{
		DestinationUUID: common.StringPtr(uuid.New()),
	}
	mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(&destDesc, nil)
	inputHost, _ := NewInputHost(sName, s.mockService, mockMeta, nil)
	ctx, _ := utilGetThriftContextWithPath("foo")
	err := inputHost.OpenPublisherStream(ctx, s.mockPub)
	s.Error(err)
	assert.IsType(s.T(), &ReplicaNotExistsError{}, err)
}
func (s *InputHostSuite) TestInputHostPublishMessage() {
	count := 10
	inputHost, _ := NewInputHost("inputhost-test", s.mockService, s.mockMeta, nil)
	ctx, cancel := utilGetThriftContextWithPath("foo")
	defer cancel()
	s.mockAppend.On("Write", mock.Anything).Return(nil)
	ackCh := make(chan *cherami.InputHostCommand, count)
	s.mockPub.On("Write", mock.Anything).Return(
		func(ack *cherami.InputHostCommand) error {
			ackCh <- ack
			return nil
		})
	for i := 0; i < count; i++ {
		aMsg := store.NewAppendMessageAck()
		aMsg.SequenceNumber = common.Int64Ptr(int64(i + 1))
		aMsg.Status = common.CheramiStatusPtr(cherami.Status_OK)
		aMsg.Address = common.Int64Ptr(int64(i + 100))
		msg := cherami.NewPutMessage()
		msg.ID = common.StringPtr(strconv.Itoa(i))
		msg.Data = []byte(fmt.Sprintf("hello-%d", i))
		msg.UserContext = map[string]string{"UserMsgId": fmt.Sprintf("user-msg-%d", i)}
		s.mockAppend.On("Read").Return(aMsg, nil).Once()
		s.mockPub.On("Read").Return(msg, nil).Once()
	}
	s.mockAppend.On("Read").Return(nil, io.EOF)
	s.mockPub.On("Read").Return(nil, io.EOF)
	err := inputHost.OpenPublisherStream(ctx, s.mockPub)
	s.NoError(err)
	extStats, _ := s.mockMeta.ListInputHostExtentsStats(nil, nil)
	s.Len(extStats.GetExtentStatsList(), 1)
	extUUID := extStats.GetExtentStatsList()[0].Extent.ExtentUUID
	s.NotNil(extUUID)
	for i := 0; i < count; i++ {
		select {
		case ack := <-ackCh:
			s.Equal(cherami.InputHostCommandType_ACK, ack.GetType())
			s.Equal(strconv.Itoa(i), ack.GetAck().GetID())
			s.Equal(fmt.Sprintf("user-msg-%d", i), ack.GetAck().GetUserContext()["UserMsgId"])
			receipt := ack.GetAck().GetReceipt()
			parts := strings.Split(receipt, ":")
			s.Equal(*extUUID, parts[0])
			s.Equal(strconv.Itoa(i+1), parts[1])
			s.Equal(fmt.Sprintf("%8x", i+100), parts[2])
		default:
		}
	}
	inputHost.Shutdown()
}
func (s *InputHostSuite) _TestInputHostPublishMessageLogDestination() {
	*s.desinationDescription = *shared.NewDestinationDescription()
	s.desinationDescription.DestinationUUID = common.StringPtr(uuid.New())
	logType := shared.DestinationType_LOG
	s.desinationDescription.Type = &logType
	count := 10
	inputHost, _ := NewInputHost("inputhost-test", s.mockService, s.mockMeta, nil)
	ctx, cancel := utilGetThriftContextWithPath("foo")
	defer cancel()
	s.mockAppend.On("Write", mock.Anything).Return(nil)
	ackCh := make(chan *cherami.InputHostCommand, count)
	s.mockPub.On("Write", mock.Anything).Return(
		func(ack *cherami.InputHostCommand) error {
			ackCh <- ack
			return nil
		})
	for i := 0; i < count*2; i += 2 {
		aMsg := store.NewAppendMessageAck()
		aMsg.SequenceNumber = common.Int64Ptr(int64(i + 1))
		aMsg.Status = common.CheramiStatusPtr(cherami.Status_OK)
		msg := cherami.NewPutMessage()
		msg.ID = common.StringPtr(strconv.Itoa(i))
		msg.Data = []byte(fmt.Sprintf("hello-%d", i))
		msg.UserContext = map[string]string{"UserMsgId": fmt.Sprintf("user-msg-%d", i)}
		s.mockAppend.On("Read").Return(aMsg, nil).Once()
		aMsg2 := store.NewAppendMessageAck()
		aMsg2.SequenceNumber = common.Int64Ptr(int64(i + 2))
		aMsg2.Status = common.CheramiStatusPtr(cherami.Status_OK)
		s.mockAppend.On("Read").Return(aMsg2, nil).Once()
		s.mockPub.On("Read").Return(msg, nil).Once()
	}
	s.mockAppend.On("Read").Return(nil, io.EOF)
	s.mockPub.On("Read").Return(nil, io.EOF)
	err := inputHost.OpenPublisherStream(ctx, s.mockPub)
	s.NoError(err)
	extStats, _ := s.mockMeta.ListInputHostExtentsStats(nil, nil)
	s.Len(extStats.GetExtentStatsList(), 1)
	extUUID := extStats.GetExtentStatsList()[0].Extent.ExtentUUID
	s.NotNil(extUUID)
	for i := 0; i < count*2; i += 2 {
		select {
		case ack := <-ackCh:
			s.Equal(cherami.InputHostCommandType_ACK, ack.GetType())
			s.Equal(strconv.Itoa(i), ack.GetAck().GetID())
			s.Equal(fmt.Sprintf("user-msg-%d", i), ack.GetAck().GetUserContext()["UserMsgId"])
			receipt := ack.GetAck().GetReceipt()
			parts := strings.Split(receipt, ":")
			s.Equal(*extUUID, parts[0])
			s.Equal(strconv.Itoa(i/2+1), parts[1])
		}
	}
	inputHost.Shutdown()
}
func (s *InputHostSuite) TestInputHostMultipleClients() {
	numClients := 3
	numStoreStreams := 1
	numExtents := 1
	for i := 0; i < numClients; i++ {
		s.mockAppend.On("Write", mock.Anything).Return(nil).Once()
		expectedWatermark := int64(i + 1)  
		s.mockAppend.On("Write", mock.MatchedBy(func(m *store.AppendMessage) bool {
			return m.GetFullyReplicatedWatermark() == expectedWatermark
		})).Return(nil).Once()
	}
	inputHost, _ := NewInputHost("inputhost-test", s.mockService, s.mockMeta, nil)
	mockPub := make([]*mockin.MockBInOpenPublisherStreamInCall, numClients)
	for i := 0; i < numClients; i++ {
		mockPub[i] = new(mockin.MockBInOpenPublisherStreamInCall)
		mockPub[i].On("Write", mock.Anything).Return(nil)
		mockPub[i].On("Done").Return(nil)
		mockPub[i].On("Flush").Return(nil)
		aMsg := store.NewAppendMessageAck()
		aMsg.SequenceNumber = common.Int64Ptr(int64(i + 1))
		aMsg.Status = common.CheramiStatusPtr(cherami.Status_OK)
		msg := cherami.NewPutMessage()
		msg.ID = common.StringPtr(strconv.Itoa(i))
		msg.Data = []byte(fmt.Sprintf("hello-%d", i))
		s.mockAppend.On("Read").Return(aMsg, nil).Once()
		mockPub[i].On("Read").Return(msg, nil).Once()
		mockPub[i].On("Read").Return(nil, io.EOF).On("Write", mock.Anything).Return(nil)
		ctx, cancel := utilGetThriftContextWithPath("foo")
		defer cancel()
		err := inputHost.OpenPublisherStream(ctx, mockPub[i])
		s.NoError(err)
	}
	inputHost.pathMutex.RLock()
	s.Equal(numExtents, len(inputHost.pathCache))
	for _, pathCache := range inputHost.pathCache {
		pathCache.RLock()
		for _, extInfo := range pathCache.extentCache {
			s.Equal(numStoreStreams, len(extInfo.connection.streams))
		}
		pathCache.RUnlock()
	}
	inputHost.pathMutex.RUnlock()
	for i := 0; i < numClients; i++ {
		mockPub[i].On("Read").Return(nil, io.EOF).Once()
		mockPub[i].On("Write").Return(nil, io.EOF).Once()
	}
	s.mockAppend.On("Write", mock.Anything).Return(io.EOF)
	s.mockAppend.On("Read").Return(nil, io.EOF)
	inputHost.Shutdown()
}
func (s *InputHostSuite) TestInputHostCacheTime() {
	numExtents := 1
	inputHost, _ := NewInputHost("inputhost-test", s.mockService, s.mockMeta, &InOptions{
		CacheIdleTimeout: 2 * time.Second,
	})
	ctx, cancel := utilGetThriftContextWithPath("foo")
	defer cancel()
	s.mockAppend.On("Write", mock.Anything).Return(nil).Once()
	s.mockAppend.On("Read").Return(&store.AppendMessageAck{}, nil).Once()
	s.mockPub.On("Write", mock.Anything).Return(nil).Once()
	s.mockPub.On("Read").Return(&cherami.PutMessage{}, io.EOF).WaitUntil(time.After(3 * time.Second))
	err := inputHost.OpenPublisherStream(ctx, s.mockPub)
	s.NoError(err)
	time.Sleep(3 * time.Second)
	inputHost.pathMutex.RLock()
	s.Equal(numExtents, len(inputHost.pathCache))
	for _, pathCache := range inputHost.pathCache {
		pathCache.RLock()
		for _, conn := range pathCache.connections {
			conn.lk.Lock()
			s.Equal(false, conn.closed, "connection must be closed")
			conn.lk.Unlock()
		}
		pathCache.RUnlock()
	}
	inputHost.pathMutex.RUnlock()
	s.mockAppend.On("Write", mock.Anything).Return(io.EOF)
	s.mockAppend.On("Read").Return(nil, io.EOF)
	inputHost.Shutdown()
}
func (s *InputHostSuite) TestInputHostLoadUnloadRace() {
	numAttempts := 10
	inputHost, _ := NewInputHost("inputhost-test", s.mockService, s.mockMeta, nil)
	ctx, cancel := utilGetThriftContextWithPath("foo")
	defer cancel()
	s.mockAppend.On("Write", mock.Anything).Return(io.EOF).Times(numAttempts)
	s.mockAppend.On("Read").Return(&store.AppendMessageAck{}, io.EOF).Times(numAttempts)
	s.mockPub.On("Write", mock.Anything).Return(io.EOF).Times(numAttempts)
	s.mockPub.On("Read").Return(&cherami.PutMessage{}, io.EOF).Times(numAttempts)
	var wg sync.WaitGroup
	for i := 0; i < numAttempts; i++ {
		wg.Add(1)
		go func(waitG *sync.WaitGroup) {
			inputHost.OpenPublisherStream(ctx, s.mockPub)
			waitG.Done()
		}(&wg)
	}
	time.Sleep(1 * time.Second)
	pathCache, _ := inputHost.getPathCacheByDestPath("foo")
	s.NotNil(pathCache)
	inputHost.unloadAll()
	s.Equal(false, pathCache.isActiveNoLock(), "pathCache should not be active")
	wg.Wait()
	inputHost.Shutdown()
}
func (s *InputHostSuite) TestInputHostReconfigure() {
	destUUID := uuid.New()
	extUUID1 := uuid.New()
	extUUID2 := uuid.New()
	hostUUID := uuid.New()
	ch, _ := tchannel.NewChannel("cherami-test-in", nil)
	mockService := new(common.MockService)
	mockService.On("GetConfig").Return(s.cfg.GetServiceConfig(common.InputServiceName))
	mockService.On("GetMetricsReporter").Return(common.NewMetricReporterWithHostname(configure.NewCommonServiceConfig()))
	mockService.On("GetDConfigClient").Return(dconfig.NewDconfigClient(configure.NewCommonServiceConfig(), common.InputServiceName))
	mockService.On("GetTChannel").Return(ch)
	mockService.On("IsLimitsEnabled").Return(true)
	mockService.On("GetHostConnLimit").Return(100)
	mockService.On("GetHostConnLimitPerSecond").Return(100)
	mockService.On("GetWSConnector").Return(s.mockWSConnector, nil)
	mockMeta := new(mockmeta.TChanMetadataService)
	s.mockAppend.On("Write", mock.Anything).Return(nil)
	s.mockPub.On("Write", mock.Anything).Return(nil)
	aMsg := store.NewAppendMessageAck()
	msg := cherami.NewPutMessage()
	appendTicker := time.NewTicker(5 * time.Second)
	defer appendTicker.Stop()
	pubTicker := time.NewTicker(5 * time.Second)
	defer pubTicker.Stop()
	s.mockAppend.On("Read").Return(aMsg, io.EOF).WaitUntil(appendTicker.C)
	s.mockPub.On("Read").Return(msg, io.EOF).WaitUntil(pubTicker.C)
	s.mockStore.On("OpenAppendStream", mock.Anything).Return(s.mockAppend, nil)
	destDesc := shared.NewDestinationDescription()
	destDesc.DestinationUUID = common.StringPtr(destUUID)
	mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(destDesc, nil)
	mExt := shared.NewExtent()
	mExt.ExtentUUID = common.StringPtr(extUUID1)
	mExt.InputHostUUID = common.StringPtr(hostUUID)
	mExt.StoreUUIDs = make([]string, 0, 4)
	rpm := common.NewMockRingpopMonitor()
	rpm.Add(common.InputServiceName, mExt.GetInputHostUUID(), "127.0.0.1")
	for i := 1; i < 2; i++ {
		uuid := uuid.New()
		rpm.Add(common.StoreServiceName, uuid, fmt.Sprintf("127.0.2.%v", i))
		mExt.StoreUUIDs = append(mExt.StoreUUIDs, uuid)
	}
	mockLoadReporterDaemonFactory := setupMockLoadReporterDaemonFactory()
	mockService.On("GetLoadReporterDaemonFactory").Return(mockLoadReporterDaemonFactory)
	mockService.On("GetHostUUID").Return("99999999-9999-9999-9999-999999999999")
	mockService.On("GetRingpopMonitor").Return(rpm)
	mockService.On("GetClientFactory").Return(s.mockClientFactory)
	mExtStats := shared.NewExtentStats()
	mExtStats.Extent = mExt
	mExtStats.Status = common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN)
	mListExtStats := metadata.NewListInputHostExtentsStatsResult_()
	mListExtStats.ExtentStatsList = append(mListExtStats.ExtentStatsList, mExtStats)
	mockMeta.On("ListInputHostExtentsStats", mock.Anything, mock.Anything).Return(mListExtStats, nil)
	inputHost, _ := NewInputHost("inputhost-test", mockService, mockMeta, nil)
	ctx, cancel := utilGetThriftContextWithPath("foo")
	defer cancel()
	go func() {
		inputHost.OpenPublisherStream(ctx, s.mockPub)
	}()
	time.Sleep(1 * time.Second)
	pathCache, ok := inputHost.getPathCacheByDestPath("foo")
	s.True(ok)
	pathCache.Lock()
	s.Equal(1, len(pathCache.extentCache))
	pathCache.Unlock()
	mExt1 := shared.NewExtent()
	mExt1.ExtentUUID = common.StringPtr(extUUID2)
	mExt1.InputHostUUID = common.StringPtr(hostUUID)
	mExt1.StoreUUIDs = make([]string, 0, 4)
	for i := 1; i < 2; i++ {
		uuid := uuid.New()
		rpm.Add(common.StoreServiceName, uuid, fmt.Sprintf("127.0.2.%v", i))
		mExt1.StoreUUIDs = append(mExt1.StoreUUIDs, uuid)
	}
	mExtStats1 := shared.NewExtentStats()
	mExtStats1.Extent = mExt1
	mExtStats1.Status = common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN)
	mListExtStats.ExtentStatsList = append(mListExtStats.ExtentStatsList, mExtStats1)
	mockMeta.On("ListInputHostExtentsStats", mock.Anything, mock.Anything).Return(mListExtStats, nil)
	reconfigType := admin.NotificationType_ALL
	req := admin.NewDestinationsUpdatedRequest()
	req.UpdateUUID = common.StringPtr(uuid.New())
	destUpdate := admin.NewDestinationUpdatedNotification()
	destUpdate.DestinationUUID = common.StringPtr(destUUID)
	destUpdate.Type = &reconfigType
	req.Updates = append(req.Updates, destUpdate)
	err := inputHost.DestinationsUpdated(ctx, req)
	s.NoError(err)
	time.Sleep(1 * time.Second)
	pathCache, ok = inputHost.getPathCacheByDestPath("foo")
	if ok {
		pathCache.Lock()
		s.Equal(2, len(pathCache.extentCache))
		pathCache.Unlock()
	}
	inputHost.Shutdown()
}
func (s *InputHostSuite) TestFailedOpenAppendStream() {
	ch, _ := tchannel.NewChannel("cherami-test-in", nil)
	mockService := new(common.MockService)
	mockService.On("GetConfig").Return(s.cfg.GetServiceConfig(common.InputServiceName))
	mockService.On("GetMetricsReporter").Return(common.NewMetricReporterWithHostname(configure.NewCommonServiceConfig()))
	mockService.On("GetDConfigClient").Return(dconfig.NewDconfigClient(configure.NewCommonServiceConfig(), common.InputServiceName))
	mockService.On("GetTChannel").Return(ch)
	mockService.On("IsLimitsEnabled").Return(true)
	mockService.On("GetHostConnLimit").Return(100)
	mockService.On("GetHostConnLimitPerSecond").Return(100)
	mockService.On("GetHostUUID").Return("99999999-9999-9999-9999-999999999998")
	mockWSConnector := new(mockcommon.MockWSConnector)
	mockService.On("GetWSConnector").Return(mockWSConnector, nil)
	mockWSConnector.On("OpenAppendStream", mock.Anything, mock.Anything).Return(s.mockAppend, fmt.Errorf("no store found"))
	mockClientFactory := new(mockcommon.MockClientFactory)
	mockService.On("GetClientFactory").Return(mockClientFactory)
	mockMeta := new(mockmeta.TChanMetadataService)
	destDesc := shared.NewDestinationDescription()
	destDesc.DestinationUUID = common.StringPtr(uuid.New())
	mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(destDesc, nil)
	mExt := shared.NewExtent()
	mExt.ExtentUUID = common.StringPtr(uuid.New())
	mExt.InputHostUUID = common.StringPtr("99999999-9999-9999-9999-999999999998")
	mExt.StoreUUIDs = make([]string, 0, 4)
	rpm := common.NewMockRingpopMonitor()
	rpm.Add(common.InputServiceName, mExt.GetInputHostUUID(), "127.0.0.1")
	for i := 1; i < 2; i++ {
		uuid := uuid.New()
		rpm.Add(common.StoreServiceName, uuid, fmt.Sprintf("127.0.2.%v", i))
		mExt.StoreUUIDs = append(mExt.StoreUUIDs, uuid)
	}
	mExtStats := shared.NewExtentStats()
	mExtStats.Extent = mExt
	mExtStats.Status = common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN)
	mListExtStats := metadata.NewListInputHostExtentsStatsResult_()
	mListExtStats.ExtentStatsList = append(mListExtStats.ExtentStatsList, mExtStats)
	mockMeta.On("ListInputHostExtentsStats", mock.Anything, mock.Anything).Return(mListExtStats, nil)
	mockLoadReporterDaemonFactory := setupMockLoadReporterDaemonFactory()
	mockService.On("GetLoadReporterDaemonFactory").Return(mockLoadReporterDaemonFactory)
	mockService.On("GetRingpopMonitor").Return(rpm)
	mockService.On("GetClientFactory").Return(mockClientFactory)
	inputHost, _ := NewInputHost("inputhost-test", mockService, mockMeta, nil)
	ctx, cancel := utilGetThriftContextWithPath("foo")
	defer cancel()
	s.mockPub.On("Read").Return(nil, io.EOF)
	err := inputHost.OpenPublisherStream(ctx, s.mockPub)
	s.Error(err)
	inputHost.Shutdown()
}
func (s *InputHostSuite) _TestInputHostPutMessageBatch() {
	succMsgCount := 5
	failMsgCount := 5
	totalMsgCount := succMsgCount + failMsgCount
	extUUID := uuid.New()
	destinationPath := "foo"
	ch, _ := tchannel.NewChannel("cherami-test-in", nil)
	mockService := new(common.MockService)
	mockService.On("GetConfig").Return(s.cfg.GetServiceConfig(common.InputServiceName))
	mockService.On("GetDConfigClient").Return(dconfig.NewDconfigClient(configure.NewCommonServiceConfig(), common.InputServiceName))
	mockService.On("GetMetricsReporter").Return(common.NewMetricReporterWithHostname(configure.NewCommonServiceConfig()))
	mockService.On("GetTChannel").Return(ch)
	mockService.On("IsLimitsEnabled").Return(false)
	mockService.On("GetHostConnLimit").Return(100)
	mockService.On("GetHostConnLimitPerSecond").Return(100)
	mockService.On("GetHostUUID").Return("99999999-9999-9999-9999-999999999997")
	mockClientFactory := new(mockcommon.MockClientFactory)
	mockService.On("GetClientFactory").Return(mockClientFactory)
	mockLoadReporterDaemonFactory := setupMockLoadReporterDaemonFactory()
	mockService.On("GetLoadReporterDaemonFactory").Return(mockLoadReporterDaemonFactory)
	mockMeta := new(mockmeta.TChanMetadataService)
	destDesc := shared.NewDestinationDescription()
	destDesc.DestinationUUID = common.StringPtr(uuid.New())
	mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(destDesc, nil)
	mExt := shared.NewExtent()
	mExt.ExtentUUID = common.StringPtr(extUUID)
	mExt.InputHostUUID = common.StringPtr("99999999-9999-9999-9999-999999999997")
	mExt.StoreUUIDs = make([]string, 0, 4)
	rpm := common.NewMockRingpopMonitor()
	rpm.Add(common.InputServiceName, mExt.GetInputHostUUID(), "127.0.0.1")
	for i := 1; i < 2; i++ {
		uuid := uuid.New()
		rpm.Add(common.StoreServiceName, uuid, fmt.Sprintf("127.0.2.%v", i))
		mExt.StoreUUIDs = append(mExt.StoreUUIDs, uuid)
	}
	mExtStats := shared.NewExtentStats()
	mExtStats.Extent = mExt
	mExtStats.Status = common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN)
	mListExtStats := metadata.NewListInputHostExtentsStatsResult_()
	mListExtStats.ExtentStatsList = append(mListExtStats.ExtentStatsList, mExtStats)
	mockMeta.On("ListInputHostExtentsStats", mock.Anything, mock.Anything).Return(mListExtStats, nil)
	mockService.On("GetRingpopMonitor").Return(rpm)
	mockService.On("GetClientFactory").Return(mockClientFactory)
	mockClientFactory.On("GetThriftStoreClient", mock.Anything, mock.Anything).Return(store.TChanBStore(s.mockStore), nil)
	mockClientFactory.On("ReleaseThriftStoreClient", mock.Anything).Return(nil)
	mockClientFactory.On("GetAdminControllerClient").Return(nil, fmt.Errorf("no real controller"))
	s.mockStore.On("OpenAppendStream", mock.Anything).Return(s.mockAppend, nil).Once()
	s.mockAppend.On("Write", mock.Anything).Return(nil)
	inputHost, _ := NewInputHost("inputhost-test", mockService, mockMeta, nil)
	ctx, cancel := utilGetThriftContextWithPath(destinationPath)
	defer cancel()
	var wg sync.WaitGroup
	var putMessageAcks *cherami.PutMessageBatchResult_
	createMessage := func(seq int, sts cherami.Status) (*store.AppendMessageAck, *cherami.PutMessage) {
		aMsg := store.NewAppendMessageAck()
		aMsg.SequenceNumber = common.Int64Ptr(int64(seq + 1))
		aMsg.Status = common.CheramiStatusPtr(sts)
		msg := cherami.NewPutMessage()
		msg.ID = common.StringPtr(strconv.Itoa(seq))
		msg.Data = []byte(fmt.Sprintf("hello-%d", seq))
		return aMsg, msg
	}
	putMessageRequest := &cherami.PutMessageBatchRequest{DestinationPath: &destinationPath}
	for i := 0; i < succMsgCount; i++ {
		aMsg, msg := createMessage(i, cherami.Status_OK)
		s.mockAppend.On("Read").Return(aMsg, nil).Once()
		putMessageRequest.Messages = append(putMessageRequest.Messages, msg)
	}
	for i := succMsgCount; i < totalMsgCount; i++ {
		aMsg, msg := createMessage(i, cherami.Status_FAILED)
		s.mockAppend.On("Read").Return(aMsg, nil).Once()
		putMessageRequest.Messages = append(putMessageRequest.Messages, msg)
	}
	wg.Add(1)
	go func() {
		var err error
		putMessageAcks, err = inputHost.PutMessageBatch(ctx, putMessageRequest)
		s.NoError(err)
		wg.Done()
	}()
	wg.Wait()
	s.mockAppend.On("Read").Return(nil, io.EOF)
	s.Len(putMessageAcks.GetSuccessMessages(), succMsgCount)
	s.Len(putMessageAcks.GetFailedMessages(), failMsgCount)
	pathCache, ok := inputHost.getPathCacheByDestPath(destinationPath)
	s.Equal(true, ok, "destination should be in the resolver cache now")
	s.NotNil(pathCache)
	inputHost.pathMutex.Lock()
	pathCache.Lock()
	pathCache.prepareForUnload()
	pathCache.Unlock()
	inputHost.pathMutex.Unlock()
	pathCache.unload()
	_, ok = inputHost.getPathCacheByDestUUID(destDesc.GetDestinationUUID())
	s.Equal(false, ok, "destination should now be gone from the resolver cache now")
	_, ok = inputHost.getPathCacheByDestPath(destinationPath)
	s.Equal(false, ok, "destination should now be gone from the resolver cache now")
	inputHost.Shutdown()
}