feat: add watchman integration

Author: cdmistman

src/app/process.go

@@ -5,8 +5,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/cakturk/go-netstat/netstat"
-	"github.com/f1bonacc1/process-compose/src/types"
 	"io"
 	"math/rand"
 	"os"
@@ -17,6 +15,9 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/cakturk/go-netstat/netstat"
+	"github.com/f1bonacc1/process-compose/src/types"
+
 	"github.com/f1bonacc1/process-compose/src/command"
 	"github.com/f1bonacc1/process-compose/src/health"
 	"github.com/f1bonacc1/process-compose/src/pclog"
@@ -65,6 +66,13 @@ type Process struct {
 	isMain           bool
 	extraArgs        []string
 	isStopped        atomic.Bool
+	isDevRestart     atomic.Bool
+	devWatchers      []devWatch
+}
+
+type devWatch struct {
+	sub    *WatchmanSub
+	config *types.Watch
 }
 
 func NewProcess(
@@ -77,6 +85,7 @@ func NewProcess(
 	printLogs bool,
 	isMain bool,
 	extraArgs []string,
+	watchman *Watchman,
 ) *Process {
 	colNumeric := rand.Intn(int(color.FgHiWhite)-int(color.FgHiBlack)) + int(color.FgHiBlack)
 
@@ -104,6 +113,7 @@ func NewProcess(
 	proc.setUpProbes()
 	proc.procCond = *sync.NewCond(proc)
 	proc.procStartedCond = *sync.NewCond(proc)
+	proc.setUpWatchman(watchman)
 	return proc
 }
 
@@ -139,9 +149,9 @@ func (p *Process) run() int {
 
 		p.startProbes()
 
-		//Wait should wait for I/O consumption, but if the execution is too fast
-		//e.g. echo 'hello world' the output will not reach the pipe
-		//TODO Fix this
+		// Wait should wait for I/O consumption, but if the execution is too fast
+		// e.g. echo 'hello world' the output will not reach the pipe
+		// TODO Fix this
 		time.Sleep(50 * time.Millisecond)
 		_ = p.command.Wait()
 		p.Lock()
@@ -212,7 +222,6 @@ func (p *Process) getCommander() command.Commander {
 			p.mergeExtraArgs(),
 		)
 	}
-
 }
 
 func (p *Process) mergeExtraArgs() []string {
@@ -254,6 +263,11 @@ func (p *Process) isRestartable() bool {
 	p.Lock()
 	exitCode := p.getExitCode()
 	p.Unlock()
+
+	if p.isDevRestart.Swap(false) {
+		return true
+	}
+
 	if p.isStopped.Swap(false) {
 		return false
 	}
@@ -381,16 +395,35 @@ func (p *Process) isRunning() bool {
 
 func (p *Process) prepareForShutDown() {
 	// prevent restart during global shutdown or scale down
-	//p.procConf.RestartPolicy.Restart = types.RestartPolicyNo
+	// p.procConf.RestartPolicy.Restart = types.RestartPolicyNo
 	p.isStopped.Store(true)
-
 }
 
 func (p *Process) onProcessStart() {
 	if isStringDefined(p.procConf.LogLocation) {
 		p.logger.Open(p.getLogPath(), p.procConf.LoggerConfig)
 	}
 
+	for _, watch := range p.devWatchers {
+		go func() {
+			for {
+				files, ok := watch.sub.Recv()
+				if !ok {
+					log.
+						Debug().
+						Msg("watchman sub closed, exiting")
+					return
+				}
+
+				if len(files) == 0 {
+					continue
+				}
+
+				p.isDevRestart.Store(true)
+			}
+		}()
+	}
+
 	p.Lock()
 	p.started = true
 	p.Unlock()
@@ -457,8 +490,8 @@ func (p *Process) updateProcState() {
 		p.procState.Name = p.getName()
 	}
 	p.procState.IsRunning = isRunning
-
 }
+
 func (p *Process) setStartTime(startTime time.Time) {
 	p.timeMutex.Lock()
 	defer p.timeMutex.Unlock()
@@ -663,6 +696,13 @@ func (p *Process) onReadinessCheckEnd(isOk, isFatal bool, err string) {
 	}
 }
 
+func (p *Process) setUpWatchman(watchman *Watchman) {
+	for _, config := range p.procConf.Watch {
+		recv := watchman.Subscribe(config)
+		p.devWatchers = append(p.devWatchers, devWatch{recv, config})
+	}
+}
+
 func (p *Process) validateProcess() error {
 	if isStringDefined(p.procConf.WorkingDir) {
 		stat, err := os.Stat(p.procConf.WorkingDir)

src/app/project_runner.go

@@ -2,16 +2,17 @@ package app
 
 import (
 	"fmt"
-	"github.com/f1bonacc1/process-compose/src/config"
-	"github.com/f1bonacc1/process-compose/src/pclog"
-	"github.com/f1bonacc1/process-compose/src/types"
 	"os"
 	"os/user"
 	"runtime"
 	"slices"
 	"sync"
 	"time"
 
+	"github.com/f1bonacc1/process-compose/src/config"
+	"github.com/f1bonacc1/process-compose/src/pclog"
+	"github.com/f1bonacc1/process-compose/src/types"
+
 	"github.com/rs/zerolog/log"
 )
 
@@ -32,6 +33,7 @@ type ProjectRunner struct {
 	mainProcessArgs   []string
 	isTuiOn           bool
 	isOrderedShutDown bool
+	watchman          Watchman
 }
 
 func (p *ProjectRunner) GetLexicographicProcessNames() ([]string, error) {
@@ -67,7 +69,7 @@ func (p *ProjectRunner) Run() int {
 		p.logger.Open(p.project.LogLocation, p.project.LoggerConfig)
 		defer p.logger.Close()
 	}
-	//zerolog.SetGlobalLevel(zerolog.PanicLevel)
+	// zerolog.SetGlobalLevel(zerolog.PanicLevel)
 	log.Debug().Msgf("Spinning up %d processes. Order: %q", len(runOrder), nameOrder)
 	for _, proc := range runOrder {
 		newConf := proc
@@ -108,6 +110,7 @@ func (p *ProjectRunner) runProcess(config *types.ProcessConfig) {
 		printLogs,
 		isMain,
 		extraArgs,
+		&p.watchman,
 	)
 	p.addRunningProcess(process)
 	p.waitGroup.Add(1)
@@ -128,7 +131,6 @@ func (p *ProjectRunner) runProcess(config *types.ProcessConfig) {
 func (p *ProjectRunner) waitIfNeeded(process *types.ProcessConfig) error {
 	for k := range process.DependsOn {
 		if runningProc := p.getRunningProcess(k); runningProc != nil {
-
 			switch process.DependsOn[k].Condition {
 			case types.ProcessConditionCompleted:
 				runningProc.waitForCompletion()
@@ -158,7 +160,6 @@ func (p *ProjectRunner) waitIfNeeded(process *types.ProcessConfig) error {
 		} else {
 			log.Error().Msgf("Error: process %s depends on %s, but it isn't running", process.ReplicaName, k)
 		}
-
 	}
 	return nil
 }
@@ -596,6 +597,7 @@ func (p *ProjectRunner) renameProcess(name string, newName string) {
 		p.project.Processes[newName] = config
 	}
 }
+
 func (p *ProjectRunner) removeProcessLogs(name string) *pclog.ProcessLogBuffer {
 	p.logsMutex.Lock()
 	defer p.logsMutex.Unlock()
@@ -724,7 +726,6 @@ func bToMb(b uint64) uint64 {
 }
 
 func NewProjectRunner(opts *ProjectOpts) (*ProjectRunner, error) {
-
 	hostname, err := os.Hostname()
 	if err != nil {
 		log.Err(err).Msg("Failed get hostname")