diff --git a/.claude/skills/verify-wsstat/SKILL.md b/.claude/skills/verify-wsstat/SKILL.md new file mode 100644 index 0000000..5dc7355 --- /dev/null +++ b/.claude/skills/verify-wsstat/SKILL.md @@ -0,0 +1,61 @@ +--- +name: verify-wsstat +description: Build and drive the wsstat CLI end-to-end against the dev-stack mock server to verify changes at the CLI surface. +--- + +# Verifying wsstat changes + +wsstat is a CLI; its surface is the terminal. Verify by running the built +binary against the repo's own mock server and reading stdout/stderr + exit +codes. Do NOT write a throwaway WebSocket server — `dev/` already provides one +with per-feature endpoints (see the endpoint table in `dev/README.md`). + +## Build + server + +```bash +make build # -> ./bin/wsstat +./dev/run.sh up # Dockerized mock on ws://localhost:17080 and wss://localhost:17443 +``` + +No Docker? Run the mock on the host instead: + +```bash +cd dev/mock-server && PORT=17080 TLS_PORT=17443 go run . & +``` + +## Drive + +Pick the endpoint that isolates the changed feature (`/echo`, `/jsonrpc`, +`/stream?rate=N`, `/subscriptions` for stateful multi-frame conversations, +`/slow`, `/headers`, `/close-abrupt`, `/push`, …): + +```bash +./bin/wsstat measure -t ping ws://localhost:17080/echo +./bin/wsstat stream -c 2 -o json -t '{"method":"subscribe","subscription":{}}' ws://localhost:17080/subscriptions +``` + +## Make it stick + +A one-off invocation proves the change today; the suites keep proving it: + +- `dev/smoke-test.sh` (`make smoke`) — one assertion per feature. A new feature + gets one `check` line. +- `dev/soak-test.sh` (`make soak`) — the combination matrix. A new flag gets a + POSITIVE row per alias, a REJECT row per validation rule (a rule that exits 0 + is a silent accept), and an EFFECT check if its only failure mode is being + ignored. +- New server behavior needed? Add an endpoint to `dev/mock-server/main.go` and + document it in the `dev/README.md` table. + +Both suites also run standalone against a host-run mock: +`WSSTAT=./bin/wsstat ./dev/smoke-test.sh`. + +## Gotchas + +- Flags must come before the URL; the subcommand must come first. +- `-q`/`-v`/`--body`/`--clip` are rejected under `-o json|raw` (axis purity). +- Usage errors exit 2, runtime errors exit 1; under `-o json` runtime errors + emit a `{"type":"error"}` envelope on stdout. +- Use `ws://` (17080) to skip TLS; `wss://` (17443) is self-signed — verify via + `SSL_CERT_FILE=<(curl http://localhost:17080/ca.pem)` or `-k`. +- Bare hosts default to `wss://`. diff --git a/CHANGELOG.md b/CHANGELOG.md index e46d3e9..62350a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Repeatable `-t` in `stream` mode.** `wsstat stream` now accepts `-t/--text` multiple times and sends each message in argv order on the same connection, enabling multi-frame conversations (e.g. subscribe then unsubscribe) against a single server-side session. Sends are spaced by the new `--send-delay ` flag (default `1s`) so responses interleave predictably under `-o json`. The first message remains the subscribe payload; receiving (`-c`, `--once`, `--timeout`) is unchanged and a single `-t` behaves exactly as before; if the receive limit is reached before all messages are sent, the remaining sends are skipped. `measure` mode rejects repeated `-t`. + ## [3.1.1] - 2026-07-13 ### Fixed diff --git a/README.md b/README.md index 817f9e7..b8382cf 100644 --- a/README.md +++ b/README.md @@ -200,6 +200,20 @@ the first event: wsstat stream --once -t '{"method":"subscribe_ticker"}' wss://example.org/ws ``` +In `stream`, `-t` may be repeated to hold a multi-frame conversation on a single +connection: each message is sent in argv order, spaced by `--send-delay` +(default `1s`) so the server can answer each frame before the next arrives. The +first message is the subscribe payload; receiving is unchanged, so `-c`, +`--once`, and `--timeout` still bound the read side. If the receive limit is +reached before all messages are sent, the remaining sends are skipped: + +```sh +wsstat stream -c 4 -o json \ + -t '{"method":"subscribe","subscription":{"type":"trades","coin":"BTC"}}' \ + -t '{"method":"unsubscribe","subscription":{"type":"trades","coin":"BTC"}}' \ + wss://example.org/ws +``` + ### Output Output is split across three orthogonal axes: diff --git a/cmd/wsstat/cliux_test.go b/cmd/wsstat/cliux_test.go index 6915d33..4105133 100644 --- a/cmd/wsstat/cliux_test.go +++ b/cmd/wsstat/cliux_test.go @@ -16,37 +16,46 @@ func TestResolveTextPayload(t *testing.T) { t.Run("plain value passes through", func(t *testing.T) { t.Parallel() - c := &commonFlags{text: "ping"} + c := &commonFlags{text: textList{"ping"}} require.NoError(t, resolveTextPayload(c)) - assert.Equal(t, "ping", c.text) + assert.Equal(t, textList{"ping"}, c.text) }) - t.Run("empty value passes through", func(t *testing.T) { + t.Run("empty value is dropped", func(t *testing.T) { t.Parallel() - c := &commonFlags{text: ""} + c := &commonFlags{text: textList{""}} require.NoError(t, resolveTextPayload(c)) assert.Empty(t, c.text) }) t.Run("escaped leading at", func(t *testing.T) { t.Parallel() - c := &commonFlags{text: "@@literal"} + c := &commonFlags{text: textList{"@@literal"}} require.NoError(t, resolveTextPayload(c)) - assert.Equal(t, "@literal", c.text) + assert.Equal(t, textList{"@literal"}, c.text) }) t.Run("reads from file verbatim", func(t *testing.T) { t.Parallel() path := filepath.Join(t.TempDir(), "payload.json") require.NoError(t, os.WriteFile(path, []byte("{\"a\":1}\n"), 0o600)) - c := &commonFlags{text: "@" + path} + c := &commonFlags{text: textList{"@" + path}} require.NoError(t, resolveTextPayload(c)) - assert.Equal(t, "{\"a\":1}\n", c.text, "trailing newline is preserved") + assert.Equal(t, textList{"{\"a\":1}\n"}, c.text, "trailing newline is preserved") + }) + + t.Run("expands each entry independently", func(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "payload.json") + require.NoError(t, os.WriteFile(path, []byte("from file"), 0o600)) + c := &commonFlags{text: textList{"plain", "@" + path, ""}} + require.NoError(t, resolveTextPayload(c)) + assert.Equal(t, textList{"plain", "from file"}, c.text, "order kept, empties dropped") }) t.Run("missing file errors", func(t *testing.T) { t.Parallel() - c := &commonFlags{text: "@" + filepath.Join(t.TempDir(), "nope")} + c := &commonFlags{text: textList{"@" + filepath.Join(t.TempDir(), "nope")}} err := resolveTextPayload(c) require.Error(t, err) assert.Contains(t, err.Error(), "reading --text payload") @@ -64,9 +73,9 @@ func TestResolveTextPayload(t *testing.T) { os.Stdin = f defer func() { os.Stdin = prev }() - c := &commonFlags{text: "@-"} + c := &commonFlags{text: textList{"@-"}} require.NoError(t, resolveTextPayload(c)) - assert.Equal(t, "from stdin", c.text) + assert.Equal(t, textList{"from stdin"}, c.text) }) } diff --git a/cmd/wsstat/config.go b/cmd/wsstat/config.go index d21e137..da1b54c 100644 --- a/cmd/wsstat/config.go +++ b/cmd/wsstat/config.go @@ -35,7 +35,7 @@ type commonFlags struct { resolves resolveList rpcMethod string rpcVersion string - text string + text textList output string file string body string @@ -66,8 +66,8 @@ func registerCommon(fs *flag.FlagSet, c *commonFlags) { fs.StringVar(&c.rpcMethod, "rpc-method", "", "JSON-RPC method name to send (id=1, jsonrpc=2.0)") fs.StringVar(&c.rpcVersion, "rpc-version", "2.0", "JSON-RPC version for --rpc-method: 2.0 or 1.0") - fs.StringVar(&c.text, "t", "", "text message to send (@file or @- reads payload from a file or stdin)") - fs.StringVar(&c.text, "text", "", "text message to send (@file or @- reads payload from a file or stdin)") + fs.Var(&c.text, "t", "text message to send (repeatable in stream mode; @file or @- reads payload from a file or stdin)") + fs.Var(&c.text, "text", "text message to send (repeatable in stream mode; @file or @- reads payload from a file or stdin)") fs.StringVar(&c.output, "o", "text", "output contract: text, json, or raw") fs.StringVar(&c.output, "output", "text", "output contract: text, json, or raw") @@ -154,6 +154,10 @@ func resolveCommon(fs *flag.FlagSet, c *commonFlags, mode app.Mode) ([]app.Optio return nil, nil, err } + if err := validateMeasureMessaging(c, mode, output); err != nil { + return nil, nil, err + } + if c.timeout < 0 { return nil, nil, errors.New("--timeout must be zero or greater") } @@ -176,11 +180,6 @@ func resolveCommon(fs *flag.FlagSet, c *commonFlags, mode app.Mode) ([]app.Optio return nil, nil, err } - // Raw measure has no payload to emit without a message. - if output == app.OutputRaw && mode == app.ModeMeasure && c.text == "" && c.rpcMethod == "" { - return nil, nil, errors.New("-o raw in measure mode requires --text or --rpc-method") - } - target, err := positionalURL(fs) if err != nil { return nil, nil, err @@ -200,7 +199,7 @@ func buildCommonOptions( app.WithResolves(c.resolves.Values()), app.WithRPCMethod(c.rpcMethod), app.WithRPCVersion(rpcVersion), - app.WithTextMessage(c.text), + app.WithTextMessages(c.text.Values()), app.WithOutput(output), app.WithResponseFile(c.file), app.WithBodyRender(body), @@ -220,18 +219,34 @@ func buildCommonOptions( } } -// resolveTextPayload expands an @-prefixed --text value: "@-" reads stdin, "@path" reads a +// resolveTextPayload expands @-prefixed --text values: "@-" reads stdin, "@path" reads a // file. The bytes are sent verbatim (no trailing-newline stripping), matching the raw-output // contract. A literal leading @ is escaped as "@@". Any other value passes through unchanged. +// Entries that resolve to an empty string are dropped, so -t "" keeps meaning "no message". func resolveTextPayload(c *commonFlags) error { - if !strings.HasPrefix(c.text, "@") { - return nil + resolved := make(textList, 0, len(c.text)) + for _, entry := range c.text { + expanded, err := expandTextEntry(entry) + if err != nil { + return err + } + if expanded != "" { + resolved = append(resolved, expanded) + } } - if strings.HasPrefix(c.text, "@@") { - c.text = c.text[1:] - return nil + c.text = resolved + return nil +} + +// expandTextEntry expands a single --text value per the resolveTextPayload contract. +func expandTextEntry(entry string) (string, error) { + if !strings.HasPrefix(entry, "@") { + return entry, nil + } + if strings.HasPrefix(entry, "@@") { + return entry[1:], nil } - src := c.text[1:] + src := entry[1:] var ( data []byte err error @@ -242,16 +257,15 @@ func resolveTextPayload(c *commonFlags) error { data, err = os.ReadFile(src) } if err != nil { - return fmt.Errorf("reading --text payload: %w", err) + return "", fmt.Errorf("reading --text payload: %w", err) } - c.text = string(data) - return nil + return string(data), nil } // validateMessaging checks the messaging flags (--text / --rpc-method / --rpc-version) and // returns the resolved JSON-RPC version. func validateMessaging(c *commonFlags, set map[string]bool) (string, error) { - if c.text != "" && c.rpcMethod != "" { + if len(c.text) > 0 && c.rpcMethod != "" { return "", errors.New("mutually exclusive messaging flags: use --text or --rpc-method") } rpcVersion := strings.TrimSpace(c.rpcVersion) @@ -260,12 +274,28 @@ func validateMessaging(c *commonFlags, set map[string]bool) (string, error) { default: return "", errors.New("--rpc-version must be 1.0 or 2.0") } - if set["rpc-version"] && c.rpcMethod == "" && c.text == "" { + if set["rpc-version"] && c.rpcMethod == "" && len(c.text) == 0 { return "", errors.New("--rpc-version requires --rpc-method or --text") } return rpcVersion, nil } +// validateMeasureMessaging rejects measure-mode invocations whose messaging flags need stream +// mode or fall short of the output contract. Repeated -t drives a multi-frame conversation and +// only stream mode sends on an open connection; raw measure has no payload without a message. +func validateMeasureMessaging(c *commonFlags, mode app.Mode, output app.Output) error { + if mode != app.ModeMeasure { + return nil + } + if len(c.text) > 1 { + return errors.New("repeated -t requires the stream subcommand") + } + if output == app.OutputRaw && len(c.text) == 0 && c.rpcMethod == "" { + return errors.New("-o raw in measure mode requires --text or --rpc-method") + } + return nil +} + // splitCSV splits a comma-separated value into trimmed, non-empty parts. Returns nil for empty. func splitCSV(s string) []string { var out []string diff --git a/cmd/wsstat/config_test.go b/cmd/wsstat/config_test.go index 0416cdc..ede707e 100644 --- a/cmd/wsstat/config_test.go +++ b/cmd/wsstat/config_test.go @@ -2,6 +2,7 @@ package main import ( "testing" + "time" "github.com/jkbrsn/wsstat/v3/internal/app" "github.com/stretchr/testify/assert" @@ -72,6 +73,12 @@ func TestMeasureFlags(t *testing.T) { assert.Contains(t, err.Error(), "mutually exclusive") }) + t.Run("repeated text rejected", func(t *testing.T) { + _, _, err := buildMeasure([]string{"-t", "a", "-t", "b", "example.com"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "repeated -t requires the stream subcommand") + }) + t.Run("rpc-version accepts 1.0 with rpc-method", func(t *testing.T) { _, _, err := buildMeasure([]string{"--rpc-method", "m", "--rpc-version", "1.0", "example.com"}) require.NoError(t, err) @@ -231,6 +238,45 @@ func TestStreamFlags(t *testing.T) { _, _, err := buildStream([]string{"-b", "10", "example.com"}) require.NoError(t, err) }) + + t.Run("repeated text accepted in order", func(t *testing.T) { + client, _, err := buildStream([]string{"-t", "sub", "--text", "unsub", "example.com"}) + require.NoError(t, err) + assert.Equal(t, []string{"sub", "unsub"}, client.TextMessages()) + }) + + t.Run("send-delay defaults to one second", func(t *testing.T) { + client, _, err := buildStream([]string{"-t", "a", "-t", "b", "example.com"}) + require.NoError(t, err) + assert.Equal(t, time.Second, client.SendDelay()) + }) + + t.Run("send-delay parsed", func(t *testing.T) { + client, _, err := buildStream([]string{ + "--send-delay", "250ms", "-t", "a", "-t", "b", "example.com", + }) + require.NoError(t, err) + assert.Equal(t, 250*time.Millisecond, client.SendDelay()) + }) + + t.Run("negative send-delay rejected", func(t *testing.T) { + _, _, err := buildStream([]string{ + "--send-delay", "-1s", "-t", "a", "-t", "b", "example.com", + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "--send-delay must be zero or greater") + }) + + t.Run("send-delay without repeated text rejected", func(t *testing.T) { + for _, args := range [][]string{ + {"--send-delay", "1s", "example.com"}, + {"--send-delay", "1s", "-t", "a", "example.com"}, + } { + _, _, err := buildStream(args) + require.Errorf(t, err, "args %v should be rejected", args) + assert.Contains(t, err.Error(), "--send-delay has no effect without repeated -t") + } + }) } func TestParseReadLimit(t *testing.T) { diff --git a/cmd/wsstat/flags.go b/cmd/wsstat/flags.go index 39f339b..0dfe53f 100644 --- a/cmd/wsstat/flags.go +++ b/cmd/wsstat/flags.go @@ -35,6 +35,26 @@ func (h *headerList) Values() []string { return append([]string(nil), *h...) } +// textList is a flag.Value implementation that accumulates repeated -t / --text entries. +// Values are kept verbatim; @-expansion and empty filtering happen in resolveTextPayload. +type textList []string + +// Set appends the text message to the list. +func (l *textList) Set(value string) error { + *l = append(*l, value) + return nil +} + +// String returns the string representation of the flag value. +func (l *textList) String() string { + return strings.Join(*l, ", ") +} + +// Values returns the list of text messages. +func (l *textList) Values() []string { + return append([]string(nil), *l...) +} + // resolveList is a flag.Value implementation that accumulates DNS resolution overrides. // Each entry has the format "host:port:address" (e.g., "example.com:443:192.168.1.1"). // The flag can be repeated to override multiple host:port combinations. diff --git a/cmd/wsstat/flags_test.go b/cmd/wsstat/flags_test.go index 4b731d9..f9cf755 100644 --- a/cmd/wsstat/flags_test.go +++ b/cmd/wsstat/flags_test.go @@ -90,6 +90,43 @@ func TestHeaderList(t *testing.T) { }) } +func TestTextList(t *testing.T) { + t.Parallel() + + t.Run("Set accumulates values in order", func(t *testing.T) { + var l textList + require.NoError(t, l.Set("subscribe")) + require.NoError(t, l.Set("unsubscribe")) + assert.Equal(t, textList{"subscribe", "unsubscribe"}, l) + }) + + t.Run("Set keeps values verbatim", func(t *testing.T) { + var l textList + require.NoError(t, l.Set(" padded ")) + require.NoError(t, l.Set("")) + assert.Equal(t, textList{" padded ", ""}, l) + }) + + t.Run("String returns comma-separated list", func(t *testing.T) { + l := textList{"one", "two"} + assert.Equal(t, "one, two", l.String()) + }) + + t.Run("String returns empty for nil", func(t *testing.T) { + var l textList + assert.Equal(t, "", l.String()) + }) + + t.Run("Values returns copy", func(t *testing.T) { + l := textList{"one", "two"} + values := l.Values() + assert.Equal(t, []string{"one", "two"}, values) + + values[0] = "modified" + assert.Equal(t, "one", l[0]) + }) +} + func TestResolveList(t *testing.T) { t.Parallel() diff --git a/cmd/wsstat/main.go b/cmd/wsstat/main.go index f56c0a8..e5e1efc 100644 --- a/cmd/wsstat/main.go +++ b/cmd/wsstat/main.go @@ -41,6 +41,7 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/jkbrsn/wsstat/v3/internal/app" ) @@ -275,6 +276,8 @@ func buildStream(args []string) (*app.Client, *url.URL, error) { count := fs.Int("c", 0, "number of events to receive; 0 = unlimited") fs.IntVar(count, "count", 0, "number of events to receive; 0 = unlimited") once := fs.Bool("once", false, "exit after the first event") + sendDelay := fs.Duration("send-delay", time.Second, + "delay between successive -t sends (with repeated -t)") buffer := fs.Int("b", 0, "delivery buffer size (messages)") fs.IntVar(buffer, "buffer", 0, "delivery buffer size (messages)") summary := fs.Duration("summary-interval", 0, @@ -296,11 +299,11 @@ func buildStream(args []string) (*app.Client, *url.URL, error) { if err != nil { return nil, nil, err } + set := setFlagNames(fs) if *count < 0 { return nil, nil, errors.New("count must be zero or greater") } if *once { - set := setFlagNames(fs) if set["c"] || set["count"] { return nil, nil, errors.New("--count cannot be combined with --once") } @@ -310,11 +313,18 @@ func buildStream(args []string) (*app.Client, *url.URL, error) { return nil, nil, errors.New("--summary-interval has no effect with -o raw") } } + if *sendDelay < 0 { + return nil, nil, errors.New("--send-delay must be zero or greater") + } + if set["send-delay"] && len(cf.text) < 2 { + return nil, nil, errors.New("--send-delay has no effect without repeated -t") + } opts = append(opts, app.WithCount(*count), app.WithStreamOnce(*once), app.WithBuffer(*buffer), app.WithSummaryInterval(*summary), + app.WithSendDelay(*sendDelay), ) client := app.NewClient(opts...) diff --git a/cmd/wsstat/usage.go b/cmd/wsstat/usage.go index 2355721..6ddc62a 100644 --- a/cmd/wsstat/usage.go +++ b/cmd/wsstat/usage.go @@ -63,7 +63,7 @@ func printCommonFlags(w io.Writer) { fmt.Fprintln(w, "Input (choose one):") fmt.Fprintln(w, " --rpc-method JSON-RPC method name to send (id=1, jsonrpc=2.0)") fmt.Fprintln(w, " --rpc-version JSON-RPC version for --rpc-method: 2.0 or 1.0 [default: 2.0]") - fmt.Fprintln(w, " -t, --text text message to send (@file or @- reads file/stdin)") + fmt.Fprintln(w, " -t, --text text message to send (@file or @- reads file/stdin; repeatable in stream mode)") fmt.Fprintln(w) fmt.Fprintln(w, "Output:") fmt.Fprintln(w, " -o, --output output contract: text, json, raw [default: text]") @@ -128,9 +128,13 @@ func printStreamUsage(w io.Writer) { fmt.Fprintln(w, "Stream:") fmt.Fprintln(w, " -c, --count number of events to receive [default: 0 = unlimited]") fmt.Fprintln(w, " --once exit after the first event") + fmt.Fprintln(w, " --send-delay delay between successive -t sends [default: 1s]") fmt.Fprintln(w, " -b, --buffer delivery buffer size in messages [default: 32]") fmt.Fprintln(w, " --summary-interval ") fmt.Fprintln(w, " print stat summaries every interval (e.g., 5s, 1m) [default: disabled]") fmt.Fprintln(w) + fmt.Fprintln(w, " Note: -t may be repeated; each message is sent in order on the same connection, --send-delay apart.") + fmt.Fprintln(w, " If the receive limit (-c, --once) is reached first, remaining sends are skipped.") + fmt.Fprintln(w) printCommonFlags(w) } diff --git a/dev/README.md b/dev/README.md index 4170552..69c0f4c 100644 --- a/dev/README.md +++ b/dev/README.md @@ -43,6 +43,7 @@ tested in isolation. | `/echo` | Echoes each frame back unchanged. Serves `-t`, `-c`, `-o`/`--body`/`--clip`, `--resolve`, verbosity. | | `/jsonrpc` | Replies with a JSON-RPC result (or `-32700` on parse error). Serves `--rpc-method`. | | `/stream` | Waits for an initial frame, then pumps JSON notifications. `?rate=N` msgs/sec, `?count=N` cap. Serves the `stream` subcommand, `--once`, `--summary-interval`, `-b`. | +| `/subscriptions` | Stateful, Hyperliquid-style subscription manager with per-connection state: `subscribe` gets a `subscriptionResponse`, a duplicate `subscribe` gets `Already subscribed`, `unsubscribe` gets a `subscriptionResponse` (or `not subscribed`). Serves repeated `-t` / `--send-delay` multi-frame conversations. | | `/large` | Replies with a valid JSON-RPC frame whose result exceeds 32 KiB. | | `/slow` | Stalls 3s before replying. Serves the `-timeout` failure path. | | `/headers` | Replies with the value of the `X-Smoke` request header. Serves `-H`/`-header`. | diff --git a/dev/mock-server/main.go b/dev/mock-server/main.go index a8fb00f..606cb3a 100644 --- a/dev/mock-server/main.go +++ b/dev/mock-server/main.go @@ -46,6 +46,7 @@ func main() { mux.HandleFunc("/echo", handle(echoBehavior)) mux.HandleFunc("/jsonrpc", handle(jsonrpcBehavior)) mux.HandleFunc("/stream", handle(streamBehavior)) + mux.HandleFunc("/subscriptions", handle(subscriptionsBehavior)) mux.HandleFunc("/large", handle(largeBehavior)) mux.HandleFunc("/slow", handle(slowBehavior)) mux.HandleFunc("/headers", handle(headersBehavior)) @@ -351,6 +352,58 @@ func streamBehavior(r *http.Request, conn *websocket.Conn) { } } +type subscriptionRequest struct { + Method string `json:"method"` + Subscription json.RawMessage `json:"subscription"` +} + +// subscriptionsBehavior is a stateful, Hyperliquid-style subscription manager: +// each connection tracks its own subscription set, so behaviors only reachable +// on a *second* frame of the same connection can be exercised. subscribe replies +// with a subscriptionResponse, a duplicate subscribe with "Already subscribed", +// unsubscribe with a subscriptionResponse (or "not subscribed"). Serves repeated +// -t / --send-delay multi-frame conversations. +func subscriptionsBehavior(_ *http.Request, conn *websocket.Conn) { + defer conn.Close(websocket.StatusNormalClosure, "") + ctx := context.Background() + subs := map[string]bool{} + for { + _, data, err := conn.Read(ctx) + if err != nil { + return + } + var req subscriptionRequest + var reply []byte + key := "" + if err := json.Unmarshal(data, &req); err == nil { + key = string(req.Subscription) + } + switch { + case key == "": + reply = []byte(`{"channel":"error","data":"bad subscription request"}`) + case req.Method == "subscribe" && subs[key]: + reply = []byte(`{"channel":"error","data":"Already subscribed"}`) + case req.Method == "subscribe": + subs[key] = true + reply, _ = json.Marshal(map[string]any{ + "channel": "subscriptionResponse", + "data": map[string]any{"method": "subscribe", "subscription": req.Subscription}, + }) + case req.Method == "unsubscribe" && subs[key]: + delete(subs, key) + reply, _ = json.Marshal(map[string]any{ + "channel": "subscriptionResponse", + "data": map[string]any{"method": "unsubscribe", "subscription": req.Subscription}, + }) + default: + reply = []byte(`{"channel":"error","data":"not subscribed"}`) + } + if err := conn.Write(ctx, websocket.MessageText, reply); err != nil { + return + } + } +} + // rawOrNull returns the raw JSON value, or a JSON null when it is empty. func rawOrNull(raw json.RawMessage) json.RawMessage { if len(raw) == 0 { diff --git a/dev/smoke-test.sh b/dev/smoke-test.sh index 2226607..2e41a05 100755 --- a/dev/smoke-test.sh +++ b/dev/smoke-test.sh @@ -53,6 +53,16 @@ check_teardown_bound() { [[ "$ms" -lt 2500 ]] } +# check_multi_send drives a two-frame conversation on one connection: subscribe, +# then a duplicate subscribe --send-delay later. The second reply must be the +# server's "Already subscribed" rejection, which is only reachable when both +# frames land on the same server-side session. +check_multi_send() { + local sub='{"method":"subscribe","subscription":{"type":"trades","coin":"BTC"}}' + "$WSSTAT" stream -c 2 --send-delay 100ms -t "$sub" -t "$sub" "$WS_URL/subscriptions" 2>/dev/null \ + | grep -q "Already subscribed" +} + # check_stream_raw_clean asserts `stream -o raw` emits only verbatim payload # bytes: the concatenated JSON frames, with no "Streaming subscription events" # header and no "Subscription summary" block. Frames are undelimited, so a clean @@ -141,6 +151,7 @@ check "stream bounded" "$WSSTAT" stream -t sub -c 3 "$WS_URL/stream?rate=10" check "buffer size" "$WSSTAT" stream -b 8 -t sub -c 3 "$WS_URL/stream?rate=10" check "stream raw clean" check_stream_raw_clean check "summary-interval" check_summary_interval +check "repeated -t conversation" check_multi_send # --- Failure & edge paths --------------------------------------------------- check "timeout trips" bash -c "! $WSSTAT -timeout 1s -t hi $WS_URL/slow" diff --git a/dev/soak-test.sh b/dev/soak-test.sh index 868ce3b..dcf06ed 100755 --- a/dev/soak-test.sh +++ b/dev/soak-test.sh @@ -94,6 +94,21 @@ summary_interval_fires() { [[ "$n" -ge 2 ]] } json_measure_total() { "$B" -o json -t hi "$WS_URL/echo" 2>/dev/null | jq -es 'any(.[]; .durations_ms.total != null)' >/dev/null; } + +# Repeated -t conversation payloads against the stateful /subscriptions endpoint. +SUB_BTC='{"method":"subscribe","subscription":{"type":"trades","coin":"BTC"}}' +UNSUB_BTC='{"method":"unsubscribe","subscription":{"type":"trades","coin":"BTC"}}' +# multi_send_ordered asserts the full conversation semantics: three frames on one +# connection produce, in order, a subscribe ack, the duplicate rejection (only +# reachable on a second frame of the same session), and an unsubscribe ack. +# -o json gives one NDJSON line per reply, so order is asserted by line number. +multi_send_ordered() { + "$B" stream -c 3 -o json --send-delay 100ms \ + -t "$SUB_BTC" -t "$SUB_BTC" -t "$UNSUB_BTC" "$WS_URL/subscriptions" >"$OUTF" 2>/dev/null || return 1 + sed -n 1p "$OUTF" | grep -q '"method":"subscribe"' || return 1 + sed -n 2p "$OUTF" | grep -q "Already subscribed" || return 1 + sed -n 3p "$OUTF" | grep -q '"method":"unsubscribe"' +} json_stream_method() { "$B" stream -o json -t s -c 3 "$WS_URL/stream?rate=10" 2>/dev/null | jq -es 'all(.[]; .schema_version != null and .type != null) and any(.[]; .type == "subscription_message" and .payload.method == "subscription")' >/dev/null; } # _pty_maxwidth COLS -- CMD... -> widest output line in columns (ANSI/CR stripped) @@ -231,6 +246,9 @@ ok "stream --color always" -- "$B" stream --color always -c 3 -t s "$WS_ ok "stream -H + -c" -- "$B" stream -H "X-Smoke: 1" -c 3 -t s "$WS_URL/stream?rate=10" ok "stream --timeout + -c" -- "$B" stream --timeout 5s -c 3 -t s "$WS_URL/stream?rate=10" ok "stream once + -o json" -- "$B" stream --once -o json -t s "$WS_URL/stream?rate=10" +ok "stream repeated -t" -- "$B" stream -c 2 --send-delay 100ms -t "$SUB_BTC" -t "$UNSUB_BTC" "$WS_URL/subscriptions" +ok "stream repeated --text (alias)" -- "$B" stream -c 2 --send-delay 100ms --text "$SUB_BTC" --text "$UNSUB_BTC" "$WS_URL/subscriptions" +ok "stream repeated -t default delay" -- "$B" stream -c 2 -t "$SUB_BTC" -t "$UNSUB_BTC" "$WS_URL/subscriptions" # =========================================================================== section "REJECT: mutually exclusive / invalid-value rules" @@ -261,6 +279,10 @@ reject "stream -c negative" "zero or greater" -- "$B" stream -c reject "stream --once + -c" "cannot be combined" -- "$B" stream --once -c 3 -t s "$WS_URL/stream" reject "stream --once + --count" "cannot be combined" -- "$B" stream --once --count 3 -t s "$WS_URL/stream" reject "stream summary + -o raw" "no effect with -o raw" -- "$B" stream --summary-interval 1s -o raw -t s "$WS_URL/stream" +reject "stream --send-delay single -t" "no effect without repeated -t" -- "$B" stream --send-delay 1s -t s "$WS_URL/stream" +reject "stream --send-delay no -t" "no effect without repeated -t" -- "$B" stream --send-delay 1s "$WS_URL/stream" +reject "stream --send-delay negative" "zero or greater" -- "$B" stream --send-delay -1s -t a -t b "$WS_URL/stream" +reject "measure repeated -t" "stream subcommand" -- "$B" -t a -t b "$WS_URL/echo" section "REJECT: v2 flags removed in v3" reject "-subscribe" "removed in v3" -- "$B" -subscribe -t hi "$WS_URL/stream" @@ -276,6 +298,7 @@ reject "measure + --once" "not defined" -- "$B" --once -t hi "$WS_UR reject "measure + -b" "not defined" -- "$B" -b 8 -t hi "$WS_URL/echo" reject "measure + --buffer" "not defined" -- "$B" --buffer 8 -t hi "$WS_URL/echo" reject "measure + --summary-interval" "not defined" -- "$B" --summary-interval 1s -t hi "$WS_URL/echo" +reject "measure + --send-delay" "not defined" -- "$B" --send-delay 1s -t hi "$WS_URL/echo" # =========================================================================== section "EFFECT: output contracts produce the right bytes" @@ -299,6 +322,8 @@ section "EFFECT: stream output contracts" pred "stream -o raw is header/summary-free" stream_raw_clean # --summary-interval fires at least one periodic summary before the count ends. pred "stream --summary-interval fires periodically" summary_interval_fires +# Repeated -t: three staggered frames on one connection, replies in send order. +pred "stream repeated -t conversation is ordered" multi_send_ordered section "EFFECT: -o json is valid, schema-stable JSON" if [[ $HAVE_JQ -eq 1 ]]; then diff --git a/internal/app/client.go b/internal/app/client.go index 7897fd8..fe33d3a 100644 --- a/internal/app/client.go +++ b/internal/app/client.go @@ -62,7 +62,7 @@ type Client struct { resolves map[string]string // DNS resolution overrides: "host:port" → "address" rpcMethod string // JSON-RPC method (no params) rpcVersion string // JSON-RPC version to speak: "2.0" (default) or "1.0" - textMessage string // Text message + textMessages []string // Text messages; stream sends each in order, measure allows one subprotocols []string // WebSocket subprotocols to negotiate // Output @@ -88,6 +88,7 @@ type Client struct { once bool // stream: exit after the first event buffer int summaryInterval time.Duration + sendDelay time.Duration // stream: delay between successive text message sends // TLS configuration insecure bool // skip TLS certificate verification @@ -153,9 +154,21 @@ func WithRPCVersion(version string) Option { return func(c *Client) { c.rpcVersion = version } } -// WithTextMessage configures a text message to send. +// WithTextMessage configures a single text message to send. An empty message means none. func WithTextMessage(msg string) Option { - return func(c *Client) { c.textMessage = msg } + return func(c *Client) { + if msg == "" { + c.textMessages = nil + return + } + c.textMessages = []string{msg} + } +} + +// WithTextMessages configures the text messages to send. Stream mode sends each message in +// order on the same connection, spaced by the send delay; measure mode allows at most one. +func WithTextMessages(msgs []string) Option { + return func(c *Client) { c.textMessages = msgs } } // WithOutput sets the whole-stdout contract (text, json, or raw). @@ -222,6 +235,11 @@ func WithSummaryInterval(interval time.Duration) Option { return func(c *Client) { c.summaryInterval = interval } } +// WithSendDelay sets the stream-mode delay between successive text message sends. +func WithSendDelay(d time.Duration) Option { + return func(c *Client) { c.sendDelay = d } +} + // WithInsecure configures whether to skip TLS certificate verification. func WithInsecure(insecure bool) Option { return func(c *Client) { c.insecure = insecure } @@ -300,6 +318,12 @@ func (c *Client) Quiet() bool { return c.quiet } // RPCMethod returns the configured RPC method. func (c *Client) RPCMethod() string { return c.rpcMethod } +// TextMessages returns the configured text messages. +func (c *Client) TextMessages() []string { return c.textMessages } + +// SendDelay returns the stream-mode delay between successive text message sends. +func (c *Client) SendDelay() time.Duration { return c.sendDelay } + // wsstatOptions builds wsstat options based on client configuration. func (c *Client) wsstatOptions() []wsstat.Option { var opts []wsstat.Option @@ -355,7 +379,7 @@ func debugLogger(w io.Writer) zerolog.Logger { // based on client configuration. Returns timing results and the server response. // // The measurement method is determined by client settings: -// - If textMessage is set: sends text messages and measures echo latency +// - If a text message is set: sends text messages and measures echo latency // - If rpcMethod is set: sends JSON-RPC requests and measures response latency // - Otherwise: sends WebSocket ping frames and measures pong latency // @@ -371,7 +395,7 @@ func (c *Client) MeasureLatency( var result *MeasurementResult switch { - case c.textMessage != "": + case len(c.textMessages) > 0: result, err = c.measureText(ctx, target, header) case c.rpcMethod != "": result, err = c.measureJSON(ctx, target, header) @@ -403,7 +427,7 @@ func (c *Client) Validate() error { return errors.New("count must be zero or greater") } - if c.textMessage != "" && c.rpcMethod != "" { + if len(c.textMessages) > 0 && c.rpcMethod != "" { return errors.New("mutually exclusive messaging flags") } @@ -435,6 +459,10 @@ func (c *Client) Validate() error { return errors.New("summary-interval must be zero or greater") } + if c.sendDelay < 0 { + return errors.New("send-delay must be zero or greater") + } + if c.mode == ModeStream { if c.once && c.count > 1 { return errors.New("count must be 0 or 1 when --once is set") @@ -442,6 +470,10 @@ func (c *Client) Validate() error { return nil } + if len(c.textMessages) > 1 { + return errors.New("multiple text messages require stream mode") + } + if c.count == 0 { c.count = 1 } diff --git a/internal/app/client_subscription_test.go b/internal/app/client_subscription_test.go index cd6c342..1f9a413 100644 --- a/internal/app/client_subscription_test.go +++ b/internal/app/client_subscription_test.go @@ -99,7 +99,7 @@ func TestStreamSubscriptionRespectsCount(t *testing.T) { c := &Client{ count: 2, mode: ModeStream, - textMessage: "start", + textMessages: []string{"start"}, quiet: true, } require.NoError(t, c.Validate()) @@ -118,6 +118,48 @@ func TestStreamSubscriptionRespectsCount(t *testing.T) { require.NoError(t, <-errCh) } +func TestStreamSubscriptionSendsMessagesInOrder(t *testing.T) { + t.Parallel() + + server := newConversationTestServer(t) + defer server.cleanup() + + c := &Client{ + count: 3, + mode: ModeStream, + textMessages: []string{"subscribe", "resubscribe", "unsubscribe"}, + sendDelay: 10 * time.Millisecond, + quiet: true, + } + require.NoError(t, c.Validate()) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + errCh := make(chan error, 1) + go func() { + errCh <- c.StreamSubscription(ctx, server.wsURL) + }() + require.NoError(t, <-errCh, "stream should exit after receiving one ack per send") + + for _, want := range []string{"subscribe", "resubscribe", "unsubscribe"} { + select { + case got := <-server.received: + assert.Equal(t, want, got) + default: + t.Fatalf("server did not receive %q", want) + } + } +} + +func TestPendingSends(t *testing.T) { + t.Parallel() + + assert.Nil(t, (&Client{}).pendingSends()) + assert.Nil(t, (&Client{textMessages: []string{"only"}}).pendingSends()) + assert.Equal(t, []string{"b", "c"}, + (&Client{textMessages: []string{"a", "b", "c"}}).pendingSends()) +} + func TestStreamSubscriptionUnlimitedRequiresCancel(t *testing.T) { t.Parallel() @@ -126,7 +168,7 @@ func TestStreamSubscriptionUnlimitedRequiresCancel(t *testing.T) { c := &Client{ mode: ModeStream, - textMessage: "start", + textMessages: []string{"start"}, quiet: true, } require.NoError(t, c.Validate()) @@ -222,7 +264,7 @@ func TestStreamSubscriptionRawIsByteClean(t *testing.T) { count: 2, mode: ModeStream, output: OutputRaw, - textMessage: "start", + textMessages: []string{"start"}, } require.NoError(t, c.Validate()) @@ -367,7 +409,7 @@ func TestSubscriptionPayload(t *testing.T) { }) t.Run("text takes precedence over rpc", func(t *testing.T) { - client := &Client{textMessage: "text", rpcMethod: "method"} + client := &Client{textMessages: []string{"text"}, rpcMethod: "method"} msgType, payload, err := client.subscriptionPayload() require.NoError(t, err) assert.Equal(t, wsstat.TextMessage, msgType) diff --git a/internal/app/client_validation_test.go b/internal/app/client_validation_test.go index e8e009d..e0cfe17 100644 --- a/internal/app/client_validation_test.go +++ b/internal/app/client_validation_test.go @@ -15,7 +15,7 @@ func TestClientValidate(t *testing.T) { }) t.Run("mutually exclusive", func(t *testing.T) { - c := &Client{count: 1, textMessage: "hi", rpcMethod: "foo"} + c := &Client{count: 1, textMessages: []string{"hi"}, rpcMethod: "foo"} assert.Error(t, c.Validate()) }) @@ -26,7 +26,7 @@ func TestClientValidate(t *testing.T) { }) t.Run("valid", func(t *testing.T) { - c := &Client{count: 2, textMessage: "hi"} + c := &Client{count: 2, textMessages: []string{"hi"}} require.NoError(t, c.Validate()) }) @@ -88,6 +88,21 @@ func TestClientValidate(t *testing.T) { assert.Error(t, c.Validate()) }) + t.Run("negative send delay", func(t *testing.T) { + c := &Client{sendDelay: -1} + assert.Error(t, c.Validate()) + }) + + t.Run("multiple text messages require stream mode", func(t *testing.T) { + c := &Client{textMessages: []string{"a", "b"}} + assert.Error(t, c.Validate()) + }) + + t.Run("multiple text messages allowed in stream mode", func(t *testing.T) { + c := &Client{mode: ModeStream, textMessages: []string{"a", "b"}} + require.NoError(t, c.Validate()) + }) + t.Run("invalid color", func(t *testing.T) { c := &Client{colorMode: "purple"} assert.Error(t, c.Validate()) @@ -114,7 +129,7 @@ func TestValidateComplexCombinations(t *testing.T) { }, { name: "text and rpc method both set", - client: Client{textMessage: "hi", rpcMethod: "test"}, + client: Client{textMessages: []string{"hi"}, rpcMethod: "test"}, wantErr: true, errMsg: "mutually exclusive", }, diff --git a/internal/app/formatting.go b/internal/app/formatting.go index db2cb22..81c91a6 100644 --- a/internal/app/formatting.go +++ b/internal/app/formatting.go @@ -108,6 +108,13 @@ func tickerC(t *time.Ticker) <-chan time.Time { return t.C } +func timerC(t *time.Timer) <-chan time.Time { + if t == nil { + return nil + } + return t.C +} + func repeat[T any](value T, count int) []T { result := make([]T, count) for i := range result { diff --git a/internal/app/measurement.go b/internal/app/measurement.go index cbf8cc5..59c7898 100644 --- a/internal/app/measurement.go +++ b/internal/app/measurement.go @@ -17,7 +17,7 @@ func (c *Client) measureText( target *url.URL, header http.Header, ) (*MeasurementResult, error) { - msgs := repeat(c.textMessage, c.count) + msgs := repeat(c.textMessages[0], c.count) opts := append(c.wsstatOptions(), wsstat.WithHeaders(header)) result, rawResponses, err := wsstat.MeasureText(ctx, target, msgs, opts...) diff --git a/internal/app/sink_test.go b/internal/app/sink_test.go index fc471d2..135e533 100644 --- a/internal/app/sink_test.go +++ b/internal/app/sink_test.go @@ -88,7 +88,7 @@ func TestStreamSubscriptionRecordsToSink(t *testing.T) { c := &Client{ count: 2, mode: ModeStream, - textMessage: "start", + textMessages: []string{"start"}, quiet: true, } require.NoError(t, c.Validate()) diff --git a/internal/app/subscription.go b/internal/app/subscription.go index 76f56d0..9127b24 100644 --- a/internal/app/subscription.go +++ b/internal/app/subscription.go @@ -72,6 +72,9 @@ func (c *Client) runSubscriptionLoop( defer ticker.Stop() } + sender := newPendingSender(c.pendingSends(), c.sendDelay) + defer sender.stop() + messageIndex := 0 limit := c.count @@ -103,6 +106,8 @@ func (c *Client) runSubscriptionLoop( c.handleSubscriptionTick(wsClient, target) return nil } + case <-sender.c(): + sender.send(wsClient) case <-tickerC(ticker): c.handleSubscriptionTick(wsClient, target) if c.output == OutputText { @@ -112,6 +117,54 @@ func (c *Client) runSubscriptionLoop( } } +// pendingSends returns the text messages to send after the initial subscribe payload. +func (c *Client) pendingSends() []string { + if len(c.textMessages) < 2 { + return nil + } + return c.textMessages[1:] +} + +// pendingSender staggers the post-subscribe text messages. It is driven from the +// subscription loop goroutine, so ordering against the transport's write pump is +// guaranteed without extra coordination. With no pending messages its channel is +// nil and the loop's select never fires. +type pendingSender struct { + msgs []string + delay time.Duration + timer *time.Timer +} + +// newPendingSender creates a sender for msgs, arming the first send after delay. +func newPendingSender(msgs []string, delay time.Duration) *pendingSender { + s := &pendingSender{msgs: msgs, delay: delay} + if len(msgs) > 0 { + s.timer = time.NewTimer(delay) + } + return s +} + +// c returns the channel that fires when the next message is due. +func (s *pendingSender) c() <-chan time.Time { + return timerC(s.timer) +} + +// send writes the next message and re-arms the timer while messages remain. +func (s *pendingSender) send(ws *wsstat.WSStat) { + ws.WriteMessage(wsstat.TextMessage, []byte(s.msgs[0])) + s.msgs = s.msgs[1:] + if len(s.msgs) > 0 { + s.timer.Reset(s.delay) + } +} + +// stop releases the timer. +func (s *pendingSender) stop() { + if s.timer != nil { + s.timer.Stop() + } +} + // emitMessage records a received update to the response sink (no-op when --file is unset) // and prints it to stdout. Recording is independent of the stdout print, so it fires // regardless of -q/verbosity. @@ -122,10 +175,10 @@ func (c *Client) emitMessage(index int, msg wsstat.SubscriptionMessage) error { return c.printSubscriptionMessage(index, msg) } -// subscriptionPayload returns the payload to be sent to the server. +// subscriptionPayload returns the initial payload to be sent to the server. func (c *Client) subscriptionPayload() (int, []byte, error) { - if c.textMessage != "" { - return wsstat.TextMessage, []byte(c.textMessage), nil + if len(c.textMessages) > 0 { + return wsstat.TextMessage, []byte(c.textMessages[0]), nil } if c.rpcMethod != "" { req := buildRPCRequest(c.rpcMethod, c.rpcVersion) diff --git a/internal/app/testing_helpers_test.go b/internal/app/testing_helpers_test.go index 872341f..ab1019d 100644 --- a/internal/app/testing_helpers_test.go +++ b/internal/app/testing_helpers_test.go @@ -160,6 +160,54 @@ func newSubscriptionTestServer(t *testing.T) subscriptionTestServer { } } +// conversationTestServer is a test server that records every inbound frame and +// echoes each back prefixed with "ack:", for testing multi-message stream sends. +type conversationTestServer struct { + wsURL *url.URL + received <-chan string + cleanup func() +} + +// newConversationTestServer creates a new conversation test server. +func newConversationTestServer(t *testing.T) conversationTestServer { + t.Helper() + + received := make(chan string, 8) //revive:disable-line:add-constant test buffer size + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true}) + if err != nil { + return + } + defer func() { + _ = conn.CloseNow() + }() + conn.SetReadLimit(-1) + ctx := r.Context() + + for { + _, data, err := conn.Read(ctx) + if err != nil { + return + } + received <- string(data) + reply := append([]byte("ack:"), data...) + if err := conn.Write(ctx, websocket.MessageText, reply); err != nil { + return + } + } + })) + + wsURL, err := url.Parse("ws" + strings.TrimPrefix(server.URL, "http")) + require.NoError(t, err) + + return conversationTestServer{ + wsURL: wsURL, + received: received, + cleanup: server.Close, + } +} + func decodeJSONLine(t *testing.T, output string) map[string]any { t.Helper() trimmed := strings.TrimSpace(output)