feat: add concurrent and condition logic

.vscode/settings.json

@@ -22,6 +22,7 @@
         "nehemming",
         "nolint",
         "onfail",
+        "ref'd",
         "runbook",
         "testdata",
         "yamlformatter"

pkg/builtin/cleaner.go

@@ -96,12 +96,21 @@ func buildRemoveOp(capComm *rocket.CapComm, config *Remove) (rocket.ExecuteFunc,
 		}
 
 		// clean
-		return deleteFiles(files, config.Log)
+		return deleteFiles(files, getLogFromCapComm(capComm, config.Log))
 	}
 
 	return fn, nil
 }
 
+// getLogFromCapComm returns the capComm logger or nil based on the log bool value.
+func getLogFromCapComm(capComm *rocket.CapComm, log bool) loggee.Logger {
+	if log {
+		return capComm.Log()
+	}
+
+	return nil
+}
+
 func globFile(specs []string) ([]string, error) {
 	files := make([]string, 0, len(specs))
 	for _, spec := range specs {
@@ -115,7 +124,7 @@ func globFile(specs []string) ([]string, error) {
 	return toDistinctStrings(files...), nil
 }
 
-func deleteFiles(files []string, log bool) error {
+func deleteFiles(files []string, log loggee.Logger) error {
 	for _, file := range files {
 		stat, err := os.Stat(filepath.FromSlash(file))
 		if err != nil && !os.IsNotExist(err) {
@@ -137,8 +146,8 @@ func deleteFiles(files []string, log bool) error {
 		}
 
 		// log
-		if log {
-			loggee.Infof("removed %s", friendlyRelativePath(file))
+		if log != nil {
+			log.Infof("removed %s", friendlyRelativePath(file))
 		}
 	}
 

pkg/builtin/cleaner_test.go

@@ -133,7 +133,7 @@ func removeRun(t *testing.T, missionName string) {
 func TestDeleteMissing(t *testing.T) {
 	files := []string{"filenotexists.gogo"}
 
-	err := deleteFiles(files, false)
+	err := deleteFiles(files, nil)
 	if err != nil {
 		t.Error("unexpected", err)
 	}

pkg/builtin/copy.go

@@ -103,7 +103,7 @@ func (copyType) Prepare(ctx context.Context, capComm *rocket.CapComm, task rocke
 		}
 
 		// copy
-		return copyFiles(execCtx, files, destSpec, overwrite, copyCfg.Log)
+		return copyFiles(execCtx, files, destSpec, overwrite, getLogFromCapComm(capComm, copyCfg.Log))
 	}
 
 	return fn, nil
@@ -212,7 +212,7 @@ func toDistinctAbsRelSlice(files ...AbsRel) []AbsRel {
 	return res
 }
 
-func copyFiles(ctx context.Context, sources []AbsRel, dest DestSpec, allowOverwrite, log bool) error {
+func copyFiles(ctx context.Context, sources []AbsRel, dest DestSpec, allowOverwrite bool, log loggee.Logger) error {
 	for _, source := range sources {
 		if ctx.Err() != nil {
 			return ctx.Err()
@@ -225,7 +225,7 @@ func copyFiles(ctx context.Context, sources []AbsRel, dest DestSpec, allowOverwr
 	return nil
 }
 
-func copyFile(source AbsRel, dest DestSpec, allowOverwrite, log bool) error {
+func copyFile(source AbsRel, dest DestSpec, allowOverwrite bool, log loggee.Logger) error {
 	// Get the source files permission
 	stat, err := os.Stat(source.Abs)
 	if err != nil {
@@ -243,10 +243,11 @@ func copyFile(source AbsRel, dest DestSpec, allowOverwrite, log bool) error {
 	if err != nil {
 		return err
 	}
+
 	if destAbsRel == nil || source.Abs == destAbsRel.Abs {
 		// skipping
-		if log {
-			loggee.Infof("skipping %s", source.Rel)
+		if log != nil {
+			log.Infof("skipping %s", source.Rel)
 		}
 		return nil
 	}
@@ -264,8 +265,8 @@ func copyFile(source AbsRel, dest DestSpec, allowOverwrite, log bool) error {
 	}
 
 	// log
-	if log {
-		loggee.Infof("copy %s => %s", source.Rel, destAbsRel.Rel)
+	if log != nil {
+		log.Infof("copy %s => %s", source.Rel, destAbsRel.Rel)
 	}
 
 	return nil

pkg/builtin/fetch.go

@@ -1,3 +1,19 @@
+/*
+Copyright (c) 2021 The cirocket Authors (Neil Hemming)
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package builtin
 
 import (
@@ -6,7 +22,6 @@ import (
 	"sync"
 
 	"github.com/mitchellh/mapstructure"
-	"github.com/nehemming/cirocket/pkg/loggee"
 	"github.com/nehemming/cirocket/pkg/rocket"
 	"github.com/pkg/errors"
 )
@@ -81,7 +96,7 @@ func (fetchType) Prepare(ctx context.Context, capComm *rocket.CapComm, task rock
 					err = e
 				} else {
 					// Log others
-					loggee.Error(e.Error())
+					capComm.Log().Error(e.Error())
 				}
 
 				// Signal stop processing other requests

pkg/builtin/mkdir.go

@@ -1,3 +1,19 @@
+/*
+Copyright (c) 2021 The cirocket Authors (Neil Hemming)
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package builtin
 
 import (
@@ -6,7 +22,6 @@ import (
 	"path/filepath"
 
 	"github.com/mitchellh/mapstructure"
-	"github.com/nehemming/cirocket/pkg/loggee"
 	"github.com/nehemming/cirocket/pkg/rocket"
 	"github.com/pkg/errors"
 )
@@ -56,7 +71,7 @@ func (mkDirType) Prepare(ctx context.Context, capComm *rocket.CapComm, task rock
 
 			// log
 			if mkDirCfg.Log {
-				loggee.Infof("created %s", dir)
+				capComm.Log().Infof("created %s", dir)
 			}
 		}
 

pkg/builtin/move.go

@@ -95,13 +95,13 @@ func (moveType) Prepare(ctx context.Context, capComm *rocket.CapComm, task rocke
 		}
 
 		// copy
-		return moveFiles(execCtx, files, destSpec, overwrite, moveCfg.Log)
+		return moveFiles(execCtx, files, destSpec, overwrite, getLogFromCapComm(capComm, moveCfg.Log))
 	}
 
 	return fn, nil
 }
 
-func moveFiles(ctx context.Context, sources []AbsRel, dest DestSpec, allowOverwrite, log bool) error {
+func moveFiles(ctx context.Context, sources []AbsRel, dest DestSpec, allowOverwrite bool, log loggee.Logger) error {
 	for _, source := range sources {
 		if ctx.Err() != nil {
 			return ctx.Err()
@@ -114,7 +114,7 @@ func moveFiles(ctx context.Context, sources []AbsRel, dest DestSpec, allowOverwr
 	return nil
 }
 
-func moveFile(source AbsRel, dest DestSpec, allowOverwrite, log bool) error {
+func moveFile(source AbsRel, dest DestSpec, allowOverwrite bool, log loggee.Logger) error {
 	// Get the source files permission
 	stat, err := os.Stat(source.Abs)
 	if err != nil {
@@ -132,8 +132,8 @@ func moveFile(source AbsRel, dest DestSpec, allowOverwrite, log bool) error {
 	}
 	if destAbsRel == nil || source.Abs == destAbsRel.Abs {
 		// skipping
-		if log {
-			loggee.Infof("skipping %s", source.Rel)
+		if log != nil {
+			log.Infof("skipping %s", source.Rel)
 		}
 		return nil
 	}
@@ -145,8 +145,8 @@ func moveFile(source AbsRel, dest DestSpec, allowOverwrite, log bool) error {
 	}
 
 	// log
-	if log {
-		loggee.Infof("move %s => %s", source.Rel, destAbsRel.Rel)
+	if log != nil {
+		log.Infof("move %s => %s", source.Rel, destAbsRel.Rel)
 	}
 
 	return nil

pkg/builtin/run.go

@@ -24,7 +24,6 @@ import (
 
 	"github.com/mitchellh/mapstructure"
 	"github.com/nehemming/cirocket/pkg/cliparse"
-	"github.com/nehemming/cirocket/pkg/loggee"
 	"github.com/nehemming/cirocket/pkg/rocket"
 	"github.com/pkg/errors"
 )
@@ -126,7 +125,7 @@ func getCommandLine(ctx context.Context, capComm *rocket.CapComm, runCfg *Run) (
 	return commandLine, nil
 }
 
-func startProcessSignalHandlee(ctx context.Context, cmd *exec.Cmd) chan struct{} {
+func startProcessSignalHandlee(ctx context.Context, capComm *rocket.CapComm, cmd *exec.Cmd) chan struct{} {
 	done := make(chan struct{})
 
 	go func() {
@@ -136,7 +135,7 @@ func startProcessSignalHandlee(ctx context.Context, cmd *exec.Cmd) chan struct{}
 			if cmd.Process != nil {
 				err := cmd.Process.Signal(os.Interrupt)
 				if err != nil {
-					loggee.Warnf("run signal error: %s", err)
+					capComm.Log().Warnf("run signal error: %s", err)
 				}
 			}
 		case <-done:
@@ -185,7 +184,7 @@ func runCmd(ctx context.Context, capComm *rocket.CapComm, cmd *exec.Cmd) error {
 	}
 
 	// setup signal handler and close on exit
-	signalHandlerDoneChannel := startProcessSignalHandlee(ctx, cmd)
+	signalHandlerDoneChannel := startProcessSignalHandlee(ctx, capComm, cmd)
 	defer close(signalHandlerDoneChannel)
 
 	// Wait for process exit

pkg/rocket/assemble.go

@@ -36,7 +36,7 @@ import (
 func (mc *missionControl) Assemble(ctx context.Context, blueprintName string, sources []string, runbook string, params Params) error {
 	// blueprint, if not abs need to search sources to locate
 	// once blueprint found extract it
-	blueprint, blueprintLocation, err := mc.searchSources(ctx, blueprintName, sources, mc.missionLog())
+	blueprint, blueprintLocation, err := mc.searchSources(ctx, blueprintName, sources)
 	if err != nil {
 		return err
 	}
@@ -155,7 +155,7 @@ func mergeParamSets(sourceParams []Param, additional ...Param) []Param {
 const manifestFileName = "/blueprint.yml"
 
 func (mc *missionControl) searchSources(ctx context.Context, blueprintName string,
-	sources []string, log loggee.Logger) (*Blueprint, string, error) {
+	sources []string) (*Blueprint, string, error) {
 	blueprintName = filepath.ToSlash(blueprintName)
 
 	if strings.HasSuffix(blueprintName, manifestFileName) {
@@ -179,7 +179,7 @@ func (mc *missionControl) searchSources(ctx context.Context, blueprintName strin
 		return nil, "", fmt.Errorf("name %s is malformed", blueprintName)
 	}
 
-	return loadBlueprint(ctx, blueprintName, log, sources)
+	return loadBlueprint(ctx, blueprintName, mc.missionLog(), sources)
 }
 
 func loadBlueprint(ctx context.Context, blueprintName string,