From a43adfcf90d92e7d823a94224446360ed5f67bf3 Mon Sep 17 00:00:00 2001 From: LeonardoTrapani Date: Thu, 14 Aug 2025 11:30:36 +0200 Subject: [PATCH] improve daemon's toggle --- internal/daemon/daemon.go | 118 ++++++++++++++------------------ internal/pipeline/pipeline.go | 80 ++++++++++++++-------- internal/recording/recording.go | 18 ++--- 3 files changed, 109 insertions(+), 107 deletions(-) diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 92755ac..e72009c 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -10,7 +10,6 @@ import ( "os/signal" "sync" "syscall" - "time" "github.com/leonardotrapani/hyprvoice/internal/bus" "github.com/leonardotrapani/hyprvoice/internal/notify" @@ -34,9 +33,8 @@ type Daemon struct { ctx context.Context cancel context.CancelFunc - pipeline pipeline.Pipeline - pipelineCancel context.CancelFunc - statusCh <-chan Status + pipeline pipeline.Pipeline + actionChannel chan<- pipeline.Action } func New(n notify.Notifier) *Daemon { @@ -61,38 +59,35 @@ func (d *Daemon) Status() Status { } func (d *Daemon) startStatusReader(ctx context.Context, statusCh <-chan Status) { - go func() { - defer func() { - // Always clean up when this goroutine exits - d.mu.Lock() - d.status = Idle - d.statusCh = nil - d.pipeline = nil - d.pipelineCancel = nil - d.mu.Unlock() - }() - - for { - select { - case status, ok := <-statusCh: - if !ok { - return // Channel closed - } - - d.mu.Lock() - oldStatus := d.status - d.status = status - d.mu.Unlock() - - if oldStatus != status { - log.Printf("Status changed: %s -> %s", oldStatus, status) - } - - case <-ctx.Done(): - return // Context cancelled - } - } + defer func() { + // Always clean up when this goroutine exits + d.mu.Lock() + d.status = Idle + d.pipeline = nil + d.actionChannel = nil + d.mu.Unlock() }() + + for { + select { + case status, ok := <-statusCh: + if !ok { + return // Channel closed + } + + d.mu.Lock() + oldStatus := d.status + d.status = status + d.mu.Unlock() + + if oldStatus != status { + log.Printf("Status changed: %s -> %s", oldStatus, status) + } + + case <-ctx.Done(): + return // Context cancelled + } + } } func (d *Daemon) Run() error { @@ -139,7 +134,6 @@ func (d *Daemon) Run() error { log.Printf("Accept error: %v", err) return fmt.Errorf("accept failed: %w", err) } - go d.handle(c) } } @@ -165,7 +159,7 @@ func (d *Daemon) handle(c net.Conn) { fmt.Fprint(c, "OK toggled\n") case 's': status := d.Status() - fmt.Fprintf(c, "STATUS recording=%s\n", status) + fmt.Fprintf(c, "STATUS status=%s\n", status) case 'v': fmt.Fprintf(c, "STATUS proto=%s\n", bus.ProtoVer) case 'q': @@ -178,44 +172,36 @@ func (d *Daemon) handle(c net.Conn) { } func (d *Daemon) toggle() { - d.mu.Lock() - defer d.mu.Unlock() - - var notification func() - switch d.status { case Idle: - ctx, cancel := context.WithTimeout(d.ctx, 5*time.Minute) - p := pipeline.New() - d.pipeline = p - d.pipelineCancel = cancel - d.statusCh = p.Run(ctx) - notification = d.notifier.RecordingStarted + // Defensive: if a prior pipeline exists, stop and wait before starting a new one + if d.pipeline != nil { + d.pipeline.Stop() + d.pipeline = nil + } - // Start status reader for this pipeline - d.startStatusReader(ctx, d.statusCh) + p := pipeline.New() + statusCh, actionCh := p.Run(d.ctx) + + d.pipeline = p + d.actionChannel = actionCh + + go d.notifier.RecordingStarted() + + go d.startStatusReader(d.ctx, statusCh) case Recording: - if d.pipelineCancel != nil { - d.pipelineCancel() // Context cleanup handles the rest - } - notification = d.notifier.RecordingEnded + go d.notifier.RecordingEnded() + d.pipeline.Stop() case Transcribing: - if d.pipeline != nil { - d.pipeline.Inject() + select { + case d.actionChannel <- pipeline.Inject: + default: } - // No notification for injection start case Injecting: - if d.pipelineCancel != nil { - d.pipelineCancel() // Context cleanup handles the rest - } - notification = d.notifier.RecordingEnded - } - - // Send notification after releasing lock - if notification != nil { - go notification() + go d.notifier.RecordingEnded() + d.pipeline.Stop() // aborted during injection } } diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index e2516da..15c4713 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -3,6 +3,7 @@ package pipeline import ( "context" "log" + "sync" "time" ) @@ -15,38 +16,48 @@ const ( Injecting Status = "injecting" ) +// Action represents commands sent to the pipeline to drive state +// transitions or trigger work. Keeping this separate from Status +// preserves directionality: callers send Actions, pipeline emits Statuses. +type Action int + +const ( + Inject Action = iota +) + type Pipeline interface { - Run(ctx context.Context) <-chan Status - Inject() + Run(ctx context.Context) (<-chan Status, chan<- Action) + Stop() } type pipeline struct { - injectCh chan struct{} + actionCh chan Action + wg sync.WaitGroup + cancel context.CancelFunc } func New() Pipeline { return &pipeline{ - injectCh: make(chan struct{}, 1), + actionCh: make(chan Action, 1), } } -func (p *pipeline) Run(ctx context.Context) <-chan Status { +func (p *pipeline) Run(ctx context.Context) (<-chan Status, chan<- Action) { statusCh := make(chan Status, 1) - go p.run(ctx, statusCh) - return statusCh -} - -func (p *pipeline) Inject() { - select { - case p.injectCh <- struct{}{}: - default: - } + runCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) + p.cancel = cancel + p.wg.Add(1) + go p.run(runCtx, statusCh) + return statusCh, p.actionCh } func (p *pipeline) run(ctx context.Context, statusCh chan<- Status) { - defer close(statusCh) + defer func() { + // Ensure status channel is closed and wg is decremented + close(statusCh) + p.wg.Done() + }() - // Start recording log.Printf("Pipeline: Starting recording") statusCh <- Recording @@ -60,20 +71,26 @@ func (p *pipeline) run(ctx context.Context, statusCh chan<- Status) { return } - // Wait for injection or timeout + // TODO: when integrating the recorder, ensure its context is cancelled here on exit paths + // Wait for an action or timeout select { - case <-p.injectCh: - log.Printf("Pipeline: Injection started") - statusCh <- Injecting + case action := <-p.actionCh: + switch action { + case Inject: + log.Printf("Pipeline: Injection started") + statusCh <- Injecting - // Injection work - select { - case <-time.After(1 * time.Second): - log.Printf("Pipeline: Injection complete") - statusCh <- Idle // Instead of Completed - case <-ctx.Done(): - log.Printf("Pipeline: Stopped during injection") - return + // Injection work + select { + case <-time.After(1 * time.Second): + log.Printf("Pipeline: Injection complete") + statusCh <- Idle + case <-ctx.Done(): + log.Printf("Pipeline: Stopped during injection") + return + } + default: + // Unknown action: ignore for now } case <-time.After(10 * time.Second): // use context timeout @@ -85,3 +102,10 @@ func (p *pipeline) run(ctx context.Context, statusCh chan<- Status) { return } } + +func (p *pipeline) Stop() { + if p.cancel != nil { + p.cancel() + } + p.wg.Wait() +} diff --git a/internal/recording/recording.go b/internal/recording/recording.go index 5474461..1c311db 100644 --- a/internal/recording/recording.go +++ b/internal/recording/recording.go @@ -101,12 +101,8 @@ func (r *Recorder) Stop() error { if cancel != nil { cancel() } - - return nil -} - -func (r *Recorder) Wait() { r.wg.Wait() + return nil } func (r *Recorder) captureLoop(ctx context.Context, frameCh chan<- AudioFrame, errCh chan<- error) { @@ -144,16 +140,16 @@ func (r *Recorder) captureLoop(ctx context.Context, frameCh chan<- AudioFrame, e return } - r.mu.Lock() - r.cmd = cmd - r.mu.Unlock() - if err := cmd.Start(); err != nil { r.emitErr(errCh, fmt.Errorf("start pw-record: %w", err)) r.requestCancel() return } + r.mu.Lock() + r.cmd = cmd + r.mu.Unlock() + // Log stderr lines to aid diagnostics. go func() { scanner := bufio.NewScanner(stderr) @@ -221,7 +217,6 @@ func (r *Recorder) emitErr(errCh chan<- error, err error) { select { case errCh <- err: default: - // Best-effort; avoid blocking } log.Printf("Recording error: %v", err) } @@ -246,9 +241,6 @@ func CheckPipeWireAvailable(ctx context.Context) error { return fmt.Errorf("pw-record not found: %w (install pipewire-tools)", err) } // Use a short timeout to avoid hangs on misconfigured systems. - if ctx == nil { - ctx = context.Background() - } checkCtx, cancel := context.WithTimeout(ctx, 2*time.Second) defer cancel() cmd := exec.CommandContext(checkCtx, "pw-cli", "info")