Initial commit

README.md

@@ -0,0 +1,20 @@
+# emt-zmq
+
+Interactive ZeroMQ terminal tool for debugging ZMQ PUSH and REP sockets.
+
+## Features
+
+- **Downlink (PUSH)** on port 6002 — send messages to connected PULL sockets
+- **Uplink (REP)** on port 6001 — receive requests and send replies
+
+## Usage
+
+```bash
+go run main.go
+```
+
+- Type a message and press Enter to send on the selected channel
+- Press Tab to cycle between push and rep channels
+- Press Ctrl+C to exit
+
+Messages are displayed in hex encoding.

debug_listen/main.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	zmq "github.com/go-zeromq/zmq4"
+)
+
+func main() {
+	ctx := context.Background()
+
+	pull := zmq.NewPull(ctx)
+	err := pull.Listen("tcp://0.0.0.0:6001")
+	if err != nil {
+		fmt.Printf("listen error: %v\n", err)
+		return
+	}
+
+	conn, err := net.DialTimeout("tcp", "127.0.0.1:6001", 2*time.Second)
+	if err != nil {
+		fmt.Printf("dial error: %v\n", err)
+	} else {
+		fmt.Println("port 6001 is open!")
+		conn.Close()
+	}
+
+	pull.Close()
+	time.Sleep(500 * time.Millisecond)
+}

go.mod

@@ -0,0 +1,13 @@
+module emt-zmq
+
+go 1.25.1
+
+require (
+	github.com/go-zeromq/goczmq/v4 v4.2.2 // indirect
+	github.com/go-zeromq/zmq4 v0.17.0 // indirect
+	github.com/pebbe/zmq4 v1.4.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
+	golang.org/x/sys v0.43.0 // indirect
+	golang.org/x/term v0.42.0 // indirect
+	golang.org/x/text v0.15.0 // indirect
+)

go.sum

@@ -0,0 +1,14 @@
+github.com/go-zeromq/goczmq/v4 v4.2.2 h1:HAJN+i+3NW55ijMJJhk7oWxHKXgAuSBkoFfvr8bYj4U=
+github.com/go-zeromq/goczmq/v4 v4.2.2/go.mod h1:Sm/lxrfxP/Oxqs0tnHD6WAhwkWrx+S+1MRrKzcxoaYE=
+github.com/go-zeromq/zmq4 v0.17.0 h1:r12/XdqPeRbuaF4C3QZJeWCt7a5vpJbslDH1rTXF+Kc=
+github.com/go-zeromq/zmq4 v0.17.0/go.mod h1:EQxjJD92qKnrsVMzAnx62giD6uJIPi1dMGZ781iCDtY=
+github.com/pebbe/zmq4 v1.4.0 h1:gO5P92Ayl8GXpPZdYcD62Cwbq0slSBVVQRIXwGSJ6eQ=
+github.com/pebbe/zmq4 v1.4.0/go.mod h1:nqnPueOapVhE2wItZ0uOErngczsJdLOGkebMxaO8r48=
+golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
+golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/term v0.42.0 h1:UiKe+zDFmJobeJ5ggPwOshJIVt6/Ft0rcfrXZDLWAWY=
+golang.org/x/term v0.42.0/go.mod h1:Dq/D+snpsbazcBG5+F9Q1n2rXV8Ma+71xEjTRufARgY=
+golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
+golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=

main.go

@@ -0,0 +1,173 @@
+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)
+		}
+	}
+}

test_client/main.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+	"time"
+
+	zmq "github.com/go-zeromq/zmq4"
+)
+
+func main() {
+	ctx := context.Background()
+
+	push := zmq.NewPush(ctx)
+	err := push.Dial("tcp://127.0.0.1:6001")
+	if err != nil {
+		fmt.Printf("push dial error: %v\n", err)
+		return
+	}
+	defer push.Close()
+
+	rep := zmq.NewRep(ctx)
+	err = rep.Dial("tcp://127.0.0.1:6002")
+	if err != nil {
+		fmt.Printf("rep dial error: %v\n", err)
+		return
+	}
+	defer rep.Close()
+
+	time.Sleep(500 * time.Millisecond)
+
+	err = push.Send(zmq.NewMsg([]byte("downlink message")))
+	if err != nil {
+		fmt.Printf("push send error: %v\n", err)
+		return
+	}
+	fmt.Printf("push sent: %s -> %s\n", "downlink message", hex.EncodeToString([]byte("downlink message")))
+
+	go func() {
+		for {
+			msg, err := rep.Recv()
+			if err != nil {
+				fmt.Printf("rep recv error: %v\n", err)
+				return
+			}
+			for _, frame := range msg.Frames {
+				fmt.Printf("rep received request: %s\n", hex.EncodeToString(frame))
+			}
+			rep.Send(zmq.NewMsg([]byte("reply from rep")))
+		}
+	}()
+
+	time.Sleep(1 * time.Second)
+}