zephinzer/godev

package main

import (
    "sync"
)

// ExecutionGroupCount keeps track of the execution group count for
// display in the verbose logs - helps to differentiate between
// the different execution groups
var ExecutionGroupCount = 0

// ExecutionGroup runs all commands in parallel
type ExecutionGroup struct {
    commands  []*Command
    waitGroup sync.WaitGroup
    logger    *Logger
}

// IsRunning is for the Runner to check if the execution group
// is still running
func (executionGroup *ExecutionGroup) IsRunning() bool {
    for _, command := range executionGroup.commands {
        if command.IsRunning() {
            return true
        }
    }
    return false
}

// Run starts the execution group's commands in parallel
// and waits for all of them to exit
func (executionGroup *ExecutionGroup) Run() {
    ExecutionGroupCount++
    defer executionGroup.logger.Debugf("execution group[%v] exited", ExecutionGroupCount)
    executionGroup.logger.Debugf("execution group[%v] is starting...", ExecutionGroupCount)
    for _, command := range executionGroup.commands {
        if err := command.IsValid(); err != nil {
            executionGroup.logger.Error(err)
        } else {
            go func(commandStatus *chan error) {
                for {
                    select {
                    case err := <-*commandStatus: // Command letting us know its done
                        executionGroup.handleCommandStatus(command, err)
                        return
                    default:
                    }
                }
            }(command.GetStatus())
            executionGroup.logger.Tracef("command[%s] is starting", command.GetID())
            executionGroup.waitGroup.Add(1)
            go command.Run()
        }
    }
    executionGroup.logger.Tracef("waiting for commands to complete running...")
    executionGroup.waitGroup.Wait()
}

// Terminate terminates this instance of the execution group, used when
// the Runner receives a signal to start a new pipeline
func (executionGroup *ExecutionGroup) Terminate() {
    for _, command := range executionGroup.commands {
        if command.IsRunning() {
            executionGroup.logger.Tracef("sending SIGINT to command %v", command.GetID())
            command.SendInterrupt()
            executionGroup.logger.Tracef("SIGINT sent to command %v", command.GetID())
        }
    }
}

func (executionGroup *ExecutionGroup) handleCommandStatus(command *Command, err error) {
    defer func() {
        if r := recover(); r != nil {
            executionGroup.logger.Warn(r)
        }
    }()
    if err != nil {
        executionGroup.logger.Warnf("command[%s] exited with: %s", command.GetID(), err)
    } else {
        executionGroup.logger.Debugf("command[%s] exited without error", command.GetID())
    }
    executionGroup.waitGroup.Done()
}