vpp-maglev/cmd/maglevc/watch.go

// Copyright (c) 2026, Pim van Pelt <pim@ipng.ch>

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"strconv"
	"syscall"

	"golang.org/x/sys/unix"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"

	"git.ipng.ch/ipng/vpp-maglev/internal/grpcapi"
)

func dynWatchEventOpts(_ context.Context, _ grpcapi.MaglevClient) []string {
	return []string{"num", "log", "backend", "frontend"}
}

// runWatchEvents implements 'watch events [num <n>] [log [level <level>]] [backend] [frontend]'.
// All tokens after 'events' are captured as args by the circular slot node in the tree.
// If none of log/backend/frontend are mentioned, all three default to true.
func runWatchEvents(ctx context.Context, client grpcapi.MaglevClient, args []string) error {
	var maxEvents int // 0 = unlimited
	var wantLog, wantBackend, wantFrontend bool
	logLevel := ""
	anyExplicit := false

	for i := 0; i < len(args); {
		switch args[i] {
		case "num":
			if i+1 >= len(args) {
				return fmt.Errorf("num requires a count argument")
			}
			n, err := strconv.Atoi(args[i+1])
			if err != nil || n < 1 {
				return fmt.Errorf("num: invalid count %q", args[i+1])
			}
			maxEvents = n
			i += 2
		case "log":
			wantLog = true
			anyExplicit = true
			if i+1 < len(args) && args[i+1] == "level" {
				if i+2 >= len(args) {
					return fmt.Errorf("log level requires a level argument")
				}
				logLevel = args[i+2]
				i += 3
			} else {
				i++
			}
		case "backend":
			wantBackend = true
			anyExplicit = true
			i++
		case "frontend":
			wantFrontend = true
			anyExplicit = true
			i++
		default:
			return fmt.Errorf("unknown watch option %q; expected: num, log, backend, frontend", args[i])
		}
	}

	if !anyExplicit {
		wantLog, wantBackend, wantFrontend = true, true, true
	}

	boolp := func(b bool) *bool { v := b; return &v }
	req := &grpcapi.WatchRequest{
		Log:      boolp(wantLog),
		LogLevel: logLevel,
		Backend:  boolp(wantBackend),
		Frontend: boolp(wantFrontend),
	}

	// Cancel the stream on keypress or signal.
	watchCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	go watchStopOnKeypress(watchCtx, cancel)

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	defer signal.Stop(sigCh)
	go func() {
		select {
		case <-sigCh:
			cancel()
		case <-watchCtx.Done():
		}
	}()

	stream, err := client.WatchEvents(watchCtx, req)
	if err != nil {
		return err
	}

	marshaler := protojson.MarshalOptions{}
	count := 0
	for {
		ev, err := stream.Recv()
		if err != nil {
			if watchCtx.Err() != nil {
				return nil // stopped by keypress or signal
			}
			if st, ok := status.FromError(err); ok && st.Code() == codes.Canceled {
				return nil
			}
			return err
		}
		data, err := marshaler.Marshal(ev)
		if err != nil {
			return fmt.Errorf("marshal event: %w", err)
		}
		fmt.Printf("%s\n", data)
		count++
		if maxEvents > 0 && count >= maxEvents {
			return nil
		}
	}
}

// watchStopOnKeypress puts stdin into cbreak mode (when it is a terminal) and
// calls cancel when any byte arrives. Cbreak mode disables canonical (line)
// input so a single keypress is sufficient, while preserving output
// post-processing (OPOST/ONLCR) so that fmt.Printf("\n") still produces the
// expected carriage-return+newline on screen. Falls back gracefully when stdin
// is not a tty. The goroutine exits when ctx is cancelled.
func watchStopOnKeypress(ctx context.Context, cancel context.CancelFunc) {
	fd := int(os.Stdin.Fd())
	if old, err := stdinCbreak(fd); err == nil {
		defer unix.IoctlSetTermios(fd, unix.TCSETSF, old) //nolint:errcheck
	}

	readDone := make(chan struct{})
	go func() {
		defer close(readDone)
		buf := make([]byte, 1)
		os.Stdin.Read(buf) //nolint:errcheck
	}()

	select {
	case <-readDone:
		cancel()
	case <-ctx.Done():
	}
}

// stdinCbreak sets the terminal referred to by fd into cbreak mode: canonical
// input and echo are disabled (so single keystrokes are immediately available)
// but output post-processing is left untouched (so \n still maps to \r\n).
// Returns the previous termios so the caller can restore it, or an error if fd
// is not a terminal.
func stdinCbreak(fd int) (*unix.Termios, error) {
	old, err := unix.IoctlGetTermios(fd, unix.TCGETS)
	if err != nil {
		return nil, err // not a terminal
	}
	t := *old
	t.Lflag &^= unix.ICANON | unix.ECHO | unix.ECHOE | unix.ECHOK | unix.ECHONL
	t.Cc[unix.VMIN] = 1
	t.Cc[unix.VTIME] = 0
	if err := unix.IoctlSetTermios(fd, unix.TCSETS, &t); err != nil {
		return nil, err
	}
	return old, nil
}