-
Notifications
You must be signed in to change notification settings - Fork 249
feat(router): improved heartbeats for subscriptions #2141
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 14 commits
Commits
Show all changes
16 commits
Select commit
Hold shift + click to select a range
ca1e63f
chore: use wip go-tools 0f9f721b
endigma c53c95c
feat: heartbeat for SSE, heartbeat per writer
endigma 77c72fb
Use full write/flush methods for multipart
endigma 27b078d
chore: use wip go-tools 097167a6
endigma 1dce0b6
Rename MultipartSubHeartbeatInterval to SubHeartbeatInterval
endigma c812aad
chore: use wip go-tools 5a1ebf7a
endigma 3420a7c
chore: use wip go-tools 8c167f36
endigma a781bbe
chore: use wip go-tools fadc6bc8
endigma da99da4
Rename all mentions of multipart in heartbeat interval settings
endigma 5345b8b
test: new e2e for heartbeats
endigma 749a75d
fix typo
endigma 666e864
Fix typo, bump engine
endigma b6ee4ad
Make SSE complete events work properly with graphql-sse
endigma ed77e07
Add empty data line to test
endigma f99e42a
Use release graphql-go-tools
endigma 6108dcc
chore: don't block on channel close, could cause test timeouts
endigma File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,200 @@ | ||
| package integration | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "bytes" | ||
| "fmt" | ||
| "net/http" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "github.com/wundergraph/cosmo/router-tests/testenv" | ||
| "github.com/wundergraph/cosmo/router/core" | ||
| ) | ||
|
|
||
| func readMultipartPrefix(reader *bufio.Reader) error { | ||
| blankHeader, _, err := reader.ReadLine() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if len(blankHeader) != 0 { | ||
| return fmt.Errorf("expected blank header, got %q", blankHeader) | ||
| } | ||
|
|
||
| graphQLHeader, _, err := reader.ReadLine() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if string(graphQLHeader) != "--graphql" { | ||
| return fmt.Errorf("expected graphql header, got %q", graphQLHeader) | ||
| } | ||
|
|
||
| contentTypeHeader, _, err := reader.ReadLine() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if string(contentTypeHeader) != "Content-Type: application/json" { | ||
| return fmt.Errorf("expected content type header, got %q", contentTypeHeader) | ||
| } | ||
|
|
||
| blankFooter, _, err := reader.ReadLine() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if len(blankFooter) != 0 { | ||
| return fmt.Errorf("expected blank footer, got %q", blankFooter) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func TestHeartbeats(t *testing.T) { | ||
| subscriptionHeartbeatInterval := time.Millisecond * 300 | ||
|
|
||
| t.Run("should work correctly for multipart", func(t *testing.T) { | ||
| testenv.Run(t, &testenv.Config{ | ||
| RouterOptions: []core.Option{ | ||
| core.WithSubscriptionHeartbeatInterval(subscriptionHeartbeatInterval), | ||
| }, | ||
| }, func(t *testing.T, xEnv *testenv.Environment) { | ||
| client := http.Client{ | ||
| Timeout: time.Second * 100, | ||
| } | ||
|
|
||
| subscribePayload := []byte(`{"query":"subscription { countEmp(max: 5, intervalMilliseconds: 550) }"}`) | ||
|
|
||
| req := xEnv.MakeGraphQLMultipartRequest(http.MethodPost, bytes.NewReader(subscribePayload)) | ||
| resp, gErr := client.Do(req) | ||
| require.NoError(t, gErr) | ||
| require.Equal(t, http.StatusOK, resp.StatusCode) | ||
|
|
||
| defer resp.Body.Close() | ||
| reader := bufio.NewReader(resp.Body) | ||
|
|
||
| messages := make(chan string, 1) | ||
|
|
||
| go func() { | ||
| defer close(messages) | ||
| for { | ||
| err := readMultipartPrefix(reader) | ||
| if err != nil { | ||
| return | ||
| } | ||
|
|
||
| line, _, err := reader.ReadLine() | ||
| if err != nil { | ||
| return | ||
| } | ||
|
|
||
| fmt.Println(string(line)) | ||
| messages <- string(line) | ||
| } | ||
| }() | ||
|
|
||
| for i := 0; i <= 5; i++ { | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, messages, func(t *testing.T, msg string) { | ||
| assert.Equal(t, fmt.Sprintf(`{"payload":{"data":{"countEmp":%d}}}`, i), msg) | ||
| }) | ||
|
|
||
| testenv.AwaitChannelWithT(t, 5*time.Second, messages, func(t *testing.T, msg string) { | ||
| assert.Equal(t, `{}`, msg) | ||
| }) | ||
| } | ||
|
endigma marked this conversation as resolved.
|
||
|
|
||
| // Channel should be closed after all heartbeats are received | ||
| _, ok := <-messages | ||
| require.False(t, ok, "channel should be closed") | ||
| }) | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| }) | ||
|
|
||
| t.Run("should work correctly for sse", func(t *testing.T) { | ||
| testenv.Run(t, &testenv.Config{ | ||
| RouterOptions: []core.Option{ | ||
| core.WithSubscriptionHeartbeatInterval(subscriptionHeartbeatInterval), | ||
| }, | ||
| }, func(t *testing.T, xEnv *testenv.Environment) { | ||
| client := http.Client{ | ||
| Timeout: time.Second * 100, | ||
| } | ||
|
|
||
| subscribePayload := []byte(`{"query":"subscription { countEmp(max: 5, intervalMilliseconds: 550) }"}`) | ||
|
|
||
| req, err := http.NewRequest(http.MethodPost, xEnv.GraphQLRequestURL(), bytes.NewReader(subscribePayload)) | ||
| require.NoError(t, err) | ||
|
|
||
| req.Header.Set("Content-Type", "application/json") | ||
| req.Header.Set("Accept", "text/event-stream") | ||
| req.Header.Set("Connection", "keep-alive") | ||
| req.Header.Set("Cache-Control", "no-cache") | ||
|
|
||
| resp, err := client.Do(req) | ||
| require.NoError(t, err) | ||
| require.Equal(t, http.StatusOK, resp.StatusCode) | ||
|
|
||
| defer resp.Body.Close() | ||
| reader := bufio.NewReader(resp.Body) | ||
|
|
||
| lines := make(chan string, 50) | ||
|
|
||
| go func() { | ||
| defer close(lines) | ||
| for { | ||
| line, _, err := reader.ReadLine() | ||
| if err != nil { | ||
| return | ||
| } | ||
| lines <- string(line) | ||
| } | ||
| }() | ||
|
|
||
| // Assert the expected SSE sequence | ||
| for i := 0; i <= 5; i++ { | ||
| // Expect "event: next" | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, "event: next", line) | ||
| }) | ||
|
|
||
| // Expect data line with count | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, fmt.Sprintf(`data: {"data":{"countEmp":%d}}`, i), line) | ||
| }) | ||
|
|
||
| // Expect blank line | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, "", line) | ||
| }) | ||
|
|
||
| // Expect heartbeat | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, ":heartbeat", line) | ||
| }) | ||
|
|
||
| // Expect blank line after heartbeat | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, "", line) | ||
| }) | ||
| } | ||
|
|
||
| // Expect completion event | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, "event: complete", line) | ||
| }) | ||
|
|
||
| // Expect empty data line event | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, "data: ", line) | ||
| }) | ||
|
|
||
| // Expect blank line after complete | ||
| testenv.AwaitChannelWithT(t, 5*time.Second, lines, func(t *testing.T, line string) { | ||
| assert.Equal(t, "", line) | ||
| }) | ||
| }) | ||
| }) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.