scanEvents iterates SSE events in the given scanner. The iterated error is terminal: if encountered, the stream is corrupt or broken and should no longer be used. TODO(rfindley): consider a different API here that makes failure modes more apparent.
(r io.Reader)
| 67 | // TODO(rfindley): consider a different API here that makes failure modes more |
| 68 | // apparent. |
| 69 | func scanEvents(r io.Reader) iter.Seq2[Event, error] { |
| 70 | reader := bufio.NewReader(r) |
| 71 | |
| 72 | // TODO: investigate proper behavior when events are out of order, or have |
| 73 | // non-standard names. |
| 74 | var ( |
| 75 | eventKey = []byte("event") |
| 76 | idKey = []byte("id") |
| 77 | dataKey = []byte("data") |
| 78 | retryKey = []byte("retry") |
| 79 | ) |
| 80 | |
| 81 | return func(yield func(Event, error) bool) { |
| 82 | // iterate event from the wire. |
| 83 | // https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#examples |
| 84 | // |
| 85 | // - `key: value` line records. |
| 86 | // - Consecutive `data: ...` fields are joined with newlines. |
| 87 | // - Unrecognized fields are ignored. Since we only care about 'event', 'id', and |
| 88 | // 'data', these are the only three we consider. |
| 89 | // - Lines starting with ":" are ignored. |
| 90 | // - Records are terminated with two consecutive newlines. |
| 91 | var ( |
| 92 | evt Event |
| 93 | dataBuf *bytes.Buffer // if non-nil, preceding field was also data |
| 94 | ) |
| 95 | yieldEvent := func() bool { |
| 96 | if dataBuf != nil { |
| 97 | evt.Data = dataBuf.Bytes() |
| 98 | dataBuf = nil |
| 99 | } |
| 100 | if evt.Empty() { |
| 101 | return true |
| 102 | } |
| 103 | if !yield(evt, nil) { |
| 104 | return false |
| 105 | } |
| 106 | evt = Event{} |
| 107 | return true |
| 108 | } |
| 109 | for { |
| 110 | line, err := reader.ReadBytes('\n') |
| 111 | if err != nil && !errors.Is(err, io.EOF) { |
| 112 | yield(Event{}, fmt.Errorf("error reading event: %v", err)) |
| 113 | return |
| 114 | } |
| 115 | line = bytes.TrimRight(line, "\r\n") |
| 116 | isEOF := errors.Is(err, io.EOF) |
| 117 | |
| 118 | if len(line) == 0 { |
| 119 | if !yieldEvent() { |
| 120 | return |
| 121 | } |
| 122 | if isEOF { |
| 123 | return |
| 124 | } |
| 125 | continue |
| 126 | } |
searching dependent graphs…