@@ -12,6 +12,8 @@ import (
1212 "strings"
1313 "time"
1414
15+ "golang.org/x/term"
16+
1517 "gopkg.in/yaml.v2"
1618
1719 "github.com/samber/lo"
@@ -44,9 +46,9 @@ func processRun(c *g.CliSC) (ok bool, err error) {
4446 Source : sling.Source {Options : & sling.SourceOptions {}},
4547 Target : sling.Target {Options : & sling.TargetOptions {}},
4648 }
47- replicationCfgPath := ""
48- pipelineCfgPath := ""
49- taskCfgStr := ""
49+
50+ var replicationCfgPath , pipelineCfgPath , directoryPath string
51+
5052 showExamples := false
5153 selectStreams := []string {}
5254
@@ -78,9 +80,42 @@ func processRun(c *g.CliSC) (ok bool, err error) {
7880 case "pipeline" :
7981 env .SetTelVal ("run_mode" , "pipeline" )
8082 pipelineCfgPath = cast .ToString (v )
81- case "config" :
82- env .SetTelVal ("run_mode" , "task" )
83- taskCfgStr = cast .ToString (v )
83+ case "directory" :
84+ env .SetTelVal ("run_mode" , "directory" )
85+ directoryPath = cast .ToString (v )
86+ case "path" :
87+ filePath := cast .ToString (v )
88+ fileInfo , err := os .Stat (filePath )
89+ if err != nil {
90+ return true , g .Error (err , "error accessing path: %s" , filePath )
91+ }
92+
93+ if fileInfo .IsDir () {
94+ env .SetTelVal ("run_mode" , "directory" )
95+ directoryPath = filePath
96+ } else {
97+ dirPath , err := os .Getwd ()
98+ if err != nil {
99+ return true , g .Error (err , "unable to determine working dir. Try using flags" )
100+ }
101+ runFile := sling.RunFile {File : g .InfoToFileItem (dirPath , filePath , fileInfo )}
102+ valid , err := runFile .IsValid ()
103+ if err != nil {
104+ return true , g .Error (err , "could not load file: %s" , filePath )
105+ } else if valid {
106+ switch runFile .Type {
107+ case sling .RunFileReplication :
108+ env .SetTelVal ("run_mode" , "replication" )
109+ replicationCfgPath = filePath
110+ case sling .RunFilePipeline :
111+ env .SetTelVal ("run_mode" , "pipeline" )
112+ pipelineCfgPath = filePath
113+ }
114+ } else {
115+ return true , g .Error ("file is not a valid replication or pipeline: %s" , filePath )
116+ }
117+ }
118+
84119 case "src-conn" :
85120 cfg .Source .Conn = cast .ToString (v )
86121 case "src-stream" , "src-table" , "src-sql" , "src-file" :
@@ -191,11 +226,7 @@ func processRun(c *g.CliSC) (ok bool, err error) {
191226 return ok , nil
192227 }
193228
194- if val := os .Getenv ("SLING_TASK_CONFIG" ); val != "" {
195- taskCfgStr = val
196- }
197-
198- if taskCfgStr != "" {
229+ if taskCfgStr := os .Getenv ("SLING_TASK_CONFIG" ); taskCfgStr != "" {
199230 err = cfg .Unmarshal (taskCfgStr )
200231 if err != nil {
201232 return ok , g .Error (err , "could not parse task configuration" )
@@ -229,7 +260,12 @@ runReplication:
229260 defer printUpdateAvailable ()
230261 }
231262
232- if pipelineCfgPath != "" {
263+ if directoryPath != "" {
264+ err = runDirectory (directoryPath )
265+ if err != nil {
266+ return ok , g .Error (err , "failure running directory (see docs @ https://docs.slingdata.io)" )
267+ }
268+ } else if pipelineCfgPath != "" {
233269 err = runPipeline (pipelineCfgPath )
234270 if err != nil {
235271 return ok , g .Error (err , "failure running pipeline (see docs @ https://docs.slingdata.io)" )
@@ -654,6 +690,82 @@ func runPipeline(pipelineCfgPath string) (err error) {
654690 return
655691}
656692
693+ func runDirectory (directoryPath string ) (err error ) {
694+ if ! g .PathExists (directoryPath ) {
695+ return g .Error ("directory does not exist: %s" , directoryPath )
696+ }
697+
698+ files , err := g .ListDirRecursive (directoryPath )
699+ if err != nil {
700+ return g .Error (err , "could not list files from %s" , directoryPath )
701+ }
702+
703+ // collect replication or pipeline
704+ runFiles := sling.RunFiles {}
705+ for _ , file := range files {
706+ if strings .HasSuffix (file .Name , ".yaml" ) || strings .HasSuffix (file .Name , ".yml" ) {
707+ runFile := sling.RunFile {File : file }
708+ valid , err := runFile .IsValid ()
709+ if err != nil {
710+ return g .Error (err , "could not load file: %s" , file .RelPath )
711+ } else if valid {
712+ runFiles = append (runFiles , runFile )
713+ }
714+ }
715+ }
716+
717+ // get terminal width, fallback to 80 if unable to determine
718+ width := 80
719+ if w , _ , err := term .GetSize (int (os .Stdout .Fd ())); err == nil {
720+ width = int (float64 (w ) * 0.75 )
721+ }
722+
723+ // change working dir
724+ if err = os .Chdir (directoryPath ); err != nil {
725+ return g .Error (err , "could not change working directory to %s" , directoryPath )
726+ }
727+
728+ // run them
729+ startTime := time .Now ()
730+ eG := g.ErrorGroup {}
731+ successes := 0
732+ for _ , runFile := range runFiles .Order () {
733+ ctx = g .NewContext (context .Background ()) // reset master context
734+
735+ fileName := g .F ("%s (%s)" , runFile .File .RelPath , runFile .Type )
736+ remainingWidth := width - len (fileName ) - 2 // subtract 2 for spaces around filename
737+ leftSigns := remainingWidth / 2
738+ rightSigns := remainingWidth - leftSigns // accounts for odd numbers
739+ env .Println (env .GreenString (g .F ("%s %s %s" , strings .Repeat ("=" , leftSigns ), fileName , strings .Repeat ("=" , rightSigns ))))
740+
741+ switch runFile .Type {
742+ case sling .RunFilePipeline :
743+ err = runPipeline (runFile .File .RelPath )
744+ case sling .RunFileReplication :
745+ err = runReplication (runFile .File .RelPath , nil )
746+ }
747+ if eG .Capture (err , runFile .File .RelPath ) {
748+ g .LogError (err )
749+ } else {
750+ successes ++
751+ }
752+ }
753+
754+ delta := time .Since (startTime )
755+ env .Println (env .GreenString (strings .Repeat ("=" , width )))
756+ successStr := env .GreenString (g .F ("%d Successes" , successes ))
757+ failureStr := g .F ("%d Failures" , len (eG .Errors ))
758+ if len (eG .Errors ) > 0 {
759+ failureStr = env .RedString (failureStr )
760+ } else {
761+ failureStr = env .GreenString (failureStr )
762+ }
763+
764+ g .Info ("Sling Run Completed in %s | %s | %s\n " , g .DurationString (delta ), successStr , failureStr )
765+
766+ return eG .Err ()
767+ }
768+
657769func parsePayload (payload string , validate bool ) (options map [string ]any , err error ) {
658770 payload = strings .TrimSpace (payload )
659771 if payload == "" {
0 commit comments