1package brokeripc
2
3import (
4 "testing"
5 "time"
6
7 "github.com/stretchr/testify/require"
8
9 "github.com/bnema/vev/internal/adapters/brokerwire"
10 "github.com/bnema/vev/internal/ports"
11)
12
13// TestStreamCreditGrantDispatch proves how the session settles one client
14// credit grant: a grant past the window settles only its stream, a grant for a
15// retired stream is discarded, and a grant for a stream that was never opened
16// fails the whole connection.
17func TestStreamCreditGrantDispatch(t *testing.T) {
18 open := func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope, id ports.BrokerStreamID) {
19 t.Helper()
20 raw.send(t, brokerwire.OpenStream{
21 Epoch: scope.Epoch, Connection: scope.Connection, Stream: id,
22 Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(),
23 StartMode: ports.BrokerDaemonStartIfNeeded,
24 })
25 opened, ok := raw.recv(t).(brokerwire.StreamOpened)
26 require.True(t, ok, "stream %d must open", id)
27 require.Equal(t, id, opened.Stream)
28 }
29 grant := func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope, id ports.BrokerStreamID) {
30 t.Helper()
31 raw.send(t, brokerwire.StreamWindowUpdate{Epoch: scope.Epoch, Connection: scope.Connection, Stream: id, Credit: 1})
32 }
33 requireStreamFailed := func(t *testing.T, raw *rawCarriage, id ports.BrokerStreamID) {
34 t.Helper()
35 closed, ok := raw.recv(t).(brokerwire.StreamClosed)
36 require.True(t, ok, "an over-window grant must be answered with StreamClosed")
37 require.Equal(t, id, closed.Stream)
38 require.True(t, closed.HasError)
39 }
40
41 for _, tc := range []struct {
42 name string
43 run func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope)
44 }{
45 {"grant over the window settles only its stream", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
46 open(t, raw, scope, 1)
47 open(t, raw, scope, 2)
48 // A fresh stream already holds the whole window, so any grant overflows.
49 grant(t, raw, scope, 1)
50 requireStreamFailed(t, raw, 1)
51 open(t, raw, scope, 3) // the connection still serves new streams
52 }},
53 {"grant for a retired stream is discarded", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
54 open(t, raw, scope, 1)
55 grant(t, raw, scope, 1)
56 requireStreamFailed(t, raw, 1)
57 grant(t, raw, scope, 1) // stream 1 is retired now
58 open(t, raw, scope, 2)
59 }},
60 {"grant for a never-opened stream fails the connection", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
61 open(t, raw, scope, 1)
62 grant(t, raw, scope, 9)
63 require.Error(t, raw.awaitReadError(t, 5*time.Second))
64 }},
65 } {
66 t.Run(tc.name, func(t *testing.T) {
67 e := startEndpoint(t, Config{})
68 raw := rawDial(t, e, brokerwire.DefaultCeilings())
69 tc.run(t, raw, raw.register(t))
70 })
71 }
72}