1package nushell23import (4 "context"5 "fmt"6 "io"7 "strings"8)910// Streams: a cell run as a stream prints each value its pipeline yields11// as soon as the pipeline yields it, one NUON value to a line, rather12// than the whole value once the pipeline ends, so a pipeline that never13// ends (tail -f app.log | lines, watch) is followed as it goes. The14// lines are what nuon.Reader reads as a stream of rows.1516// StreamScript is what nu runs for job as a stream: the tables bound to17// their names, as Script binds them, then the pipeline, each value it18// yields printed as NUON on a line of its own.19func StreamScript(job Job, names []string) string {20 var b strings.Builder21 for i, name := range names {22 fmt.Fprintf(&b, "let %s = (open --raw $env.NU012_TABLE_%d | from nuon)\n", name, i)23 }24 b.WriteString("do {\n" + job.Command + "\n} | each {|row| $row | to nuon --raw | print } | ignore\n")25 return b.String()26}2728// Stream runs job as a stream, writing to w a line of NUON for each29// value the pipeline yields, until the pipeline ends or ctx is done. It30// has no timeout: the stream runs until it's stopped.31func Stream(ctx context.Context, r Runner, job Job, w io.Writer) error {32 err := r.Run(ctx, job, StreamScript(job, sortedNames(job.Tables)), w)33 if ctx.Err() != nil {34 return context.Cause(ctx)35 }36 return err37}