🔥 Feature (v3): Add buffered streaming support #3131
Conversation
Create a new `*DefaultCtx` method called `SendStreamWriter()` that maps to fasthttp's `Response.SetBodyStreamWriter()`
- Adds Test_Ctx_SendStreamWriter to ctx_test.go
- Adds Test_Ctx_SendStreamWriter_Interrupted to ctx_test.go
- (Work-In-Progress) This test verifies that some data is
still sent before a client disconnects when using the method
`c.SendStreamWriter()`.
**Note:** Running this test reports a race condition when using
the `-race` flag or running `make test`. The test uses a channel
and mutex to prevent race conditions, but still triggers a warning.
grivera64
requested review from
gaby,
sixcolors,
ReneWerner87 and
efectn
and removed request for
a team
WalkthroughThe pull request introduces a new method, Changes
Assessment against linked issues
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
|
@grivera64 Try this: func Test_Ctx_SendStreamWriter_Interrupted(t *testing.T) {
t.Parallel()
app := New()
c := app.AcquireCtx(&fasthttp.RequestCtx{})
var mutex sync.Mutex
var wg sync.WaitGroup // WaitGroup to synchronize goroutines
startChan := make(chan bool)
interruptStreamWriter := func() {
defer wg.Done() // Notify WaitGroup when done
<-startChan
time.Sleep(5 * time.Millisecond)
mutex.Lock()
c.Response().CloseBodyStream() //nolint:errcheck // It is fine to ignore the error
mutex.Unlock()
}
wg.Add(1) // Increment WaitGroup counter before starting goroutine
err := c.SendStreamWriter(func(w *bufio.Writer) {
go interruptStreamWriter()
startChan <- true
for lineNum := 1; lineNum <= 5; lineNum++ {
mutex.Lock()
fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error
mutex.Unlock()
if err := w.Flush(); err != nil {
if lineNum < 3 {
t.Errorf("unexpected error: %s", err)
}
return
}
time.Sleep(1500 * time.Microsecond)
}
})
require.NoError(t, err)
wg.Wait() // Wait for the interruptStreamWriter to finish
// Protect access to the response body with the mutex
mutex.Lock()
defer mutex.Unlock()
require.Equal(t, "Line 1\nLine 2\nLine 3\n", string(c.Response().Body()))
} |
|
@gaby Thanks for the recommendation! Adding the Wait Group does remove the race error, but now I am getting an empty response body. I think this may be due to one of the following:
// go test -run Test_Ctx_SendStreamWriter_Interrupted
func Test_Ctx_SendStreamWriter_Interrupted(t *testing.T) {
t.Parallel()
app := New()
c := app.AcquireCtx(&fasthttp.RequestCtx{})
var mutex sync.Mutex
var wg sync.WaitGroup
startChan := make(chan bool)
interruptStreamWriter := func() {
wg.Add(1)
defer wg.Done()
<-startChan
time.Sleep(5 * time.Millisecond)
mutex.Lock()
c.Response().CloseBodyStream() //nolint:errcheck // It is fine to ignore the error
mutex.Unlock()
}
wg.Add(1)
err := c.SendStreamWriter(func(w *bufio.Writer) {
go interruptStreamWriter()
defer wg.Done()
startChan <- true
for lineNum := 1; lineNum <= 5; lineNum++ {
mutex.Lock()
fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error
mutex.Unlock()
if err := w.Flush(); err != nil {
if lineNum < 3 {
t.Errorf("unexpected error: %s", err)
}
return
}
time.Sleep(1500 * time.Microsecond)
}
})
require.NoError(t, err)
// Wait for StreamWriter and the goroutine to finish
wg.Wait()
mutex.Lock()
require.Equal(t, "Line 1\nLine 2\nLine 3\n", string(c.Response().Body()))
mutex.Unlock()
}I will file an issue on valyala/fasthttp to ask how to mock client disconnections for this test. If it's not possible, we could remove this test case, as most of the other tests do not test for client disconnection issues. |
|
Hey all, after reading a bit more on Fiber and Fasthttp documentation, I believe that the race condition is due to I tried using a modified version of app's This is the new race condition warning: WARNING: DATA RACE
Read at 0x00c00016c130 by goroutine 9:
github.com/gofiber/fiber/v3.(*testConn).Write()
/fiber/helpers.go:628 +0x6b
bufio.(*Writer).Flush()
/usr/local/go/src/bufio/bufio.go:639 +0xee
github.com/valyala/fasthttp.writeChunk()
/go/pkg/mod/github.com/valyala/[email protected]/http.go:2250 +0x10b
github.com/valyala/fasthttp.writeBodyChunked()
/go/pkg/mod/github.com/valyala/[email protected]/http.go:2170 +0xce
github.com/valyala/fasthttp.(*Response).writeBodyStream()
/go/pkg/mod/github.com/valyala/[email protected]/http.go:2066 +0x338
github.com/valyala/fasthttp.(*Response).Write()
/go/pkg/mod/github.com/valyala/[email protected]/http.go:1967 +0x2c4
github.com/valyala/fasthttp.writeResponse()
/go/pkg/mod/github.com/valyala/[email protected]/server.go:2589 +0xb8
github.com/valyala/fasthttp.(*Server).serveConn()
/go/pkg/mod/github.com/valyala/[email protected]/server.go:2432 +0x1ead
github.com/valyala/fasthttp.(*Server).ServeConn()
/go/pkg/mod/github.com/valyala/[email protected]/server.go:2042 +0x154
github.com/gofiber/fiber/v3.(*App).TestWithInterrupt.func1()
/fiber/app.go:975 +0xde
...This was the old warning when directly using github.com/valyala/fasthttp.(*Response).SetBodyStream()
/go/pkg/mod/github.com/valyala/[email protected]/http.go:249 +0x4f
github.com/valyala/fasthttp.(*Response).SetBodyStreamWriter()
/go/pkg/mod/github.com/valyala/[email protected]/http.go:292 +0x64
...I believe fixing this race warning in the cleanest way possible would require an upstream PR to fasthttp (most likely adding a mutex for Based on Fiber's current codebase, there doesn't seem to be other interrupt tests (while other tests ignore output when the response times out). If this isn't something we should be testing for, we could just remove the interrupt test and keep the remaining tests. What are your thoughts on this? Please let me know if you want to see the modified Edit: I will still try to work with the modified |
|
After editing // TestWithInterrupt is used for internal debugging by passing a *http.Request with an interruptAfter duration.
func (app *App) TestWithInterrupt(req *http.Request, interruptAfter time.Duration) (*http.Response, error)With this change, I was able to use that in the test case as written below as a fix: func Test_Ctx_SendStreamWriter_Interrupted_New(t *testing.T) {
t.Parallel()
app := New(Config{StreamRequestBody: true})
app.Get("/", func(c Ctx) error {
return c.SendStreamWriter(func(w *bufio.Writer) {
for lineNum := 1; lineNum <= 5; lineNum++ {
time.Sleep(time.Duration(lineNum) * time.Millisecond)
fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error
if err := w.Flush(); err != nil {
if lineNum <= 3 {
t.Errorf("unexpected error: %s", err)
}
return
}
}
})
})
resp, err := app.TestWithInterrupt(httptest.NewRequest(MethodGet, "/", nil), 8*time.Millisecond)
require.NoError(t, err, "app.TestWithInterrupt(req)")
body, err := io.ReadAll(resp.Body)
require.NotNil(t, err)
require.Equal(t, "Line 1\nLine 2\nLine 3\n", string(body))
}Would you all like me to write an issue/PR for adding |
|
Hello all, I hope you all have been doing well. I created an issue for the TestWithInterrupt() I proposed above. Please check it out here: |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Outside diff range and nitpick comments (3)
ctx_interface_gen.go (2)
286-287: LGTM: New SendStreamWriter methodThe
SendStreamWritermethod is a great addition for buffered streaming support. The signature is well-designed and consistent with other methods in the interface.Consider adding a brief comment above the method to describe its purpose and usage, similar to other methods in this interface. For example:
// SendStreamWriter sets the response body stream writer for buffered streamingThis would improve the documentation and make it easier for developers to understand the method's purpose at a glance.
Line range hint
1-387: Summary: Successful implementation of buffered streaming supportThe changes to
ctx_interface_gen.goeffectively introduce buffered streaming support to the Fiber v3 framework:
- The addition of the "bufio" import provides the necessary functionality for buffered I/O operations.
- The new
SendStreamWritermethod in theCtxinterface allows for setting a response body stream writer, enabling buffered streaming.These changes align well with the PR objectives and maintain consistency with the existing Fiber API. The implementation is non-breaking and focused, adding functionality without modifying existing methods.
To fully leverage this new feature:
- Ensure that the concrete implementation of the
Ctxinterface properly handles theSendStreamWritermethod.- Update the documentation to include examples of how to use this new method for Server-Side Events (SSE) and large file downloads.
- Consider adding unit tests that specifically target this new method to ensure its correct functionality and integration with the rest of the framework.
ctx.go (1)
1673-1679: LGTM! Consider enhancing error handling.The implementation of
SendStreamWriterlooks good and aligns well with the PR objectives. It correctly usesfasthttp.StreamWriterto set the body stream writer.Consider enhancing the error handling. Currently, the method always returns
nil. You might want to capture and return any potential errors that could occur during the stream writing process. This could be done by modifying thestreamWriterfunction signature to return an error and handling it accordingly.-func (c *DefaultCtx) SendStreamWriter(streamWriter func(*bufio.Writer)) error { - c.fasthttp.Response.SetBodyStreamWriter(fasthttp.StreamWriter(streamWriter)) - - return nil +func (c *DefaultCtx) SendStreamWriter(streamWriter func(*bufio.Writer) error) error { + return c.fasthttp.Response.SetBodyStreamWriter(fasthttp.StreamWriter(func(w *bufio.Writer) error { + return streamWriter(w) + })) }This change would allow for more robust error handling and propagation.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (4)
- ctx.go (2 hunks)
- ctx_interface_gen.go (2 hunks)
- ctx_test.go (2 hunks)
- docs/api/ctx.md (1 hunks)
🔇 Additional comments (2)
ctx_interface_gen.go (1)
6-6: LGTM: Import addition for bufio packageThe addition of the "bufio" import is correct and necessary for the new
SendStreamWritermethod. It's properly placed within the existing import block.docs/api/ctx.md (1)
1874-1914: Excellent addition ofSendStreamWritermethod documentation!The added documentation for the
SendStreamWritermethod is well-structured, clear, and informative. It provides:
- A concise description of the method's purpose.
- The method signature for easy reference.
- A basic usage example that demonstrates how to use the method.
- An advanced example showcasing manual flushing for more complex scenarios.
- Important notes about the flushing behavior, which is crucial for users to understand.
The documentation style and format are consistent with the rest of the file, making it easy for users to understand and integrate into their workflow. Great job on providing comprehensive and user-friendly documentation for this new method!
|
|
||
| err = c.SendStreamWriter(func(w *bufio.Writer) { | ||
| for lineNum := 1; lineNum <= 5; lineNum++ { | ||
| fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error |
There was a problem hiding this comment.
Check the error returned by fmt.Fprintf
To ensure that any write errors are handled, consider checking the error returned by fmt.Fprintf.
Apply this diff to check the error:
- fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error
+ if _, err := fmt.Fprintf(w, "Line %d\n", lineNum); err != nil {
+ t.Errorf("unexpected error: %s", err)
+ return
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error | |
| if _, err := fmt.Fprintf(w, "Line %d\n", lineNum); err != nil { | |
| t.Errorf("unexpected error: %s", err) | |
| return | |
| } |
| c := app.AcquireCtx(&fasthttp.RequestCtx{}) | ||
|
|
||
| err := c.SendStreamWriter(func(w *bufio.Writer) { | ||
| w.WriteString("Don't crash please") //nolint:errcheck, revive // It is fine to ignore the error |
There was a problem hiding this comment.
Check the error returned by w.WriteString
Even in test code, it's good practice to handle errors returned by write operations to ensure any unexpected issues are caught.
Apply this diff to check the error:
- w.WriteString("Don't crash please") //nolint:errcheck, revive // It is fine to ignore the error
+ if _, err := w.WriteString("Don't crash please"); err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| w.WriteString("Don't crash please") //nolint:errcheck, revive // It is fine to ignore the error | |
| if _, err := w.WriteString("Don't crash please"); err != nil { | |
| t.Errorf("unexpected error: %s", err) | |
| } |
| <-startChan | ||
| time.Sleep(5 * time.Millisecond) | ||
| mutex.Lock() | ||
| c.Response().CloseBodyStream() //nolint:errcheck // It is fine to ignore the error |
There was a problem hiding this comment.
Check the error returned by c.Response().CloseBodyStream()
Even when errors are expected, explicitly handling them improves code robustness and clarity.
Apply this diff to check the error:
- c.Response().CloseBodyStream() //nolint:errcheck // It is fine to ignore the error
+ if err := c.Response().CloseBodyStream(); err != nil {
+ t.Errorf("failed to close body stream: %s", err)
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| c.Response().CloseBodyStream() //nolint:errcheck // It is fine to ignore the error | |
| if err := c.Response().CloseBodyStream(); err != nil { | |
| t.Errorf("failed to close body stream: %s", err) | |
| } |
| startChan <- true | ||
| for lineNum := 1; lineNum <= 5; lineNum++ { | ||
| mutex.Lock() | ||
| fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error |
There was a problem hiding this comment.
Check the error returned by fmt.Fprintf
As with other write operations, it's advisable to handle the error returned by fmt.Fprintf to catch any potential issues.
Apply this diff to check the error:
- fmt.Fprintf(w, "Line %d\n", lineNum) //nolint:errcheck, revive // It is fine to ignore the error
+ if _, err := fmt.Fprintf(w, "Line %d\n", lineNum); err != nil {
+ if lineNum < 3 {
+ t.Errorf("unexpected error: %s", err)
+ }
+ return
+ }Committable suggestion was skipped due to low confidence.
Description
This feature adds buffered streaming support to Fiber v3 through the new Ctx method
SendStreamWriter:c.SendStreamWriter()essentially is a wrapper for calling fasthttp'sSetBodyStreamWritermethod. This feature wraps this method in the same way thatc.SendStream()wraps fasthttp'sSetBodyStream().With this feature, Fiber users can send shorter segments of content over persistent HTTP connections. This functionality is important for several web-based applications such as:
For example, a simple self-contained SSE example using this new feature can be setup as the following:
Fixes #3127
Type of change
Please delete options that are not relevant.
CURRENT STATUS
Features
Ctx.SendStreamWriter()to ctx.goUnit Tests
Add
Test_Ctx_SendStream_Writerto ctx_test.goAdd
Test_Ctx_SendStreamWriter_Interruptedto ctx_test.goDocumentation
Ctx.SendStreamWriter()docs to docs/api/ctx.mdBenchmarks
Checklist
Before you submit your pull request, please make sure you meet these requirements:
/docs/directory for Fiber's documentation.