Added the notion of streams

This includes a stream to a file

cmd/cmsh/main.go

@@ -23,7 +23,7 @@ func main() {
 			break
 		}
 
-		if err := inst.Eval(ctx, line); err != nil {
+		if err := inst.EvalAndDisplay(ctx, line); err != nil {
 			log.Printf("%T: %v", err, err)
 		}
 	}

cmdlang/ast.go

@@ -19,8 +19,13 @@ type astCmd struct {
 	Args []astCmdArg `parser:"@@*"`
 }
 
-var parser = participle.MustBuild[astCmd]()
+type astPipeline struct {
+	First *astCmd   `parser:"@@"`
+	Rest  []*astCmd `parser:"( '|' @@ )*"`
+}
 
-func parse(r io.Reader) (*astCmd, error) {
+var parser = participle.MustBuild[astPipeline]()
+
+func parse(r io.Reader) (*astPipeline, error) {
 	return parser.Parse("test", r)
 }

cmdlang/builtins.go

@@ -1,21 +1,104 @@
 package cmdlang
 
 import (
+	"bufio"
 	"context"
-	"log"
+	"errors"
+	"io"
+	"os"
 	"strings"
 )
 
-func echoBuiltin(ctx context.Context, args invocationArgs) error {
+func echoBuiltin(ctx context.Context, args invocationArgs) (object, error) {
 	if len(args.args) == 0 {
-		log.Print()
-		return nil
+		return asStream(""), nil
 	}
 
 	var line strings.Builder
 	for _, arg := range args.args {
 		line.WriteString(arg)
 	}
-	log.Print(line.String())
+
+	return asStream(line.String()), nil
+}
+
+func toUpperBuiltin(ctx context.Context, args invocationArgs) (object, error) {
+	// Handle args
+	return mapFilterStream{
+		in: args.inStream,
+		mapFn: func(x object) (object, bool) {
+			s, ok := x.(string)
+			if !ok {
+				return nil, false
+			}
+			return strings.ToUpper(s), true
+		},
+	}, nil
+}
+
+func catBuiltin(ctx context.Context, args invocationArgs) (object, error) {
+	if err := args.expectArgn(1); err != nil {
+		return nil, err
+	}
+
+	return &fileLinesStream{filename: args.args[0]}, nil
+}
+
+type fileLinesStream struct {
+	filename string
+	f        *os.File
+	scnr     *bufio.Scanner
+}
+
+func (f *fileLinesStream) next() (object, error) {
+	var err error
+
+	// We open the file on the first pull. That way, an unconsumed stream won't result in a FD leak
+	if f.f == nil {
+		f.f, err = os.Open(f.filename)
+		if err != nil {
+			return nil, err
+		}
+		f.scnr = bufio.NewScanner(f.f)
+	}
+
+	if f.scnr.Scan() {
+		return f.scnr.Text(), nil
+	}
+	if f.scnr.Err() == nil {
+		return nil, io.EOF
+	}
+	return nil, f.scnr.Err()
+}
+
+func (f *fileLinesStream) close() error {
+	if f.f != nil {
+		return f.f.Close()
+	}
+	return nil
+}
+
+func errorTestBuiltin(ctx context.Context, args invocationArgs) (object, error) {
+	return &timeBombStream{args.inStream, 2}, nil
+}
+
+type timeBombStream struct {
+	in stream
+	x  int
+}
+
+func (ms *timeBombStream) next() (object, error) {
+	if ms.x > 0 {
+		ms.x--
+		return ms.in.next()
+	}
+	return nil, errors.New("BOOM")
+}
+
+func (ms *timeBombStream) close() error {
+	closable, ok := ms.in.(closableStream)
+	if ok {
+		return closable.close()
+	}
 	return nil
 }

cmdlang/egbuiltins.go

@@ -0,0 +1,7 @@
+package cmdlang
+
+import "context"
+
+func egLookup(ctx context.Context, args invocationArgs) (object, error) {
+	return nil, nil
+}

cmdlang/env.go

@@ -5,8 +5,16 @@ import (
 )
 
 type evalCtx struct {
-	parent   *evalCtx
-	commands map[string]invokable
+	parent        *evalCtx
+	currentStream stream
+	commands      map[string]invokable
+}
+
+func (ec *evalCtx) withCurrentStream(s stream) *evalCtx {
+	return &evalCtx{
+		parent:        ec,
+		currentStream: s,
+	}
 }
 
 func (ec *evalCtx) addCmd(name string, inv invokable) {
@@ -19,11 +27,11 @@ func (ec *evalCtx) addCmd(name string, inv invokable) {
 
 func (ec *evalCtx) lookupCmd(name string) (invokable, error) {
 	for e := ec; e != nil; e = e.parent {
-		if ec.commands == nil {
+		if e.commands == nil {
 			continue
 		}
 
-		if cmd, ok := ec.commands[name]; ok {
+		if cmd, ok := e.commands[name]; ok {
 			return cmd, nil
 		}
 

cmdlang/eval.go

@@ -10,21 +10,42 @@ import (
 type evaluator struct {
 }
 
-func (e evaluator) evaluate(ctx context.Context, ec *evalCtx, ast *astCmd) error {
+func (e evaluator) evaluate(ctx context.Context, ec *evalCtx, n *astPipeline) (object, error) {
+	res, err := e.evalCmd(ctx, ec, n.First)
+	if err != nil {
+		return nil, err
+	}
+	if len(n.Rest) == 0 {
+		return res, nil
+	}
+
+	// Command is a pipeline, so build it out
+	for _, rest := range n.Rest {
+		out, err := e.evalCmd(ctx, ec.withCurrentStream(asStream(res)), rest)
+		if err != nil {
+			return nil, err
+		}
+		res = out
+	}
+	return res, nil
+}
+
+func (e evaluator) evalCmd(ctx context.Context, ec *evalCtx, ast *astCmd) (object, error) {
 	cmd, err := ec.lookupCmd(ast.Name)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	args, err := slices.MapWithError(ast.Args, func(a astCmdArg) (string, error) {
 		return e.evalArg(ctx, ec, a)
 	})
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	return cmd.invoke(ctx, invocationArgs{
-		args: args,
+		args:     args,
+		inStream: ec.currentStream,
 	})
 }
 

cmdlang/inst.go

@@ -2,6 +2,7 @@ package cmdlang
 
 import (
 	"context"
+	"fmt"
 	"strings"
 )
 
@@ -12,6 +13,10 @@ type Inst struct {
 func New() *Inst {
 	rootEC := evalCtx{}
 	rootEC.addCmd("echo", invokableFunc(echoBuiltin))
+	rootEC.addCmd("toUpper", invokableFunc(toUpperBuiltin))
+	rootEC.addCmd("cat", invokableFunc(catBuiltin))
+
+	rootEC.addCmd("testTimebomb", invokableFunc(errorTestBuiltin))
 
 	return &Inst{
 		rootEC: &rootEC,
@@ -19,12 +24,31 @@ func New() *Inst {
 }
 
 // TODO: return value?
-func (inst *Inst) Eval(ctx context.Context, expr string) error {
+func (inst *Inst) Eval(ctx context.Context, expr string) (any, error) {
 	ast, err := parse(strings.NewReader(expr))
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	eval := evaluator{}
 	return eval.evaluate(ctx, inst.rootEC, ast)
 }
+
+func (inst *Inst) EvalAndDisplay(ctx context.Context, expr string) error {
+	res, err := inst.Eval(ctx, expr)
+	if err != nil {
+		return err
+	}
+
+	return inst.display(ctx, res)
+}
+
+func (inst *Inst) display(ctx context.Context, res object) (err error) {
+	switch v := res.(type) {
+	case stream:
+		return forEach(v, func(o object) error { return inst.display(ctx, o) })
+	case string:
+		fmt.Println(v)
+	}
+	return nil
+}

cmdlang/objs.go

@@ -1,17 +1,32 @@
 package cmdlang
 
-import "context"
+import (
+	"context"
+	"errors"
+	"strconv"
+)
+
+type object = any
 
 type invocationArgs struct {
-	args []string
+	args     []string
+	inStream stream
 }
 
+func (ia invocationArgs) expectArgn(x int) error {
+	if len(ia.args) < x {
+		return errors.New("expected at least " + strconv.Itoa(x) + " args")
+	}
+	return nil
+}
+
+// invokable is an object that can be executed as a command
 type invokable interface {
-	invoke(ctx context.Context, args invocationArgs) error
+	invoke(ctx context.Context, args invocationArgs) (object, error)
 }
 
-type invokableFunc func(ctx context.Context, args invocationArgs) error
+type invokableFunc func(ctx context.Context, args invocationArgs) (object, error)
 
-func (i invokableFunc) invoke(ctx context.Context, args invocationArgs) error {
+func (i invokableFunc) invoke(ctx context.Context, args invocationArgs) (object, error) {
 	return i(ctx, args)
 }

cmdlang/streams.go

@@ -0,0 +1,104 @@
+package cmdlang
+
+import (
+	"errors"
+	"io"
+)
+
+// stream is an object which returns a collection of objects from a source.
+// These are used to create pipelines
+type stream interface {
+	// next pulls the next object from the stream.  If an object is available, the result is the
+	// object and a nil error.  If no more objects are available, error returns io.EOF.
+	// Otherwise, an error is returned.
+	next() (object, error)
+}
+
+// forEach will iterate over all the items of a stream. The iterating function can return an error, which will
+// be returned as is. A stream that has consumed every item will return nil. The stream will automatically be closed.
+func forEach(s stream, f func(object) error) (err error) {
+	defer func() {
+		if c, ok := s.(closableStream); ok {
+			c.close()
+		}
+	}()
+
+	var sv object
+	for sv, err = s.next(); err == nil; sv, err = s.next() {
+		if err := f(sv); err != nil {
+			return err
+		}
+	}
+	if !errors.Is(err, io.EOF) {
+		return err
+	}
+	return nil
+}
+
+// closableStream is a stream that has opened resources that must be closed when the stream is
+// consumed.  The stream implementation can expect close to be called if at least one next() call is made.  Otherwise
+// closableStream cannot assume that close will be called (the pipe may be left unconsumed, for example).
+//
+// It is the job of the final iterator to call close. Any steam that consumes from another stream must
+// implement this, and call close on the parent stream.
+type closableStream interface {
+	stream
+
+	// close closes the stream
+	close() error
+}
+
+// asStream converts an object to a stream.  If t is already a stream, it's returned as is.
+// Otherwise, a singleton stream is returned.
+func asStream(v object) stream {
+	if s, ok := v.(stream); ok {
+		return s
+	}
+	return &singletonStream{t: v}
+}
+
+type emptyStream struct{}
+
+func (s emptyStream) next() (object, error) {
+	return nil, io.EOF
+}
+
+type singletonStream struct {
+	t        any
+	consumed bool
+}
+
+func (s *singletonStream) next() (object, error) {
+	if s.consumed {
+		return nil, io.EOF
+	}
+	s.consumed = true
+	return s.t, nil
+}
+
+type mapFilterStream struct {
+	in    stream
+	mapFn func(x object) (object, bool)
+}
+
+func (ms mapFilterStream) next() (object, error) {
+	for {
+		u, err := ms.in.next()
+		if err != nil {
+			return nil, err
+		}
+
+		t, ok := ms.mapFn(u)
+		if ok {
+			return t, nil
+		}
+	}
+}
+
+func (ms mapFilterStream) close() error {
+	closable, ok := ms.in.(closableStream)
+	if ok {
+		return closable.close()
+	}
+	return nil
+}