-
Notifications
You must be signed in to change notification settings - Fork 5.5k
/
reader.go
68 lines (59 loc) · 1.71 KB
/
reader.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
package influx
import (
"bytes"
"io"
"log"
"github.com/influxdata/telegraf"
)
// reader is an io.Reader for line protocol.
type reader struct {
metrics []telegraf.Metric
serializer *Serializer
offset int
buf *bytes.Buffer
}
// NewReader creates a new reader over the given metrics.
func NewReader(metrics []telegraf.Metric, serializer *Serializer) io.Reader {
return &reader{
metrics: metrics,
serializer: serializer,
offset: 0,
buf: bytes.NewBuffer(make([]byte, 0, serializer.maxLineBytes)),
}
}
// SetMetrics changes the metrics to be read.
func (r *reader) SetMetrics(metrics []telegraf.Metric) {
r.metrics = metrics
r.offset = 0
r.buf.Reset()
}
// Read reads up to len(p) bytes of the current metric into p, each call will
// only serialize at most one metric so the number of bytes read may be less
// than p. Subsequent calls to Read will read the next metric until all are
// emitted. If a metric cannot be serialized, an error will be returned, you
// may resume with the next metric by calling Read again. When all metrics
// are emitted the err is io.EOF.
func (r *reader) Read(p []byte) (int, error) {
if r.buf.Len() > 0 {
return r.buf.Read(p)
}
if r.offset >= len(r.metrics) {
return 0, io.EOF
}
for _, metric := range r.metrics[r.offset:] {
_, err := r.serializer.Write(r.buf, metric)
r.offset += 1
if err != nil {
r.buf.Reset()
if _, ok := err.(*MetricError); ok {
continue
}
// Since we are serializing multiple metrics, don't fail the
// the entire batch just because of one unserializable metric.
log.Printf("E! [serializers.influx] could not serialize metric: %v; discarding metric", err)
continue
}
break
}
return r.buf.Read(p)
}