func (p *protocol) awaitACK(count uint32) (uint32, error) {
	debug("await ack")
	var ackSeq uint32
	var err error
	for ackSeq < count {
		ackSeq, err = p.recvACK()
		if err != nil {
			debug("receive ack failed with: %v", err)
			return ackSeq, err
		}
	}
	if ackSeq > count {
		return count, fmt.Errorf(
			"invalid sequence number received (seq=%v, expected=%v)", ackSeq, count)
	}
	return ackSeq, nil
}
wait for ACK (accept partial ACK to reset timeout)
reset timeout timer for every ACK received.
package logstash
import (
	"bytes"
	"compress/zlib"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"time"
	"github.com/elastic/beats/libbeat/common"
	"github.com/elastic/beats/libbeat/logp"
)
type protocol struct {
	conn          net.Conn
	timeout       time.Duration
	compressLevel int
	eventsBuffer *bytes.Buffer
}
var (
	ErrProtocolError = errors.New("lumberjack protocol error")
	errAllEventsEncoding = errors.New("failed to encode all events")
)
var (
	codeVersion byte = '2'
	codeWindowSize    = []byte{codeVersion, 'W'}
	codeJSONDataFrame = []byte{codeVersion, 'J'}
	codeCompressed    = []byte{codeVersion, 'C'}
)
func newClientProcol(
	conn net.Conn,
	timeout time.Duration,
	compressLevel int,
) (*protocol, error) {
	if compressLevel > 0 {
		tmp := bytes.NewBuffer(nil)
		w, err := zlib.NewWriterLevel(tmp, compressLevel)
		if err != nil {
			return nil, err
		}
		w.Close()
	}
	return &protocol{
		conn:          conn,
		timeout:       timeout,
		compressLevel: compressLevel,
		eventsBuffer:  bytes.NewBuffer(nil),
	}, nil
}
func (p *protocol) Close() error {
	return p.conn.Close()
}
func (p *protocol) sendEvents(events []common.MapStr) ([]common.MapStr, error) {
	conn := p.conn
	if len(events) == 0 {
		return nil, nil
	}
	debug("send events")
	outEvents, err := p.serializeEvents(events)
	count := uint32(len(outEvents))
	if count == 0 {
		debug("no events serializable")
		return nil, errAllEventsEncoding
	}
	debug("send window size")
	if err := p.sendWindowSize(count); err != nil {
		return nil, err
	}
	if p.compressLevel > 0 {
		err = p.sendCompressed(p.eventsBuffer.Bytes())
	} else {
		debug("write events")
		_, err = conn.Write(p.eventsBuffer.Bytes())
	}
	if err != nil {
		return nil, err
	}
	debug("did send %v events", count)
	return outEvents, nil
}
func (p *protocol) recvACK() (uint32, error) {
	conn := p.conn
	if err := conn.SetReadDeadline(time.Now().Add(p.timeout)); err != nil {
		return 0, err
	}
	response := make([]byte, 6)
	ackbytes := 0
	for ackbytes < 6 {
		n, err := conn.Read(response[ackbytes:])
		if err != nil {
			debug("read ack sequence failed with: %v", err)
			return 0, err
		}
		ackbytes += n
	}
	isACK := response[0] == codeVersion && response[1] == 'A'
	if !isACK {
		return 0, ErrProtocolError
	}
	seq := binary.BigEndian.Uint32(response[2:])
	return seq, nil
}