(rocketmq-dashboard) branch rocketmq-studio updated: fix(rmqctl): drain buffered notifications before stdio exit (#5033)

[lizhimin]

This is an automated email from the ASF dual-hosted git repository.

lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git


The following commit(s) were added to refs/heads/rocketmq-studio by this push:
     new 6bbf674d5 fix(rmqctl): drain buffered notifications before stdio exit 
(#5033)
6bbf674d5 is described below

commit 6bbf674d540f558891ab9cb998570f320f413cfd
Author: Hanabi <317387557+hanabi9...@users.noreply.github.com>
AuthorDate: Sat Sep 26 18:11:27 2026 +0800

    fix(rmqctl): drain buffered notifications before stdio exit (#5033)
    
    test(rmqctl): cover queued notifications after input finishes
    
    fix(rmqctl): drain buffered notifications before stdio exit
---
 rmqctl/cmd/mcp_stdio.go            | 16 +++++++++++++++-
 rmqctl/cmd/mcp_stdio_proxy_test.go | 33 +++++++++++++++++++++++++++++++++
 2 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/rmqctl/cmd/mcp_stdio.go b/rmqctl/cmd/mcp_stdio.go
index 97f9de6c4..f5f96d5b4 100644
--- a/rmqctl/cmd/mcp_stdio.go
+++ b/rmqctl/cmd/mcp_stdio.go
@@ -162,7 +162,21 @@ func (p *stdioProxy) run(ctx context.Context) error {
                        p.handleResult(result)
                }
        }
-       return p.inputErr
+       // A call can queue notifications before its result, but select may 
consume
+       // the result first. Forward the remaining notifications before exiting.
+       for {
+               select {
+               case <-ctx.Done():
+                       return nil
+               case message, ok := <-p.session.Notifications():
+                       if !ok {
+                               return p.inputErr
+                       }
+                       p.handleNotification(message)
+               default:
+                       return p.inputErr
+               }
+       }
 }
 
 func (p *stdioProxy) handleNotification(message json.RawMessage) {
diff --git a/rmqctl/cmd/mcp_stdio_proxy_test.go 
b/rmqctl/cmd/mcp_stdio_proxy_test.go
index 24b5c6ee9..41f5cf73c 100644
--- a/rmqctl/cmd/mcp_stdio_proxy_test.go
+++ b/rmqctl/cmd/mcp_stdio_proxy_test.go
@@ -124,3 +124,36 @@ func TestStdioProxyForwardsServerNotification(t 
*testing.T) {
                t.Fatalf("stdout missing call response:\n%s", output)
        }
 }
+
+func TestStdioProxyDrainsNotificationsAfterInputAndCallsFinish(t *testing.T) {
+       readErr := errors.New("input read failed")
+       for _, tc := range []struct {
+               name string
+               err  error
+       }{
+               {name: "EOF"},
+               {name: "read error", err: readErr},
+       } {
+               t.Run(tc.name, func(t *testing.T) {
+                       session := newFakeMcpSession()
+                       defer session.Close()
+                       first := 
`{"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":50}}`
+                       last := 
`{"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":100}}`
+                       session.pushNotification(first)
+                       session.pushNotification(last)
+                       stdout := &bytes.Buffer{}
+                       proxy := newStdioProxy(strings.NewReader(""), stdout, 
&bytes.Buffer{}, session)
+                       defer close(proxy.done)
+                       // Model the select loop having consumed EOF and the 
last call result
+                       // before the notifications queued by that call.
+                       proxy.inputDone = true
+                       proxy.inputErr = tc.err
+                       if err := proxy.run(context.Background()); 
!errors.Is(err, tc.err) {
+                               t.Fatalf("run error = %v, want %v", err, tc.err)
+                       }
+                       if got, want := stdout.String(), first+"\n"+last+"\n"; 
got != want {
+                               t.Fatalf("stdout = %q, want %q", got, want)
+                       }
+               })
+       }
+}