func (conn *pubConnection) SetMsgsLimitPerSecond(connLimit int32) {
	atomic.StoreInt32(&conn.connMsgsLimitPerSecond, connLimit)
	conn.SetConnTokenBucketValue(int32(connLimit))
}
SetMsgsLimitPerSecond sets msgs rate limit per second for this connection
package inputhost
import (
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/common/throttler"
	"github.com/uber/cherami-server/services/inputhost/load"
	serverStream "github.com/uber/cherami-server/stream"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
)
const logHighLatencyThreshold = time.Second
var logThrottler = throttler.New(1, 5*time.Second)  
var logsThrottled int32
type (
	pubConnection struct {
		connID              connectionID
		destinationPath     string
		stream              serverStream.BInOpenPublisherStreamInCall
		logger              bark.Logger
		reconfigureClientCh chan *reconfigInfo
		putMsgCh            chan *inPutMessage
		cacheTimeout        time.Duration
		ackChannel          chan *cherami.PutMessageAck
		replyCh             chan response
		closeChannel        chan struct{}  
		waitWG              sync.WaitGroup
		notifyCloseCh       chan connectionID  
		doneCh              chan bool          
		recvMsgs      int64  
		sentAcks      int64  
		sentNacks     int64  
		sentThrottled int64  
		failedMsgs    int64  
		connTokenBucketValue   atomic.Value  
		connMsgsLimitPerSecond int32         
		lk                     sync.Mutex
		opened                 bool
		closed                 bool
		limitsEnabled          bool
		pathCache              *inPathCache
		pathWG                 *sync.WaitGroup
	}
	response struct {
		ackID       string             
		userContext map[string]string  
		putMsgRecvTime time.Time  
	}
	inPutMessage struct {
		putMsg         *cherami.PutMessage
		putMsgAckCh    chan *cherami.PutMessageAck
		putMsgRecvTime time.Time
	}
	earlyReplyAck struct {
		ackReceiveTime time.Time
		ackSentTime time.Time
	}
	reconfigInfo struct {
		updateUUID string
		cmdType    cherami.InputHostCommandType
		drainWG    *sync.WaitGroup
	}
	pubConnectionClosedCb func(connectionID)
)
const failTimeout = 3 * time.Second
const reconfigClientChSize = 50
const perConnMsgsLimitPerSecond = 10000
func newPubConnection(destinationPath string, stream serverStream.BInOpenPublisherStreamInCall, pathCache *inPathCache, m3Client metrics.Client, limitsEnabled bool, timeout time.Duration, doneCh chan bool) *pubConnection {
	conn := &pubConnection{
		connID:          pathCache.currID,
		destinationPath: destinationPath,
		logger: pathCache.logger.WithFields(bark.Fields{
			common.TagInPubConnID: common.FmtInPubConnID(int(pathCache.currID)),
			common.TagModule:      `pubConn`,
		}),
		stream:       stream,
		putMsgCh:     pathCache.putMsgCh,
		cacheTimeout: timeout,
		replyCh:             make(chan response, defaultBufferSize),
		reconfigureClientCh: make(chan *reconfigInfo, reconfigClientChSize),
		ackChannel:          make(chan *cherami.PutMessageAck, defaultBufferSize),
		closeChannel:        make(chan struct{}),
		notifyCloseCh:       pathCache.notifyConnsCloseCh,
		doneCh:              doneCh,
		limitsEnabled:       limitsEnabled,
		pathCache:           pathCache,
		pathWG:              &pathCache.connsWG,
	}
	conn.SetMsgsLimitPerSecond(common.HostPerConnMsgsLimitPerSecond)
	return conn
}
func (conn *pubConnection) open() {
	conn.lk.Lock()
	defer conn.lk.Unlock()
	if !conn.opened {
		conn.waitWG.Add(2)
		conn.pathWG.Add(1)  
		go conn.readRequestStream()
		go conn.writeAcksStream()
		conn.opened = true
		conn.logger.Info("pubConn opened")
	}
}
func (conn *pubConnection) close() {
	conn.lk.Lock()
	if conn.closed {
		conn.lk.Unlock()
		return
	}
	close(conn.closeChannel)
	conn.closed = true
	conn.waitWG.Wait()
	select {
	case conn.doneCh <- true:
	default:
	}
	conn.lk.Unlock()
	conn.notifyCloseCh <- conn.connID
	conn.pathWG.Done()
	conn.logger.WithFields(bark.Fields{
		`sentAcks`:      conn.sentAcks,
		`sentNacks`:     conn.sentNacks,
		`sentThrottled`: conn.sentThrottled,
		`failedMsgs`:    conn.failedMsgs,
	}).Info("pubConn closed")
}
func (conn *pubConnection) readRequestStream() {
	defer conn.waitWG.Done()
	var msgLength int64
	connIdleTimer := common.NewTimer(conn.cacheTimeout)
	defer connIdleTimer.Stop()
	for {
		connIdleTimer.Reset(conn.cacheTimeout)
		select {
		case <-connIdleTimer.C:
			conn.logger.WithField(`cacheTimeout`, conn.cacheTimeout).Info(`client connection idle for : ; closing it`)
			go conn.close()
			return
		default:
			msg, err := conn.stream.Read()
			if err != nil {
				conn.logger.WithField(common.TagErr, err).Info(`inputhost: PublisherStream closed on read error`)
				go conn.close()
				return
			}
			conn.pathCache.m3Client.IncCounter(metrics.PubConnectionStreamScope, metrics.InputhostMessageReceived)
			conn.pathCache.destM3Client.IncCounter(metrics.PubConnectionScope, metrics.InputhostDestMessageReceived)
			conn.pathCache.dstMetrics.Increment(load.DstMetricOverallNumMsgs)
			msgLength = int64(len(msg.Data))
			conn.pathCache.dstMetrics.Add(load.DstMetricBytesIn, msgLength)
			conn.pathCache.hostMetrics.Add(load.HostMetricBytesIn, msgLength)
			conn.pathCache.destM3Client.AddCounter(metrics.PubConnectionScope, metrics.InputhostDestMessageReceivedBytes, msgLength)
			conn.recvMsgs++
			inMsg := &inPutMessage{
				putMsg:         msg,
				putMsgAckCh:    conn.ackChannel,
				putMsgRecvTime: time.Now(),
			}
			throttled := false
			if conn.limitsEnabled {
				consumed, _ := conn.GetConnTokenBucketValue().TryConsume(1)
				throttled = !consumed
				if throttled {
					conn.logger.Warn("throttling due to rate violation")
					conn.pathCache.m3Client.IncCounter(metrics.PubConnectionStreamScope, metrics.InputhostMessageLimitThrottled)
					conn.pathCache.destM3Client.IncCounter(metrics.PubConnectionScope, metrics.InputhostDestMessageLimitThrottled)
					inMsg.putMsgAckCh <- &cherami.PutMessageAck{
						ID:          common.StringPtr(msg.GetID()),
						UserContext: msg.GetUserContext(),
						Status:      common.CheramiStatusPtr(cherami.Status_THROTTLED),
						Message:     common.StringPtr("throttling; inputhost is busy"),
					}
				}
			}
			if !throttled {
				if conn.limitsEnabled {
					select {
					case conn.putMsgCh <- inMsg:
						conn.replyCh <- response{msg.GetID(), msg.GetUserContext(), inMsg.putMsgRecvTime}
					default:
						conn.pathCache.m3Client.IncCounter(metrics.PubConnectionStreamScope, metrics.InputhostMessageChannelFullThrottled)
						conn.pathCache.destM3Client.IncCounter(metrics.PubConnectionScope, metrics.InputhostDestMessageChannelFullThrottled)
						conn.logger.Warn("throttling due to putMsgCh being filled")
						inMsg.putMsgAckCh <- &cherami.PutMessageAck{
							ID:          common.StringPtr(msg.GetID()),
							UserContext: msg.GetUserContext(),
							Status:      common.CheramiStatusPtr(cherami.Status_THROTTLED),
							Message:     common.StringPtr("throttling; inputhost is busy"),
						}
					}
				} else {
					select {
					case conn.putMsgCh <- inMsg:
						conn.replyCh <- response{msg.GetID(), msg.GetUserContext(), inMsg.putMsgRecvTime}
					case <-conn.closeChannel:
						return
					}
				}
			}
		}
	}
}
func (conn *pubConnection) writeAcksStream() {
	earlyReplyAcks := make(map[string]earlyReplyAck)  
	inflightMessages := make(map[string]response)
	defer conn.failInflightMessages(inflightMessages, earlyReplyAcks)
	flushTicker := time.NewTicker(common.FlushTimeout)  
	defer flushTicker.Stop()
	unflushedWrites := 0
	for {
		select {
		case resCh := <-conn.replyCh:
			if _, ok := earlyReplyAcks[resCh.ackID]; ok {
				conn.updateEarlyReplyAcks(resCh, earlyReplyAcks)
			} else {
				inflightMessages[resCh.ackID] = resCh
			}
		case <-flushTicker.C:
			if unflushedWrites > 0 {
				if err := conn.flushCmdToClient(unflushedWrites); err != nil {
					go conn.close()
				}
				unflushedWrites = 0
			}
		default:
			if len(inflightMessages) == 0 {
				select {
				case resCh := <-conn.replyCh:
					if _, ok := earlyReplyAcks[resCh.ackID]; ok {
						conn.updateEarlyReplyAcks(resCh, earlyReplyAcks)
					} else {
						inflightMessages[resCh.ackID] = resCh
					}
				case rInfo := <-conn.reconfigureClientCh:
					conn.pathCache.m3Client.IncCounter(metrics.PubConnectionStreamScope, metrics.InputhostReconfClientRequests)
					cmd := createReconfigureCmd(rInfo)
					if err := conn.writeCmdToClient(cmd, rInfo.drainWG); err != nil {
						go conn.close()
						return
					}
					unflushedWrites++
				case <-flushTicker.C:
					if unflushedWrites > 0 {
						if err := conn.flushCmdToClient(unflushedWrites); err != nil {
							go conn.close()
							return
						}
						unflushedWrites = 0
					}
				case <-conn.closeChannel:
					return
				}
			} else {
				select {
				case ack, ok := <-conn.ackChannel:
					if ok {
						ackReceiveTime := time.Now()
						exists, err := conn.writeAckToClient(inflightMessages, ack, ackReceiveTime)
						if err != nil {
							go conn.close()
							return
						}
						if !exists {
							earlyReplyAcks[ack.GetID()] = earlyReplyAck{
								ackReceiveTime: ackReceiveTime,
								ackSentTime:    time.Now(),
							}
						}
						unflushedWrites++
						if unflushedWrites > common.FlushThreshold {
							if err = conn.flushCmdToClient(unflushedWrites); err != nil {
								go conn.close()
								return
							}
							unflushedWrites = 0
						}
					} else {
						return
					}
				case rInfo := <-conn.reconfigureClientCh:
					conn.pathCache.m3Client.IncCounter(metrics.PubConnectionStreamScope, metrics.InputhostReconfClientRequests)
					cmd := createReconfigureCmd(rInfo)
					if err := conn.writeCmdToClient(cmd, rInfo.drainWG); err != nil {
						go conn.close()
						return
					}
					unflushedWrites++
				case <-flushTicker.C:
					if unflushedWrites > 0 {
						if err := conn.flushCmdToClient(unflushedWrites); err != nil {
							go conn.close()
							return
						}
						unflushedWrites = 0
					}
				case <-conn.closeChannel:
					return
				}
			}
		}
	}
}
func (conn *pubConnection) failInflightMessages(inflightMessages map[string]response, earlyReplyAcks map[string]earlyReplyAck) {
	defer conn.stream.Done()
	failTimer := common.NewTimer(failTimeout)
	defer failTimer.Stop()
	for quit := false; !quit && len(inflightMessages) > 0; {
		select {
		case ack, ok := <-conn.ackChannel:
			if ok {
				conn.writeAckToClient(inflightMessages, ack, time.Now())
			}
		case resCh := <-conn.replyCh:
			if _, ok := earlyReplyAcks[resCh.ackID]; ok {
				conn.updateEarlyReplyAcks(resCh, earlyReplyAcks)
			} else {
				inflightMessages[resCh.ackID] = resCh
			}
		case <-failTimer.C:
			conn.logger.WithField(`inflightMessages`, len(inflightMessages)).Info(`inputhost: timing out messages`)
			quit = true
		}
	}
	for id, resp := range inflightMessages {
		if _, ok := earlyReplyAcks[id]; !ok {
			putMsgAck := &cherami.PutMessageAck{
				ID:          common.StringPtr(id),
				UserContext: resp.userContext,
				Status:      common.CheramiStatusPtr(cherami.Status_FAILED),
				Message:     common.StringPtr("inputhost: timing out unacked message"),
			}
			d := time.Since(resp.putMsgRecvTime)
			conn.pathCache.m3Client.RecordTimer(metrics.PubConnectionStreamScope, metrics.InputhostWriteMessageBeforeAckLatency, d)
			conn.pathCache.destM3Client.RecordTimer(metrics.PubConnectionScope, metrics.InputhostDestWriteMessageBeforeAckLatency, d)
			conn.stream.Write(createAckCmd(putMsgAck))
			d = time.Since(resp.putMsgRecvTime)
			conn.pathCache.m3Client.RecordTimer(metrics.PubConnectionStreamScope, metrics.InputhostWriteMessageLatency, d)
			conn.pathCache.destM3Client.RecordTimer(metrics.PubConnectionScope, metrics.InputhostDestWriteMessageLatency, d)
			if d > logHighLatencyThreshold {
				if logThrottler.Allow() {
					conn.logger.WithFields(bark.Fields{
						common.TagDstPth:     common.FmtDstPth(conn.destinationPath),
						common.TagInPutAckID: common.FmtInPutAckID(id),
						`duration`:           d,
						`putMsgChanLen`:      len(conn.putMsgCh),
						`putMsgAckChanLen`:   len(conn.ackChannel),
						`replyChanLen`:       len(conn.replyCh),
						`throttled`:          atomic.SwapInt32(&logsThrottled, 0),
					}).Error(`failInflightMessages: publish message latency`)
				} else {
					atomic.AddInt32(&logsThrottled, 1)
				}
			}
			conn.pathCache.m3Client.IncCounter(metrics.PubConnectionStreamScope, metrics.InputhostMessageFailures)
			conn.pathCache.destM3Client.IncCounter(metrics.PubConnectionScope, metrics.InputhostDestMessageFailures)
			conn.failedMsgs++
			conn.pathCache.dstMetrics.Increment(load.DstMetricNumFailed)
		}
	}
	conn.stream.Flush()
	conn.waitWG.Done()
}
func (conn *pubConnection) flushCmdToClient(unflushedWrites int) (err error) {
	if err = conn.stream.Flush(); err != nil {
		conn.logger.WithFields(bark.Fields{common.TagErr: err, `unflushedWrites`: unflushedWrites}).Error(`inputhost: error flushing messages to client stream failed`)
	}
	return
}
func (conn *pubConnection) writeCmdToClient(cmd *cherami.InputHostCommand, drainWG *sync.WaitGroup) (err error) {
	if err = conn.stream.Write(cmd); err != nil {
		conn.logger.WithFields(bark.Fields{`cmd`: cmd, common.TagErr: err}).Info(`inputhost: Unable to Write cmd back to client`)
	}
	if cmd.GetType() == cherami.InputHostCommandType_DRAIN {
		go conn.waitForDrain(drainWG)
	}
	return
}
func (conn *pubConnection) waitForDrain(drainWG *sync.WaitGroup) {
	connWGTimer := common.NewTimer(connWGTimeout)
	defer connWGTimer.Stop()
	select {
	case <-conn.closeChannel:
		drainWG.Done()
		return
	case <-connWGTimer.C:
		drainWG.Done()
	}
	drainTimer := common.NewTimer(defaultDrainTimeout)
	defer drainTimer.Stop()
	select {
	case <-conn.closeChannel:
		return
	case <-drainTimer.C:
		conn.logger.Warn("timed out waiting for drain to finish. just closing the connection")
		go conn.close()
	}
}
func (conn *pubConnection) writeAckToClient(inflightMessages map[string]response, ack *cherami.PutMessageAck, ackReceiveTime time.Time) (exists bool, err error) {
	cmd := createAckCmd(ack)
	err = conn.writeCmdToClient(cmd, nil)
	if err != nil {
		conn.logger.
			WithField(common.TagInPutAckID, common.FmtInPutAckID(ack.GetID())).
			WithField(common.TagErr, err).Error(`inputhost: writing ack Id failed`)
	}
	exists = conn.updateInflightMap(inflightMessages, ack.GetID(), ackReceiveTime)
	if ack.GetStatus() == cherami.Status_FAILED || err != nil {
		conn.pathCache.m3Client.IncCounter(metrics.PubConnectionStreamScope, metrics.InputhostMessageFailures)
		conn.pathCache.destM3Client.IncCounter(metrics.PubConnectionScope, metrics.InputhostDestMessageFailures)
	}
	if err == nil {
		switch ack.GetStatus() {
		case cherami.Status_OK:
			conn.sentAcks++
			conn.pathCache.dstMetrics.Increment(load.DstMetricNumAcks)
		case cherami.Status_FAILED:
			conn.sentNacks++
			conn.pathCache.dstMetrics.Increment(load.DstMetricNumNacks)
		case cherami.Status_THROTTLED:
			conn.sentThrottled++
			conn.pathCache.dstMetrics.Increment(load.DstMetricNumThrottled)
		}
	}
	return
}
func (conn *pubConnection) updateInflightMap(inflightMessages map[string]response, ackID string, ackReceiveTime time.Time) bool {
	if resp, ok := inflightMessages[ackID]; ok {
		d := time.Since(resp.putMsgRecvTime)
		conn.pathCache.m3Client.RecordTimer(metrics.PubConnectionStreamScope, metrics.InputhostWriteMessageLatency, d)
		conn.pathCache.destM3Client.RecordTimer(metrics.PubConnectionScope, metrics.InputhostDestWriteMessageLatency, d)
		conn.pathCache.m3Client.RecordTimer(metrics.PubConnectionStreamScope, metrics.InputhostWriteMessageBeforeAckLatency, ackReceiveTime.Sub(resp.putMsgRecvTime))
		conn.pathCache.destM3Client.RecordTimer(metrics.PubConnectionScope, metrics.InputhostDestWriteMessageBeforeAckLatency, ackReceiveTime.Sub(resp.putMsgRecvTime))
		if d > logHighLatencyThreshold {
			if logThrottler.Allow() {
				conn.logger.WithFields(bark.Fields{
					common.TagDstPth:     common.FmtDstPth(conn.destinationPath),
					common.TagInPutAckID: common.FmtInPutAckID(ackID),
					`d`:                  d,
					`putMsgChanLen`:      len(conn.putMsgCh),
					`putMsgAckChanLen`:   len(conn.ackChannel),
					`replyChanLen`:       len(conn.replyCh),
					`throttled`:          atomic.SwapInt32(&logsThrottled, 0),
				}).Info(`publish message latency at updateInflightMap`)
			} else {
				atomic.AddInt32(&logsThrottled, 1)
			}
		}
		delete(inflightMessages, ackID)
		return ok
	}
	return false
}
func (conn *pubConnection) updateEarlyReplyAcks(resCh response, earlyReplyAcks map[string]earlyReplyAck) {
	d := time.Since(resCh.putMsgRecvTime)
	ack, _ := earlyReplyAcks[resCh.ackID]
	actualDuration := d - time.Since(ack.ackSentTime)
	if d > logHighLatencyThreshold {
		if logThrottler.Allow() {
			conn.logger.WithFields(bark.Fields{
				common.TagDstPth:     common.FmtDstPth(conn.destinationPath),
				common.TagInPutAckID: common.FmtInPutAckID(resCh.ackID),
				`d`:                  d,
				`actualDuration`:     actualDuration,
				`putMsgChanLen`:      len(conn.putMsgCh),
				`putMsgAckChanLen`:   len(conn.ackChannel),
				`replyChanLen`:       len(conn.replyCh),
				`throttled`:          atomic.SwapInt32(&logsThrottled, 0),
			}).Info(`publish message latency at updateEarlyReplyAcks and actualDuration`)
		} else {
			atomic.AddInt32(&logsThrottled, 1)
		}
	}
	conn.pathCache.m3Client.RecordTimer(metrics.PubConnectionStreamScope, metrics.InputhostWriteMessageLatency, actualDuration)
	conn.pathCache.destM3Client.RecordTimer(metrics.PubConnectionScope, metrics.InputhostDestWriteMessageLatency, actualDuration)
	actualDurationExcludeSocket := d - time.Since(ack.ackReceiveTime)
	conn.pathCache.m3Client.RecordTimer(metrics.PubConnectionStreamScope, metrics.InputhostWriteMessageBeforeAckLatency, actualDurationExcludeSocket)
	conn.pathCache.destM3Client.RecordTimer(metrics.PubConnectionScope, metrics.InputhostDestWriteMessageBeforeAckLatency, actualDurationExcludeSocket)
	delete(earlyReplyAcks, resCh.ackID)
}
func createReconfigureCmd(rInfo *reconfigInfo) *cherami.InputHostCommand {
	cmd := cherami.NewInputHostCommand()
	cmd.Reconfigure = &cherami.ReconfigureInfo{UpdateUUID: common.StringPtr(rInfo.updateUUID)}
	cmd.Type = common.CheramiInputHostCommandTypePtr(rInfo.cmdType)
	return cmd
}
func createAckCmd(ack *cherami.PutMessageAck) *cherami.InputHostCommand {
	cmd := cherami.NewInputHostCommand()
	cmd.Ack = ack
	cmd.Type = common.CheramiInputHostCommandTypePtr(cherami.InputHostCommandType_ACK)
	return cmd
}
func (conn *pubConnection) GetMsgsLimitPerSecond() int {
	return int(atomic.LoadInt32(&conn.connMsgsLimitPerSecond))
}