emt-zmq/main.go

package main

import (
	"context"
	"encoding/hex"
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"golang.org/x/term"
	zmq "github.com/go-zeromq/zmq4"
)

func main() {
	pushPort := "6002"
	repPort := "6001"

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	push := zmq.NewPush(ctx)
	err := push.Listen("tcp://*:" + pushPort)
	if err != nil {
		fmt.Fprintf(os.Stderr, "error listening PUSH: %v\n", err)
		os.Exit(1)
	}
	defer push.Close()

	rep := zmq.NewRep(ctx)
	err = rep.Listen("tcp://*:" + repPort)
	if err != nil {
		fmt.Fprintf(os.Stderr, "error listening REP: %v\n", err)
		os.Exit(1)
	}
	defer rep.Close()

	fmt.Printf("Downlink on tcp://*:%s (PUSH)\n", pushPort)
	fmt.Printf("Uplink on tcp://*:%s (REP)\n", repPort)
	fmt.Println("Type a message and press Enter to send on the selected channel.")
	fmt.Println("Press Tab to cycle between channels. Ctrl+C to exit.")

	pendingReq := make(chan struct{}, 16)

	go func() {
		for {
			msg, err := rep.Recv()
			if err != nil {
				if ctx.Err() != nil {
					return
				}
				continue
			}
			for _, frame := range msg.Frames {
				fmt.Printf("\r[rep]  < %s\n", hex.EncodeToString(frame))
			}
			pendingReq <- struct{}{}
		}
	}()

	type sendReq struct {
		ch   string
		text string
	}
	sendCh := make(chan sendReq, 64)

	go func() {
		for req := range sendCh {
			switch req.ch {
			case "push":
				err := push.Send(zmq.NewMsg([]byte(req.text)))
				if err != nil {
					fmt.Fprintf(os.Stderr, "\r[push] send error: %v\n", err)
				} else {
					fmt.Printf("\r[push] > %s\n", hex.EncodeToString([]byte(req.text)))
				}
			case "rep":
				err := rep.Send(zmq.NewMsg([]byte(req.text)))
				if err != nil {
					fmt.Fprintf(os.Stderr, "\r[rep] send error: %v\n", err)
				} else {
					fmt.Printf("\r[rep] > %s\n", hex.EncodeToString([]byte(req.text)))
				}
			}
		}
	}()

	channels := []string{"push", "rep"}
	chIdx := 0

	oldState, err := term.MakeRaw(int(os.Stdin.Fd()))
	if err != nil {
		fmt.Fprintf(os.Stderr, "error setting raw terminal: %v\n", err)
		os.Exit(1)
	}
	defer term.Restore(int(os.Stdin.Fd()), oldState)

	fmt.Printf("[%s] > ", channels[chIdx])

	buf := make([]byte, 1)
	line := make([]byte, 0, 256)

	go func() {
		<-ctx.Done()
		term.Restore(int(os.Stdin.Fd()), oldState)
	}()

	for {
		n, err := os.Stdin.Read(buf)
		if err != nil || n == 0 {
			if ctx.Err() != nil {
				return
			}
			continue
		}

		b := buf[0]

		switch {
		case b == 0x09:
			chIdx = (chIdx + 1) % len(channels)
			fmt.Printf("\r[%s] > %s", channels[chIdx], string(line))

		case b == 0x0d || b == 0x0a:
			if len(line) == 0 {
				fmt.Printf("\r\n[%s] > ", channels[chIdx])
				continue
			}
			text := string(line)
			line = line[:0]

			ch := channels[chIdx]

			switch ch {
			case "push":
				sendCh <- sendReq{ch: "push", text: text}
			case "rep":
				select {
				case <-pendingReq:
					sendCh <- sendReq{ch: "rep", text: text}
				default:
					fmt.Printf("\r[rep] no pending request to reply to\n")
				}
			}

			select {
			case <-pendingReq:
				fmt.Printf("\r[rep] pending request waiting\n")
			default:
			}

			fmt.Printf("[%s] > ", channels[chIdx])

		case b == 0x7f || b == 0x08:
			if len(line) > 0 {
				line = line[:len(line)-1]
				fmt.Printf("\b \b")
			}

		case b == 0x03:
			fmt.Printf("\r\nExiting...\n")
			cancel()
			return

		case b >= 0x20:
			line = append(line, b)
			fmt.Printf("%c", b)
		}
	}
}