Merge pull request #51 from csaunders/worker-idle-notification

Adds a flag to touch a file when workers go idle

cmd/theme/main.go

@@ -138,11 +138,12 @@ func FileManipulationCommandParser(cmd string, args []string) (result map[string
 func WatchCommandParser(cmd string, args []string) (result map[string]interface{}, set *flag.FlagSet) {
 	result = make(map[string]interface{})
 	currentDir, _ := os.Getwd()
-	var environment, directory string
+	var environment, directory, notifyFile string
 
 	set = makeFlagSet(cmd)
 	set.StringVar(&environment, "env", themekit.DefaultEnvironment, "environment to run command")
 	set.StringVar(&directory, "dir", currentDir, "directory that config.yml is located")
+	set.StringVar(&notifyFile, "notify", "", "file to touch when workers have gone idle")
 	set.Parse(args)
 
 	client, err := loadThemeClient(directory, environment)
@@ -152,6 +153,7 @@ func WatchCommandParser(cmd string, args []string) (result map[string]interface{
 	}
 
 	result["themeClient"] = client
+	result["notifyFile"] = notifyFile
 
 	return
 }

commands/watch.go

@@ -2,13 +2,16 @@ package commands
 
 import (
 	"fmt"
-	"github.com/csaunders/themekit"
 	"os"
+	"time"
+
+	"github.com/csaunders/themekit"
 )
 
 type WatchOptions struct {
 	BasicOptions
-	Directory string
+	Directory  string
+	NotifyFile string
 }
 
 func WatchCommand(args map[string]interface{}) chan bool {
@@ -18,6 +21,7 @@ func WatchCommand(args map[string]interface{}) chan bool {
 	extractThemeClient(&options.Client, args)
 	extractEventLog(&options.EventLog, args)
 	extractString(&options.Directory, "directory", args)
+	extractString(&options.NotifyFile, "notifyFile", args)
 
 	return Watch(options)
 }
@@ -32,6 +36,12 @@ func Watch(options WatchOptions) chan bool {
 	bucket := themekit.NewLeakyBucket(config.BucketSize, config.RefillRate, 1)
 	bucket.TopUp()
 	foreman := themekit.NewForeman(bucket)
+	foreman.OnIdle = func() {
+		if len(options.NotifyFile) > 0 {
+			os.Create(options.NotifyFile)
+			os.Chtimes(options.NotifyFile, time.Now(), time.Now())
+		}
+	}
 	watcher := constructFileWatcher(options.Directory, config)
 	foreman.JobQueue = watcher
 	foreman.IssueWork()

foreman.go

@@ -1,10 +1,13 @@
 package themekit
 
+import "time"
+
 type Foreman struct {
 	bucket      *LeakyBucket
 	halt        chan bool
 	JobQueue    chan AssetEvent
 	WorkerQueue chan AssetEvent
+	OnIdle      func()
 }
 
 func NewForeman(bucket *LeakyBucket) Foreman {
@@ -13,21 +16,29 @@ func NewForeman(bucket *LeakyBucket) Foreman {
 		halt:        make(chan bool),
 		JobQueue:    make(chan AssetEvent),
 		WorkerQueue: make(chan AssetEvent),
+		OnIdle:      func() {},
 	}
 }
 
 func (f Foreman) IssueWork() {
 	f.bucket.StartDripping()
 	go func() {
+		notifyProcessed := false
 		for {
 			select {
 			case job := <-f.JobQueue:
 				f.bucket.GetDrop()
+				notifyProcessed = true
 				go func() {
 					f.WorkerQueue <- job
 				}()
 			case <-f.halt:
 				return
+			case <-time.Tick(1 * time.Second):
+				if notifyProcessed {
+					notifyProcessed = false
+					f.OnIdle()
+				}
 			}
 		}
 	}()