logdog: Add bundler library.

client/internal/logdog/butler/bundler/binaryParser.go

Index: client/internal/logdog/butler/bundler/binaryParser.go
diff --git a/client/internal/logdog/butler/bundler/binaryParser.go b/client/internal/logdog/butler/bundler/binaryParser.go
new file mode 100644
index 0000000000000000000000000000000000000000..a1594302fa5a09a273491dfc1d3ecb6f1c1ea1c0
--- /dev/null
+++ b/client/internal/logdog/butler/bundler/binaryParser.go
@@ -0,0 +1,68 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package bundler
+
+import (
+	"github.com/luci/luci-go/common/logdog/protocol"
+)
+
+// binaryThreshold is the amount of binary data that we will willingly yield
+// if not requested to truncate. This helps build larger binary stream chunks.
+const defaultBinaryThreshold = 8 * 1024
+
+// binaryParser is a parser implementation for the LogDog BINARY stream type.
+type binaryParser struct {
+	baseParser
+
+	offset    int64
+	threshold int
+}
+
+var _ parser = (*binaryParser)(nil)
+
+func (p *binaryParser) nextEntry(c *constraints) (*protocol.LogEntry, error) {

[iannucci, 2015/11/13 07:16:38]

why return an error at all? this looks like it can only return nil or panic?

[dnj, 2015/11/14 00:30:36]

This is implementing the "parser" interface, which allows for an error.

The error case is employed when the stream contains invalid data. This is only
used once, though, when a datagram's size prefix exceeds a limit. The intent is
to shut down the stream when this sort of thing happens.

+	threshold := p.getThreshold()
+	if c.truncate {
+		// If we're truncating, return _any_ available data.
+		threshold = 0
+	}
+
+	count := p.Len()
+	if count <= int64(threshold) {
+		return nil, nil
+	}
+	if count > int64(c.limit) {
+		count = int64(c.limit)
+	}
+
+	// The integer conversion, since count has been bounded by our "int" limit.
+	size := int(count)
+
+	data := make([]byte, size)
+	size, _ = p.Reader().Read(data)
+	if int64(size) != count {
+		panic("partial buffer read")
+	}
+
+	ts, _ := p.firstChunkTime()
+	e := p.baseLogEntry(ts)
+	e.Content = &protocol.LogEntry_Binary{Binary: &protocol.Binary{
+		Offset: uint64(p.offset),
+		Data:   data[:size],
+	}}
+	e.Sequence = uint64(p.offset)
+
+	p.Consume(int64(size))
+	p.offset += int64(size)
+	return e, nil
+}
+
+func (p *binaryParser) getThreshold() int {
+	result := p.threshold
+	if result == 0 {
+		result = defaultBinaryThreshold
+	}
+	return result
+}