buffer.go

func inFlightBuffer(_ *jsm.Stream, consumer *jsm.Consumer, tests int) error {
	buffer := make(chan *nats.Msg, 1000)
	ib := nats.NewInbox()
	trigger := make(chan struct{})
	done := make(chan struct{})

	_, err := nc.ChanSubscribe(ib, buffer)
	if err != nil {
		return err
	}

	filler := func() {
		expected := 0

		for {
			select {
			case <-trigger:
				rs := 64 - expected

				if rs > 0 {
					log.Printf("Fetching %d", rs)
					nc.PublishRequest(consumer.NextSubject(), ib, []byte(strconv.Itoa(rs)))
					expected += rs
				}

			case _, ok := <-done:
				if !ok {
					return
				}
			}
		}
	}

	go filler()

	// fetch initial data
	trigger <- struct{}{}

	cnt := 1

	for {
		select {
		case msg := <-buffer:
			// time.Sleep(time.Millisecond)
			nc.PublishRequest(msg.Reply, ib, server.AckNext)

			log.Printf("buffer size: %d/%d (%s)", len(buffer), cap(buffer), string(msg.Data))

			if cnt == tests {
				close(done)
				continue
			}

			cnt++
			if len(buffer) < cap(buffer)/2 {
				trigger <- struct{}{}
			}

		case _, ok := <-done:
			if !ok {
				return nil
			}
		}
	}

	return nil
}