1package ui23import (4 "context"5 "io"6 "slices"7 "strconv"8 "time"910 tea "charm.land/bubbletea/v2"1112 "src.rest/cat/012/internal/live"13 "src.rest/cat/012/internal/notebook"14 "src.rest/cat/012/internal/nushell"15)1617// Cells run as streams (Run as stream): the pipeline runs until it ends18// or is stopped, and each value it yields reaches the cell's output, and19// the sheet the output was sent to, as soon as nu prints it, so20// `tail -f app.log | lines | parse ...` or `watch` is followed live. The21// run is a live.Stream, the linked regions' source: it is polled one22// poll at a time, each poll waiting for what the pipeline prints, and23// what it gives is applied as a live operation to the output's region24// (sheet.LiveOp, outside the undo history) and kept as the cell's25// output, its last live.StreamKeep values. Streams run beside the queue26// of cells run once, each on its own; Stop ends them all.2728// nbStream is a cell running as a stream.29type nbStream struct {30 *nbRun31 src *live.Stream32 count int // the run's number, [n]33 // rows and values are how many rows and values the output has had34 // so far, as it last showed them.35 rows, values int36 // shown is when the output was last updated, so a fast stream37 // updates it a few times a second rather than on every poll, and38 // behind is set when rows came since.39 shown time.Time40 behind bool41}4243// nbStreamMsg is what a stream's poll found.44type nbStreamMsg struct {45 st *nbStream46 u live.Update47 ok bool48}4950// streamOutputEvery is how often a stream's output is rebuilt, at most:51// rows reach a sheet the output was sent to on every poll.52const streamOutputEvery = 100 * time.Millisecond5354// streamSelected runs the selected code cell as a stream.55func (m *Model) streamSelected() tea.Cmd {56 v := m.nbView()57 v.StopEdit()58 c, ok := v.Cell()59 if !ok || c.Kind != notebook.Code {60 return nil61 }62 s := m.sheet63 return m.trustNotebook(func(m *Model) tea.Cmd { return m.startStream(nbQueued{s, c.ID}) })64}6566// startStream starts cell q as a stream, in place of any stream or67// queued run of it.68func (m *Model) startStream(q nbQueued) tea.Cmd {69 c, ok := cellOf(q)70 if !ok {71 return nil72 }73 m.stopStream(q.id)74 m.nb.runs.queue = slices.DeleteFunc(m.nb.runs.queue, func(x nbQueued) bool { return x == q })75 cells := q.s.NotebookCells()76 i := slices.IndexFunc(cells, func(x notebook.Cell) bool { return x.ID == c.ID })77 if why := notebook.Taken(cells, i); why != "" {78 m.failCell(q, c, why)79 return nil80 }81 run := &nbRun{nbQueued: q, source: c.Source, start: time.Now()}82 job, err := m.jobFor(q.s, c, run, true)83 if err != nil {84 m.failCell(q, c, err.Error())85 return nil86 }87 runner, parent := m.runner(), m.spans.Parent()88 src := live.NewStream(func(ctx context.Context, w io.Writer) error {89 span := parent.Start("nu.stream")90 err := nushell.Stream(ctx, runner, job, w)91 span.End()92 return err93 })94 src.Wait = streamOutputEvery // an idle poll brings the output up to date95 m.nb.runs.count++96 st := &nbStream{nbRun: run, src: src, count: m.nb.runs.count}97 if m.nb.runs.streams == nil {98 m.nb.runs.streams = map[int]*nbStream{}99 }100 m.nb.runs.streams[q.id] = st101 m.showStream(st)102 st.shown = time.Time{} // the first rows show at once103 m.feedOutput(c.Name())104 m.note = "Following cell " + strconv.Itoa(i+1) + " as a stream: Stop (i i) ends it"105 return tea.Batch(m.roomOwned(pollStream(st)), m.tickStreams())106}107108// pollStream polls st's stream on a goroutine of its own.109func pollStream(st *nbStream) tea.Cmd {110 return func() tea.Msg {111 u, ok := st.src.Poll(context.Background())112 return nbStreamMsg{st, u, ok}113 }114}115116// streamPolled applies what a stream's poll found: its rows to the region117// the output was sent to, and to the output, until the stream ends.118func (m *Model) streamPolled(msg nbStreamMsg) tea.Cmd {119 st := msg.st120 if m.nb.runs.streams[st.id] != st {121 st.src.Close() // stopped, or started again since122 return nil123 }124 c, ok := cellOf(st.nbQueued)125 if !ok {126 m.stopStream(st.id)127 return nil128 }129 st.behind = st.behind || msg.ok130 if msg.ok {131 if _, r, ok := m.book().Region(c.Name()); ok && r.Output {132 if err := m.book().ApplyLive(msg.u.Op(r.Name)); err != nil {133 m.warn = err.Error()134 }135 }136 }137 if done, err := st.src.Ended(); done {138 delete(m.nb.runs.streams, st.id)139 m.endStream(st, err)140 return nil141 }142 if st.behind && time.Since(st.shown) >= streamOutputEvery {143 m.showStream(st)144 }145 return m.roomOwned(pollStream(st))146}147148// showStream keeps what the stream has printed so far as the cell's149// output, the rows it says it has printed with it, so the two agree. The150// grid showing the output takes the rows printed since it last showed151// (nbgridstream.go).152func (m *Model) showStream(st *nbStream) {153 snap := st.src.Snapshot(st.values)154 st.shown, st.behind = time.Now(), false155 st.rows, st.values = snap.Rows, snap.Values156 old := m.book().Output(st.id)157 m.book().SetOutput(st.id, ¬ebook.Output{NUON: snap.NUON, Count: st.count, Took: time.Since(st.start),158 Source: st.source, Reads: st.reads, Selection: st.selection})159 if v := m.nb.views[st.s]; v != nil && old != nil && old.Count == st.count && snap.More != nil {160 v.Carry(st.id, old, m.book().Output(st.id), snap.More)161 }162}163164// endStream keeps a stream's output as it ended: its rows, and why it165// failed if it did.166func (m *Model) endStream(st *nbStream, err error) {167 m.showStream(st)168 if err != nil {169 o := *m.book().Output(st.id)170 o.Err = runError(err)171 m.book().SetOutput(st.id, &o)172 }173}174175// stopStream stops cell id's stream, keeping what it printed.176func (m *Model) stopStream(id int) {177 st := m.nb.runs.streams[id]178 if st == nil {179 return180 }181 delete(m.nb.runs.streams, id)182 st.src.Close()183 m.showStream(st)184}185186// stopStreams stops every stream.187func (m *Model) stopStreams() {188 for id := range m.nb.runs.streams {189 m.stopStream(id)190 }191}192193// closeStreams stops every stream's process as the program ends,194// touching nothing else: the model may be broken by a crash.195func (m *Model) closeStreams() {196 for _, st := range m.nb.runs.streams {197 st.src.Close()198 }199}200201// tickStreams redraws streaming cells' heads while any runs, so the202// toolbar and the output's run time keep up.203func (m *Model) tickStreams() tea.Cmd {204 if m.nb.still || m.nb.streamTick {205 return nil206 }207 m.nb.streamTick = true208 return tea.Tick(time.Second, func(time.Time) tea.Msg { return nbStreamTickMsg{} })209}210211// nbStreamTickMsg redraws streaming cells.212type nbStreamTickMsg struct{}213214// streamTicked keeps the ticks going while a stream runs.215func (m *Model) streamTicked() tea.Cmd {216 m.nb.streamTick = false217 if len(m.nb.runs.streams) == 0 {218 return nil219 }220 return m.tickStreams()221}