func (s *TopicCsmSuite) TestAckTimeoutExpires(c *C) {
	s.cfg.Consumer.SubscriptionTimeout = 500
	s.cfg.Consumer.AckTimeout = 700
	s.setSafe2Stop(false)
	safe2StopPollingInterval = 5
	tc := Spawn(s.ns, group, s.childSpec, s.cfg, s.lifespanCh, s.isSafe2Stop)
	c.Assert(<-s.lifespanCh, Equals, tc)
	c.Assert(clock.Advance(500), Equals, time.Duration(500))
	assertRunning(c, s.lifespanCh, 50*time.Millisecond)
	c.Assert(clock.Advance(199), Equals, time.Duration(699))
	assertRunning(c, s.lifespanCh, 50*time.Millisecond)
	c.Assert(clock.Advance(5), Equals, time.Duration(704))
	assertStopped(c, s.lifespanCh, time.Second)
}
If there has been no requests for SubscriptionTimeout, but it is not safe to
stop then the topic consumer waits until AckTimeout expires as well before
terminating.
package topiccsm
import (
	"sync"
	"testing"
	"time"
	"github.com/mailgun/holster/v4/clock"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	"github.com/mailgun/kafka-pixy/consumer"
	"github.com/mailgun/kafka-pixy/consumer/dispatcher"
	"github.com/mailgun/kafka-pixy/testhelpers"
	. "gopkg.in/check.v1"
)
const (
	group = "g"
)
func Test(t *testing.T) {
	TestingT(t)
}
type TopicCsmSuite struct {
	epoch      time.Time
	cfg        *config.Proxy
	ns         *actor.Descriptor
	requestsCh chan consumer.Request
	childSpec  dispatcher.ChildSpec
	lifespanCh chan *T
	mu         sync.Mutex
	safe2Stop  bool
	checkCount int
}
var _ = Suite(&TopicCsmSuite{})
func (s *TopicCsmSuite) SetUpSuite(c *C) {
	var err error
	s.epoch, err = time.Parse(time.RFC3339, "2009-02-19T00:00:00Z")
	c.Assert(err, IsNil)
	testhelpers.InitLogging()
}
func (s *TopicCsmSuite) SetUpTest(c *C) {
	clock.Freeze(s.epoch)
	s.cfg = testhelpers.NewTestProxyCfg("topiccsm")
	s.ns = actor.Root().NewChild("T")
	s.requestsCh = make(chan consumer.Request, 100)
	s.childSpec = dispatcher.NewChildSpec4Test(s.requestsCh)
	s.lifespanCh = make(chan *T, 2)
	s.safe2Stop = true
	s.checkCount = 0
}
func (s *TopicCsmSuite) TearDownTest(c *C) {
	clock.Unfreeze()
}
func (s *TopicCsmSuite) isSafe2Stop() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.checkCount++
	return s.safe2Stop
}
func (s *TopicCsmSuite) setSafe2Stop(safe2Stop bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.safe2Stop = safe2Stop
}
func (s *TopicCsmSuite) TestRequestResponse(c *C) {
	tc := Spawn(s.ns, group, s.childSpec, s.cfg, s.lifespanCh, s.isSafe2Stop)
	c.Assert(<-s.lifespanCh, Equals, tc)
	defer func() {
		close(s.requestsCh)  
		<-s.lifespanCh       
	}()
	requests := make([]consumer.Request, 10)
	messages := make([]consumer.Message, 10)
	eventsChs := make([]chan consumer.Event, 10)
	for i := 0; i < 10; i++ {
		requests[i] = newRequest()
		messages[i], eventsChs[i] = newMessage(i)
	}
	for _, rq := range requests {
		s.requestsCh <- rq
	}
	for _, msg := range messages {
		tc.Messages() <- msg
	}
	for i := 0; i < 10; i++ {
		c.Assert(<-requests[i].ResponseCh, DeepEquals,
			consumer.Response{Msg: messages[i]})
		c.Assert(<-eventsChs[i], DeepEquals,
			consumer.Event{T: consumer.EvOffered, Offset: messages[i].Offset})
	}
}
func (s *TopicCsmSuite) TestLongPollingExpires(c *C) {
	s.cfg.Consumer.LongPollingTimeout = 300
	tc := Spawn(s.ns, group, s.childSpec, s.cfg, s.lifespanCh, s.isSafe2Stop)
	c.Assert(<-s.lifespanCh, Equals, tc)
	defer func() {
		close(s.requestsCh)  
		<-s.lifespanCh       
	}()
	rq1 := newRequest()  
	c.Assert(clock.Advance(1), Equals, time.Duration(1))
	rq2 := newRequest()  
	c.Assert(clock.Advance(1), Equals, time.Duration(2))
	rq3 := newRequest()  
	msg1, _ := newMessage(42)
	msg2, _ := newMessage(43)
	s.requestsCh <- rq1
	s.requestsCh <- rq2
	s.requestsCh <- rq3
	c.Assert(clock.Advance(297), Equals, time.Duration(299))
	tc.Messages() <- msg1
	assertResponse(c, rq1, consumer.Response{Msg: msg1}, time.Second)
	c.Assert(clock.Advance(2), Equals, time.Duration(301))
	assertResponse(c, rq2, requestTimeoutRs, time.Second)
	tc.Messages() <- msg2
	assertResponse(c, rq3, consumer.Response{Msg: msg2}, time.Second)
}
func (s *TopicCsmSuite) TestStaleRequest(c *C) {
	s.cfg.Consumer.LongPollingTimeout = 300
	tc := Spawn(s.ns, group, s.childSpec, s.cfg, s.lifespanCh, s.isSafe2Stop)
	c.Assert(<-s.lifespanCh, Equals, tc)
	defer func() {
		close(s.requestsCh)  
		<-s.lifespanCh       
	}()
	rq := newRequest()  
	s.requestsCh <- rq
	c.Assert(clock.Advance(300), Equals, time.Duration(300))
	assertResponse(c, rq, requestTimeoutRs, time.Second)
}
func (s *TopicCsmSuite) TestSubscriptionExpires(c *C) {
	s.cfg.Consumer.SubscriptionTimeout = 500
	s.cfg.Consumer.AckTimeout = 300
	tc := Spawn(s.ns, group, s.childSpec, s.cfg, s.lifespanCh, s.isSafe2Stop)
	c.Assert(<-s.lifespanCh, Equals, tc)
	c.Assert(clock.Advance(499), Equals, time.Duration(499))
	s.requestsCh <- newRequest()
	msg, _ := newMessage(42)
	tc.Messages() <- msg
	c.Assert(clock.Advance(499), Equals, time.Duration(998))
	assertRunning(c, s.lifespanCh, 50*time.Millisecond)
	c.Assert(clock.Advance(1), Equals, time.Duration(999))
	assertStopped(c, s.lifespanCh, time.Second)
}