func (msg *udpMessage) addDatagram(
	header *mcUdpHeader,
	data []byte,
) *streambuf.Buffer {
	if msg.isComplete {
		return nil
	}

	if msg.numDatagrams == 1 {
		msg.isComplete = true
		return streambuf.NewFixed(data)
	}

	if msg.count < msg.numDatagrams {
		if msg.datagrams[header.seqNumber] != nil {
			return nil
		}
		msg.datagrams[header.seqNumber] = data
		msg.count++
	}

	if msg.count < msg.numDatagrams {
		return nil
	}

	buffer := streambuf.New(nil)
	for _, payload := range msg.datagrams {
		buffer.Append(payload)
	}
	msg.isComplete = true
	msg.datagrams = nil
	buffer.Fix()
	return buffer
}

func parseTextArgs(parser *parser, args []argDef) (err error) {
	buf := streambuf.NewFixed(parser.message.rawArgs)
	for _, arg := range args {
		debug("args rest: %s", buf.Bytes())
		err = arg.parse(parser, nil, buf)
		if err != nil {
			break
		}
	}
	return
}

func parseCounterResponse(parser *parser, buf *streambuf.Buffer) parseResult {
	msg := parser.message
	tmp := streambuf.NewFixed(msg.rawCommand)
	msg.value, _ = tmp.UintASCII(false)
	if tmp.Failed() {
		err := tmp.Err()
		debug("counter response invalid: %v", err)
		return parser.failing(err)
	}
	debug("parsed counter response: %v", msg.value)
	return parser.yieldNoData(buf)
}

func splitCommandAndArgs(line []byte) ([]byte, []byte, error) {
	command_line := streambuf.NewFixed(line)
	command, err := parseStringArg(command_line)
	if err != nil {
		return nil, nil, err
	}
	var args []byte
	if command_line.Len() > 0 {
		command_line.Advance(1)
		args = command_line.Bytes()
	}
	return command, args, command_line.Err()
}

func (r *jsonReader) nextInt() (int, error) {
	entity, raw, err := r.step()
	if err != nil {
		return 0, err
	}

	if entity != intEntity {
		return 0, errExpectedInteger
	}

	tmp := streambuf.NewFixed(raw)
	i, err := tmp.AsciiInt(false)
	return int(i), err
}

func parseBinaryCounterResponse(
	parser *parser,
	buf *streambuf.Buffer,
) parseResult {
	msg := parser.message
	if msg.IsRequest {
		return parser.contWith(buf, parseStateDataBinary)
	}

	// size already checked
	bytes, _ := buf.Collect(int(msg.bytes))
	tmp := streambuf.NewFixed(bytes)
	err := withBinaryUint64(parser, tmp, func(msg *message, value uint64) {
		msg.value = value
	})
	if err != nil {
		return parser.failing(err)
	}

	buf.Advance(8)
	return parser.yield(buf.BufferConsumed())
}

func parseInt(line []byte) (int64, error) {
	buf := streambuf.NewFixed(line)
	return buf.IntASCII(false)
	// TODO: is it an error if 'buf.Len() != 0 {}' ?
}

func readMessage(buf *streambuf.Buffer) (*message, error) {
	if !buf.Avail(2) {
		return nil, nil
	}

	version, _ := buf.ReadNetUint8At(0)
	if version != '2' {
		return nil, errors.New("version error")
	}

	code, _ := buf.ReadNetUint8At(1)
	switch code {
	case 'W':
		if !buf.Avail(6) {
			return nil, nil
		}
		size, _ := buf.ReadNetUint32At(2)
		buf.Advance(6)
		buf.Reset()
		return &message{code: code, size: size}, buf.Err()
	case 'C':
		if !buf.Avail(6) {
			return nil, nil
		}
		len, _ := buf.ReadNetUint32At(2)
		if !buf.Avail(int(len) + 6) {
			return nil, nil
		}
		buf.Advance(6)

		tmp, _ := buf.Collect(int(len))
		buf.Reset()

		dataBuf := streambuf.New(nil)
		// decompress data
		decomp, err := zlib.NewReader(streambuf.NewFixed(tmp))
		if err != nil {
			return nil, err
		}
		// dataBuf.ReadFrom(streambuf.NewFixed(tmp))
		dataBuf.ReadFrom(decomp)
		decomp.Close()

		// unpack data
		dataBuf.Fix()
		var events []*message
		for dataBuf.Len() > 0 {
			version, _ := dataBuf.ReadNetUint8()
			if version != '2' {
				return nil, errors.New("version error 2")
			}

			code, _ := dataBuf.ReadNetUint8()
			if code != 'J' {
				return nil, errors.New("expected json data frame")
			}

			seq, _ := dataBuf.ReadNetUint32()
			payloadLen, _ := dataBuf.ReadNetUint32()
			jsonRaw, _ := dataBuf.Collect(int(payloadLen))

			var doc interface{}
			err = json.Unmarshal(jsonRaw, &doc)
			if err != nil {
				return nil, err
			}

			events = append(events, &message{
				code: code,
				seq:  seq,
				doc:  doc.(map[string]interface{}),
			})
		}
		return &message{code: 'C', events: events}, nil
	default:
		return nil, errors.New("unknown code")
	}
}

func (mc *Memcache) ParseUdp(pkt *protos.Packet) {
	defer logp.Recover("ParseMemcache(UDP) exception")

	buffer := streambuf.NewFixed(pkt.Payload)
	header, err := parseUdpHeader(buffer)
	if err != nil {
		debug("parsing memcache udp header failed")
		return
	}

	debug("new udp datagram requestId=%v, seqNumber=%v, numDatagrams=%v",
		header.requestId, header.seqNumber, header.numDatagrams)

	// find connection object based on ips and ports (forward->reverse connection)
	connection, dir := mc.getUdpConnection(&pkt.Tuple)
	debug("udp connection: %p", connection)

	// get udp transaction combining forward/reverse direction 'streams'
	// for current requestId
	trans := connection.udpTransactionForId(header.requestId)
	debug("udp transaction (id=%v): %p", header.requestId, trans)

	// Clean old transaction. We do the cleaning after potentially adding a new
	// transaction to the connection object, so connection object will not be
	// cleaned accidentally (not bad, but let's rather reuse it)
	expTrans := mc.udpExpTrans.steal()
	for expTrans != nil {
		tmp := expTrans.next
		expTrans.connection.killTransaction(expTrans)
		expTrans = tmp
	}

	// get UDP transaction stream combining datagram packets in transaction
	udpMsg := trans.udpMessageForDir(&header, dir)
	if udpMsg.numDatagrams != header.numDatagrams {
		logp.Warn("number of datagram mismatches in stream")
		connection.killTransaction(trans)
		return
	}

	// try to combine datagrams into complete memcached message
	payload := udpMsg.addDatagram(&header, buffer.Bytes())
	done := false
	if payload != nil {
		// parse memcached message
		msg, err := parseUdp(&mc.config, pkt.Ts, payload)
		if err != nil {
			logp.Warn("failed to parse memcached(UDP) message: %s", err)
			connection.killTransaction(trans)
			return
		}

		// apply memcached to transaction
		done, err = mc.onUdpMessage(trans, &pkt.Tuple, dir, msg)
		if err != nil {
			logp.Warn("error processing memcache message: %s", err)
			connection.killTransaction(trans)
			done = true
		}
	}
	if !done {
		trans.timer = time.AfterFunc(mc.udpConfig.transTimeout, func() {
			debug("transaction timeout -> forward")
			mc.onUdpTrans(trans)
			mc.udpExpTrans.push(trans)
		})
	}
}

func parseInt(line []byte) (int, error) {
	buf := streambuf.NewFixed(line)
	i, err := buf.AsciiInt(false)
	return int(i), err
	// TODO: is it an error if 'buf.Len() != 0 {}' ?
}

func makeParseBinary(
	requestArgs, responseArgs []argDef,
	valueParser parserStateFn,
) parserStateFn {
	return func(parser *parser, buf *streambuf.Buffer) parseResult {
		header := buf.Snapshot()
		buf.Advance(memcacheHeaderSize)

		msg := parser.message
		if msg.IsRequest {
			msg.vbucket, _ = header.ReadNetUint16At(6)
		} else {
			msg.status, _ = header.ReadNetUint16At(6)
		}

		cas, _ := header.ReadNetUint64At(16)
		if cas != 0 {
			setCasUnique(msg, cas)
		}
		msg.opaque, _ = header.ReadNetUint32At(12)

		// check message length

		extraLen, _ := header.ReadNetUint8At(4)
		keyLen, _ := header.ReadNetUint16At(2)
		totalLen, _ := header.ReadNetUint32At(8)
		if totalLen == 0 {
			// no extra, key or value -> publish message
			return parser.yield(buf.BufferConsumed())
		}

		valueLen := int(totalLen) - (int(extraLen) + int(keyLen))
		if valueLen < 0 {
			return parser.failing(ErrLen)
		}

		if valueParser != nil && valueLen > 0 {
			if !buf.Avail(int(totalLen)) {
				return parser.needMore()
			}
		}

		var extras *streambuf.Buffer
		if extraLen > 0 {
			tmp, _ := buf.Collect(int(extraLen))
			extras = streambuf.NewFixed(tmp)
			var err error
			if msg.IsRequest && requestArgs != nil {
				err = parseBinaryArgs(parser, requestArgs, header, extras)
			} else if responseArgs != nil {
				err = parseBinaryArgs(parser, responseArgs, header, extras)
			}
			if err != nil {
				msg.AddNotes(err.Error())
			}
		}

		if keyLen > 0 {
			key, _ := buf.Collect(int(keyLen))
			keys := []memcacheString{{key}}
			msg.keys = keys
		}

		if valueLen == 0 {
			return parser.yield(buf.BufferConsumed())
		}

		// call parseDataBinary
		msg.bytes = uint(valueLen)
		if valueParser == nil {
			return parser.contWith(buf, parseStateDataBinary)
		}
		return parser.contWithShallow(buf, valueParser)
	}
}