finish s/pipeline/flow/g

Signed-off-by: Tony Worm <tony@hofstadter.io>

cmd/cuetils/cmd/flow.go

@@ -7,7 +7,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/hofstadter-io/cuetils/cmd/cuetils/flags"
-	"github.com/hofstadter-io/cuetils/pipeline"
+	"github.com/hofstadter-io/cuetils/flow"
 	"github.com/hofstadter-io/cuetils/structural"
 )
 
@@ -23,7 +23,7 @@ func init() {
 
 func FlowRun(globs []string) (err error) {
 
-	results, err := pipeline.Run(globs, &flags.RootPflags, &flags.FlowFlags)
+	results, err := flow.Run(globs, &flags.RootPflags, &flags.FlowFlags)
 	if err != nil {
 		return err
 	}

flow/pipe/pipeline.go → flow/flow/flow.go

@@ -1,4 +1,4 @@
-package pipe
+package flow
 
 import (
 	"fmt"
@@ -49,7 +49,7 @@ func (P *Flow) run(val cue.Value) error {
 
   final, err := P.Ctrl.Run(P.Context.Context)
 
-  // fmt.Println("pipe(end):", P.path, P.rpath)
+  // fmt.Println("flow(end):", P.path, P.rpath)
   P.Final = final
   if err != nil {
     s := structural.FormatCueError(err)

flow/pipelines.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/hofstadter-io/cuetils/cmd/cuetils/flags"
 	"github.com/hofstadter-io/cuetils/flow/context"
-	"github.com/hofstadter-io/cuetils/flow/pipe"
+	"github.com/hofstadter-io/cuetils/flow/flow"
 	"github.com/hofstadter-io/cuetils/structural"
 	"github.com/hofstadter-io/cuetils/utils"
 )
@@ -107,8 +107,8 @@ func listFlows(val cue.Value,  opts *flags.RootPflagpole, popts *flags.FlowFlagp
 // maybe this becomes recursive so we can find
 // nested flows, but not recurse when we find one
 // only recurse when we have something which is not a flow or task
-func findFlows(ctx *context.Context, val cue.Value, opts *flags.RootPflagpole, popts *flags.FlowFlagpole) ([]*pipe.Flow, error) {
-  pipes := []*pipe.Flow{}
+func findFlows(ctx *context.Context, val cue.Value, opts *flags.RootPflagpole, popts *flags.FlowFlagpole) ([]*flow.Flow, error) {
+  flows := []*flow.Flow{}
 
   // TODO, look for lists?
   s, err := val.Struct()
@@ -124,15 +124,15 @@ func findFlows(ctx *context.Context, val cue.Value, opts *flags.RootPflagpole, p
   _, found, keep := hasFlowAttr(val, args)
   if keep  {
     // invoke TaskFactory
-    p, err := pipe.NewFlow(ctx, val)
+    p, err := flow.NewFlow(ctx, val)
     if err != nil {
-      return pipes, err
+      return flows, err
     }
-    pipes = append(pipes, p)
+    flows = append(flows, p)
   }
 
   if found {
-    return pipes, nil
+    return flows, nil
   }
 
   iter := s.Fields(
@@ -146,11 +146,11 @@ func findFlows(ctx *context.Context, val cue.Value, opts *flags.RootPflagpole, p
 
     _, found, keep := hasFlowAttr(v, args)
     if keep  {
-      p, err := pipe.NewFlow(ctx, v)
+      p, err := flow.NewFlow(ctx, v)
       if err != nil {
-        return pipes, err
+        return flows, err
       }
-      pipes = append(pipes, p)
+      flows = append(flows, p)
     }
 
     // break recursion if flow found
@@ -161,11 +161,11 @@ func findFlows(ctx *context.Context, val cue.Value, opts *flags.RootPflagpole, p
     // recurse to search for more flows
     ps, err := findFlows(ctx, v, opts, popts)
     if err != nil {
-      return pipes, nil 
+      return flows, nil 
     }
-    pipes = append(pipes, ps...)
+    flows = append(flows, ps...)
   }
 
-  return pipes, nil
+  return flows, nil
 }
 

flow/run.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/hofstadter-io/cuetils/cmd/cuetils/flags"
 	"github.com/hofstadter-io/cuetils/flow/context"
-	"github.com/hofstadter-io/cuetils/flow/pipe"
+	"github.com/hofstadter-io/cuetils/flow/flow"
 	_ "github.com/hofstadter-io/cuetils/flow/tasks" // ensure tasks register
 	"github.com/hofstadter-io/cuetils/structural"
 	// "github.com/hofstadter-io/cuetils/utils"
@@ -44,11 +44,11 @@ func run(globs []string, opts *flags.RootPflagpole, popts *flags.FlowFlagpole) (
 
   // sharedCtx := buildSharedContext
 
-	// (refactor/pipe/many) find  flows
-  pipes := []*pipe.Flow{}
+	// (refactor/flow/many) find  flows
+  flows := []*flow.Flow{}
 	for _, in := range ins {
 
-    // (refactor/pipe/solo)
+    // (refactor/flow/solo)
     val := in.Value
 
 
@@ -101,21 +101,21 @@ func run(globs []string, opts *flags.RootPflagpole, popts *flags.FlowFlagpole) (
     if err != nil {
       return nil, err
     }
-    pipes = append(pipes, ps...)
+    flows = append(flows, ps...)
 	}
 
   if popts.List {
     return nil, nil
   }
 
-  if len(pipes) == 0 {
+  if len(flows) == 0 {
     return nil, fmt.Errorf("no flows found")
   }
 
   // start all of the flows
   // TODO, use wait group, accume errors, flag for failure modes
-  for _, pipe := range pipes {
-    err := pipe.Start()
+  for _, flow := range flows {
+    err := flow.Start()
     if err != nil {
       return nil, err
     }

flow/tasks/api/serve.go

@@ -14,7 +14,7 @@ import (
   "github.com/labstack/echo-contrib/prometheus"
 
   "github.com/hofstadter-io/cuetils/flow/context"
-  "github.com/hofstadter-io/cuetils/flow/pipe"
+  "github.com/hofstadter-io/cuetils/flow/flow"
 )
 
 func init() {
@@ -166,7 +166,7 @@ func (T *Serve) routeFromValue(path string, route cue.Value, e *echo.Echo, ctx *
     }
 
     if isPipe {
-      p, err := pipe.NewFlow(ctx, tmp)
+      p, err := flow.NewFlow(ctx, tmp)
       if err != nil {
         return err
       }

flow/tasks/msg/irc-client.go

@@ -10,7 +10,7 @@ import (
 	"gopkg.in/irc.v3"
 
   "github.com/hofstadter-io/cuetils/flow/context"
-  "github.com/hofstadter-io/cuetils/flow/pipe"
+  "github.com/hofstadter-io/cuetils/flow/flow"
 	"github.com/hofstadter-io/cuetils/utils"
 )
 
@@ -213,11 +213,11 @@ func buildIrcHandler(ct_ctx *context.Context, val cue.Value) (irc.HandlerFunc, e
     // is this a flow
     errV := v.LookupPath(cue.ParsePath("error"))
     respV := v.LookupPath(cue.ParsePath("resp"))
-    pipeV := v.LookupPath(cue.ParsePath("pipe"))
+    flowV := v.LookupPath(cue.ParsePath("flow"))
 
     fmt.Println("errV:", errV)
     fmt.Println("respV:", respV)
-    fmt.Println("pipeV:", pipeV)
+    fmt.Println("flowV:", flowV)
 
     // log any errors
     if errV.Exists() {
@@ -239,37 +239,37 @@ func buildIrcHandler(ct_ctx *context.Context, val cue.Value) (irc.HandlerFunc, e
     }
 
     // handle flows
-    if pipeV.Exists() {
+    if flowV.Exists() {
       // build new value
       v := ctx.CompileString("{...}")
-      v = v.Unify(pipeV) 
+      v = v.Unify(flowV) 
 
-      p, err := pipe.NewFlow(ct_ctx, v)
+      p, err := flow.NewFlow(ct_ctx, v)
       if err != nil {
-        fmt.Println("Error(pipe/new):", err)
+        fmt.Println("Error(flow/new):", err)
         return
       }
 
       err = p.Start()
       if err != nil {
-        fmt.Println("Error(pipe/run):", err)
+        fmt.Println("Error(flow/run):", err)
         return
       }
 
       rV := p.Final.LookupPath(cue.ParsePath("resp"))
       if !rV.Exists() {
-        fmt.Println("Error(pipe/resp): does not exist")
+        fmt.Println("Error(flow/resp): does not exist")
         return 
       }
       s, err := rV.String()
       if err != nil {
-        fmt.Println("Error(pipe/rVstr):", err)
+        fmt.Println("Error(flow/rVstr):", err)
         return
       }
 
       // fill in go-irc.Message and then turn that into a string
 
-      fmt.Println("sending(pipe/msg):", s)
+      fmt.Println("sending(flow/msg):", s)
       c.Writef("PRIVMSG %s :%s", channel, s)
 
       return

flow/tests/txtar/readfile_001.txt

@@ -1,7 +1,7 @@
-exec cuetils flow pipe.cue
+exec cuetils flow flow.cue
 # cmp stdout golden.stdout
 
--- pipe.cue --
+-- flow.cue --
 tasks: {
   @flow(readfile)
 	r: { f: "in.txt", contents: string } @task(os.ReadFile)

flow/tests/txtar/readfile_002.txt

@@ -1,7 +1,7 @@
-exec cuetils flow pipe.cue
+exec cuetils flow flow.cue
 # cmp stdout golden.stdout
 
--- pipe.cue --
+-- flow.cue --
 tasks: {
   @flow(readfile)
 	r: { f: "in.json", contents: string } @task(os.ReadFile)

flow/tests/txtar/readfile_003.txt

@@ -1,7 +1,7 @@
-! exec cuetils flow pipe.cue
+! exec cuetils flow flow.cue
 cmp stdout golden.stdout
 
--- pipe.cue --
+-- flow.cue --
 tasks: {
   @flow(readfile)
 	r: { f: "in.txt", contents: string } @task(os.ReadFile)

ideas.md

@@ -9,8 +9,11 @@ Enhance & expand
     - [x] file append
     - [x] mkdir
     - [x] memory store / load
-- [ ] rename pipeline -> run
+- [x] rename pipeline -> run
 - [ ] merge and release
+
+---
+
 - [ ] hof/flow cmd
 
 ### Docs...