1package live23import (4 "bytes"5 "context"6 "errors"7 "io"8 "sync"9 "time"1011 "src.rest/cat/012/internal/fileio"12 "src.rest/cat/012/internal/sheet"13)1415// Stream follows a program that prints a table as it makes it: NUON16// values one after another, one to a line, as a notebook cell run as a17// stream prints them (nushell.Stream). The program starts with the18// stream and runs on a goroutine of its own until it ends or the stream19// is closed. A poll waits up to Wait for something to arrive, so rows20// reach the region as soon as the program prints them, and the owner21// polls again at once rather than after Interval.22//23// It keeps the last Keep values printed, for the cell's output (NUON)24// and for reading the table whole again (Reload).25type Stream struct {26 mu sync.Mutex27 pending []byte // printed since the last poll28 ended bool29 err error // why the program ended, nil when it ended well30 arrived chan struct{}31 cancel context.CancelFunc3233 // The rest is the poller's: polls never overlap. values is guarded34 // by mu too, since the owner reads it (NUON).35 partial []byte // a line not yet ended36 tail *fileio.Tail37 values [][]byte38 printed int // values printed in all; guarded by mu39 header sheet.LiveRow40 rows []sheet.LiveRow41 total int // rows printed in all; guarded by mu42 reload bool43 done bool // the end was reported; guarded by mu4445 // Wait is how long a poll waits for the program to print.46 Wait time.Duration47 // Keep is how many of the last values are kept.48 Keep int49 // Now is the clock; tests replace it.50 Now func() time.Time51}5253// StreamKeep is how many of the last values a stream keeps.54const StreamKeep = 100005556// NewStream starts run, which prints the table to w until ctx is done,57// and follows what it prints.58func NewStream(run func(ctx context.Context, w io.Writer) error) *Stream {59 ctx, cancel := context.WithCancel(context.Background())60 s := &Stream{arrived: make(chan struct{}, 1), cancel: cancel, Wait: Interval, Keep: StreamKeep, Now: time.Now}61 s.tail, _ = fileio.NewTail(fileio.NUON, nil, nil)62 go func() {63 err := run(ctx, streamWriter{s})64 s.mu.Lock()65 s.ended, s.err = true, err66 s.mu.Unlock()67 s.kick()68 }()69 return s70}7172// streamWriter takes what the program prints.73type streamWriter struct{ s *Stream }7475func (w streamWriter) Write(p []byte) (int, error) {76 w.s.mu.Lock()77 w.s.pending = append(w.s.pending, p...)78 w.s.mu.Unlock()79 w.s.kick()80 return len(p), nil81}8283// kick tells a poll waiting that something arrived.84func (s *Stream) kick() {85 select {86 case s.arrived <- struct{}{}:87 default:88 }89}9091// Poll waits up to Wait for the program to print, and returns the rows92// it completed: all of those kept again after Reload, and the reason93// it failed once it ends with an error.94func (s *Stream) Poll(ctx context.Context) (Update, bool) {95 if done, _ := s.Ended(); done {96 return Update{}, false97 }98 t := time.NewTimer(s.Wait)99 select {100 case <-s.arrived:101 case <-t.C:102 case <-ctx.Done():103 }104 t.Stop()105 s.mu.Lock()106 got, ended, err := s.pending, s.ended, s.err107 s.pending = nil108 s.mu.Unlock()109110 u := Update{At: s.Now()}111 s.read(got, ended, &u)112 if s.reload {113 s.reload = false114 u.Reset, u.Header, u.Rows = true, s.header, s.rows115 }116 if ended && !s.done {117 s.mu.Lock()118 s.done = true119 s.mu.Unlock()120 if err != nil && !isStop(err) {121 u.Err = err.Error()122 }123 return u, true124 }125 return u, u.Reset || len(u.Rows) > 0 || u.Header != nil126}127128// read parses the complete lines of what arrived into u, keeping them.129func (s *Stream) read(got []byte, ended bool, u *Update) {130 s.partial = append(s.partial, got...)131 end := bytes.LastIndexByte(s.partial, '\n') + 1132 if ended {133 end = len(s.partial)134 }135 if end == 0 {136 return137 }138 text := s.partial[:end]139 s.partial = append([]byte(nil), s.partial[end:]...)140 var lines [][]byte141 for line := range bytes.SplitSeq(text, []byte("\n")) {142 if line = bytes.TrimSpace(line); len(line) > 0 {143 lines = append(lines, append([]byte(nil), line...))144 }145 }146 if len(lines) == 0 {147 return148 }149 res, err := s.tail.Feed(append(bytes.Join(lines, []byte("\n")), '\n'))150 if err != nil {151 u.Err = err.Error()152 }153 if res.Header != nil {154 s.header = res.Header155 u.Header = res.Header156 }157 s.mu.Lock() // the values and the counts change together (Snapshot)158 s.values = keepLast(append(s.values, lines...), s.Keep)159 s.printed += len(lines)160 s.total += len(res.Rows)161 s.mu.Unlock()162 s.rows = keepLast(append(s.rows, res.Rows...), s.Keep)163 u.Rows = res.Rows164}165166// keepLast is the last n of xs.167func keepLast[T any](xs []T, n int) []T {168 if n > 0 && len(xs) > n {169 return append(xs[:0:0], xs[len(xs)-n:]...)170 }171 return xs172}173174// isStop reports whether err is the stream being stopped rather than175// failing.176func isStop(err error) bool { return errors.Is(err, context.Canceled) }177178// Reload has the next poll give every row kept again.179func (s *Stream) Reload() { s.reload = true }180181// Close stops the program; the stream is polled no more.182func (s *Stream) Close() {183 s.cancel()184 s.tail.Close()185}186187// Ended reports whether the program has ended and the last poll gave188// what it printed, and why it failed if it did (nil when it ended well189// or was stopped).190func (s *Stream) Ended() (bool, error) {191 s.mu.Lock()192 defer s.mu.Unlock()193 if !s.done {194 return false, nil195 }196 if isStop(s.err) {197 return true, nil198 }199 return true, s.err200}201202// Rows is how many rows the program has printed so far.203func (s *Stream) Rows() int {204 s.mu.Lock()205 defer s.mu.Unlock()206 return s.total207}208209// NUON is the values kept, as a NUON list: the cell's output so far.210func (s *Stream) NUON() []byte {211 s.mu.Lock()212 defer s.mu.Unlock()213 return list(s.values)214}215216// list is values as a NUON list.217func list(values [][]byte) []byte {218 return append(append([]byte("["), bytes.Join(values, []byte(", "))...), ']')219}220221// Snapshot is what a stream has printed, read at one moment.222type Snapshot struct {223 NUON []byte // the values kept, as NUON gives them224 // More are the values printed after the first seen (Snapshot's225 // argument), as a NUON list, so what shows the output can add them226 // to what it shows rather than read it all again; nil when some of227 // them are kept no more.228 More []byte229 // Values and Rows are how many values and rows have been printed.230 Values, Rows int231}232233// Snapshot is what the stream has printed so far, and what of it came234// after the first seen values.235func (s *Stream) Snapshot(seen int) Snapshot {236 s.mu.Lock()237 defer s.mu.Unlock()238 snap := Snapshot{NUON: list(s.values), Values: s.printed, Rows: s.total}239 if n := s.printed - seen; n >= 0 && n <= len(s.values) {240 snap.More = list(s.values[len(s.values)-n:])241 }242 return snap243}