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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
package sse
import (
"bufio"
"bytes"
"context"
"io"
"time"
)
// Event holds all of the event source fields
type Event struct {
timestamp time.Time
ID []byte
Data []byte
Event []byte
Retry []byte
Comment []byte
}
func (e *Event) hasContent() bool {
return len(e.ID) > 0 || len(e.Data) > 0 || len(e.Event) > 0 || len(e.Retry) > 0
}
// EventStreamReader scans an io.Reader looking for EventStream messages.
type EventStreamReader struct {
scanner *bufio.Scanner
}
// NewEventStreamReader creates an instance of EventStreamReader.
func NewEventStreamReader(eventStream io.Reader, maxBufferSize int) *EventStreamReader {
scanner := bufio.NewScanner(eventStream)
initBufferSize := minPosInt(4096, maxBufferSize)
scanner.Buffer(make([]byte, initBufferSize), maxBufferSize)
split := func(data []byte, atEOF bool) (int, []byte, error) {
if atEOF && len(data) == 0 {
return 0, nil, nil
}
// We have a full event payload to parse.
if i, nlen := containsDoubleNewline(data); i >= 0 {
return i + nlen, data[0:i], nil
}
// If we're at EOF, we have all of the data.
if atEOF {
return len(data), data, nil
}
// Request more data.
return 0, nil, nil
}
// Set the split function for the scanning operation.
scanner.Split(split)
return &EventStreamReader{
scanner: scanner,
}
}
// Returns a tuple containing the index of a double newline, and the number of bytes
// represented by that sequence. If no double newline is present, the first value
// will be negative.
func containsDoubleNewline(data []byte) (int, int) {
// Search for each potentially valid sequence of newline characters
crcr := bytes.Index(data, []byte("\r\r"))
lflf := bytes.Index(data, []byte("\n\n"))
crlflf := bytes.Index(data, []byte("\r\n\n"))
lfcrlf := bytes.Index(data, []byte("\n\r\n"))
crlfcrlf := bytes.Index(data, []byte("\r\n\r\n"))
// Find the earliest position of a double newline combination
minPos := minPosInt(crcr, minPosInt(lflf, minPosInt(crlflf, minPosInt(lfcrlf, crlfcrlf))))
// Detemine the length of the sequence
nlen := 2
if minPos == crlfcrlf {
nlen = 4
} else if minPos == crlflf || minPos == lfcrlf {
nlen = 3
}
return minPos, nlen
}
// Returns the minimum non-negative value out of the two values. If both
// are negative, a negative value is returned.
func minPosInt(a, b int) int {
if a < 0 {
return b
}
if b < 0 {
return a
}
if a > b {
return b
}
return a
}
// ReadEvent scans the EventStream for events.
func (e *EventStreamReader) ReadEvent() ([]byte, error) {
if e.scanner.Scan() {
event := e.scanner.Bytes()
return event, nil
}
if err := e.scanner.Err(); err != nil {
if err == context.Canceled {
return nil, io.EOF
}
return nil, err
}
return nil, io.EOF
}
|
