Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f7e753969d | ||
|
|
e9d14a35f6 |
+41
-17
@@ -583,8 +583,8 @@ fn send_picoev_response(mut res picohttpparser.Response, ctx Context, response h
|
||||
|
||||
// handle_picoev_streaming_response - Handle a streaming response using PicoevStreamWriter
|
||||
// This function:
|
||||
// 1. Sends the HTTP headers for streaming
|
||||
// 2. Creates a PicoevStreamWriter
|
||||
// 1. Sends the HTTP headers for streaming directly to socket fd
|
||||
// 2. Creates a PicoevStreamWriter with the socket fd
|
||||
// 3. Executes the streaming callback
|
||||
// 4. Handles errors and cleanup
|
||||
fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context, config PicoevConfig) {
|
||||
@@ -601,14 +601,10 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
|
||||
// Determine status code
|
||||
status_code := if ctx.status_code != 0 { ctx.status_code } else { 200 }
|
||||
|
||||
// Send HTTP response line
|
||||
if status_code == 200 {
|
||||
res.http_ok()
|
||||
} else {
|
||||
res.raw('HTTP/1.1 ${status_code} ${get_status_text(status_code)}\r\n')
|
||||
}
|
||||
// Build HTTP headers string for direct socket write
|
||||
mut headers_str := 'HTTP/1.1 ${status_code} ${get_status_text(status_code)}\r\n'
|
||||
|
||||
// Send headers from context
|
||||
// Add headers from context
|
||||
mut has_transfer_encoding := false
|
||||
mut has_connection := false
|
||||
|
||||
@@ -618,7 +614,7 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
|
||||
if key_len == 14 && eq_ignore_case(key, 'content-length') {
|
||||
continue
|
||||
}
|
||||
res.header(key, value)
|
||||
headers_str += '${key}: ${value}\r\n'
|
||||
if key_len == 17 && eq_ignore_case(key, 'transfer-encoding') {
|
||||
has_transfer_encoding = true
|
||||
} else if key_len == 10 && eq_ignore_case(key, 'connection') {
|
||||
@@ -628,20 +624,27 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
|
||||
|
||||
// Ensure Transfer-Encoding: chunked is set
|
||||
if !has_transfer_encoding {
|
||||
res.header('Transfer-Encoding', 'chunked')
|
||||
headers_str += 'Transfer-Encoding: chunked\r\n'
|
||||
}
|
||||
|
||||
// Ensure Connection header is set for streaming
|
||||
if !has_connection {
|
||||
res.header('Connection', 'keep-alive')
|
||||
res.header('Keep-Alive', 'timeout=${config.timeout_secs}')
|
||||
headers_str += 'Connection: keep-alive\r\n'
|
||||
headers_str += 'Keep-Alive: timeout=${config.timeout_secs}\r\n'
|
||||
}
|
||||
|
||||
// End headers section (empty body to signal headers are complete)
|
||||
// Note: We don't call res.body() or res.end() here because we're streaming
|
||||
res.raw('\r\n')
|
||||
// End headers section
|
||||
headers_str += '\r\n'
|
||||
|
||||
// Create PicoevStreamWriter for streaming
|
||||
// Write headers directly to socket fd for immediate delivery
|
||||
fd := res.fd
|
||||
write_to_socket_fd(fd, headers_str.bytes()) or {
|
||||
eprintln('[Picoev Stream Error] Failed to write headers: ${err.msg()}')
|
||||
cleanup_stream_config(ctx)
|
||||
return
|
||||
}
|
||||
|
||||
// Create PicoevStreamWriter for streaming (uses fd directly)
|
||||
mut writer := PicoevStreamWriter.new(res)
|
||||
|
||||
// Create StreamContext with the writer
|
||||
@@ -665,6 +668,27 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
|
||||
cleanup_stream_config(ctx)
|
||||
}
|
||||
|
||||
// write_to_socket_fd - Helper function to write bytes directly to socket fd
|
||||
// Used for sending HTTP headers before streaming begins
|
||||
fn write_to_socket_fd(fd int, data []u8) ! {
|
||||
if data.len == 0 {
|
||||
return
|
||||
}
|
||||
mut total_written := 0
|
||||
for total_written < data.len {
|
||||
remaining := data.len - total_written
|
||||
ptr := unsafe { &u8(data.data) + total_written }
|
||||
written := C.send(fd, ptr, remaining, 0)
|
||||
if written < 0 {
|
||||
return error('Failed to write to socket')
|
||||
}
|
||||
if written == 0 {
|
||||
return error('Connection closed by peer')
|
||||
}
|
||||
total_written += written
|
||||
}
|
||||
}
|
||||
|
||||
// 获取状态码文本
|
||||
fn get_status_text(code int) string {
|
||||
return match code {
|
||||
|
||||
+45
-13
@@ -8,6 +8,9 @@ import time
|
||||
import picohttpparser
|
||||
import usockets
|
||||
|
||||
// C function declaration for direct socket writes
|
||||
fn C.send(sockfd int, buf voidptr, len int, flags int) int
|
||||
|
||||
// ============================================================================
|
||||
// SSE Event Structure
|
||||
// ============================================================================
|
||||
@@ -184,23 +187,26 @@ pub fn (mut ctx StreamContext) trigger_abort() {
|
||||
|
||||
// PicoevStreamWriter - StreamWriter implementation for picoev server
|
||||
// Supports chunked transfer encoding for streaming responses
|
||||
// Uses direct socket fd writes for immediate data delivery (true streaming)
|
||||
@[heap]
|
||||
pub struct PicoevStreamWriter {
|
||||
mut:
|
||||
res &picohttpparser.Response
|
||||
fd int // Socket file descriptor for direct writes
|
||||
connected bool
|
||||
}
|
||||
|
||||
// PicoevStreamWriter.new - Create a new PicoevStreamWriter with the given response
|
||||
// Extracts the fd from picohttpparser.Response for direct socket writes
|
||||
pub fn PicoevStreamWriter.new(res &picohttpparser.Response) &PicoevStreamWriter {
|
||||
return &PicoevStreamWriter{
|
||||
res: unsafe { res }
|
||||
fd: res.fd
|
||||
connected: true
|
||||
}
|
||||
}
|
||||
|
||||
// write - Write raw bytes to the stream using chunked transfer encoding
|
||||
// Format: {hex_size}\r\n{data}\r\n
|
||||
// Writes directly to socket fd for immediate delivery
|
||||
pub fn (mut w PicoevStreamWriter) write(data []u8) ! {
|
||||
if !w.connected {
|
||||
return error('Connection closed')
|
||||
@@ -210,9 +216,10 @@ pub fn (mut w PicoevStreamWriter) write(data []u8) ! {
|
||||
}
|
||||
// Write chunked format: size in hex, CRLF, data, CRLF
|
||||
chunk_header := format_chunk_size(data.len)
|
||||
w.res.raw(chunk_header)
|
||||
w.res.raw(data.bytestr())
|
||||
w.res.raw('\r\n')
|
||||
// Write directly to socket fd
|
||||
w.write_to_fd(chunk_header.bytes())!
|
||||
w.write_to_fd(data)!
|
||||
w.write_to_fd('\r\n'.bytes())!
|
||||
}
|
||||
|
||||
// write_string - Write a string to the stream using chunked transfer encoding
|
||||
@@ -225,15 +232,40 @@ pub fn (mut w PicoevStreamWriter) write_string(data string) ! {
|
||||
}
|
||||
// Write chunked format: size in hex, CRLF, data, CRLF
|
||||
chunk_header := format_chunk_size(data.len)
|
||||
w.res.raw(chunk_header)
|
||||
w.res.raw(data)
|
||||
w.res.raw('\r\n')
|
||||
// Write directly to socket fd
|
||||
w.write_to_fd(chunk_header.bytes())!
|
||||
w.write_to_fd(data.bytes())!
|
||||
w.write_to_fd('\r\n'.bytes())!
|
||||
}
|
||||
|
||||
// flush - Flush the buffer (picoev handles this automatically)
|
||||
// write_to_fd - Write bytes directly to the socket file descriptor
|
||||
// This ensures immediate delivery without buffering
|
||||
@[inline]
|
||||
fn (mut w PicoevStreamWriter) write_to_fd(data []u8) ! {
|
||||
if data.len == 0 {
|
||||
return
|
||||
}
|
||||
mut total_written := 0
|
||||
for total_written < data.len {
|
||||
remaining := data.len - total_written
|
||||
ptr := unsafe { &u8(data.data) + total_written }
|
||||
written := C.send(w.fd, ptr, remaining, 0)
|
||||
if written < 0 {
|
||||
w.connected = false
|
||||
return error('Failed to write to socket')
|
||||
}
|
||||
if written == 0 {
|
||||
w.connected = false
|
||||
return error('Connection closed by peer')
|
||||
}
|
||||
total_written += written
|
||||
}
|
||||
}
|
||||
|
||||
// flush - Flush the buffer (direct fd writes are already unbuffered)
|
||||
pub fn (mut w PicoevStreamWriter) flush() ! {
|
||||
// picoev automatically flushes data
|
||||
// No explicit flush needed
|
||||
// Direct fd writes are unbuffered, no explicit flush needed
|
||||
// But we can use TCP_NODELAY to disable Nagle's algorithm if needed
|
||||
}
|
||||
|
||||
// close - Close the stream by sending the final chunk
|
||||
@@ -241,9 +273,9 @@ pub fn (mut w PicoevStreamWriter) flush() ! {
|
||||
pub fn (mut w PicoevStreamWriter) close() ! {
|
||||
if w.connected {
|
||||
// Write the final chunk (zero-length chunk) to signal end of stream
|
||||
w.res.raw('0\r\n\r\n')
|
||||
w.res.end()
|
||||
w.write_to_fd('0\r\n\r\n'.bytes()) or {}
|
||||
w.connected = false
|
||||
// Note: Don't close the fd here, picoev manages the connection lifecycle
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+6
-6
@@ -7,8 +7,8 @@ $if windows {
|
||||
#flag -DLIBUS_USE_LIBUV
|
||||
#flag -DLIBUS_NO_SSL
|
||||
#flag -I@VMODROOT/usockets/include
|
||||
#flag -L@VMODROOT/lib
|
||||
#flag @VMODROOT/lib/libusockets_full.a
|
||||
#flag -L@VMODROOT/usockets/lib
|
||||
#flag @VMODROOT/usockets/lib/libusockets_full.a
|
||||
#flag -lws2_32 -liphlpapi -lpsapi -luserenv -lole32
|
||||
} $else $if macos {
|
||||
// macOS: 支持 Apple Silicon (M1/M2/M3) 和 Intel Mac
|
||||
@@ -26,16 +26,16 @@ $if windows {
|
||||
|
||||
// uSockets 库路径
|
||||
#flag -I@VMODROOT/usockets/include
|
||||
#flag -L@VMODROOT/lib
|
||||
#flag @VMODROOT/lib/libusockets_full.a
|
||||
#flag -L@VMODROOT/usockets/lib
|
||||
#flag @VMODROOT/usockets/lib/libusockets_full.a
|
||||
#flag -luv
|
||||
} $else {
|
||||
// Linux 和其他平台
|
||||
#flag -DLIBUS_USE_LIBUV
|
||||
#flag -DLIBUS_NO_SSL
|
||||
#flag -I@VMODROOT/usockets/include
|
||||
#flag -L@VMODROOT/lib
|
||||
#flag @VMODROOT/lib/libusockets_full.a
|
||||
#flag -L@VMODROOT/usockets/lib
|
||||
#flag @VMODROOT/usockets/lib/libusockets_full.a
|
||||
#flag -luv
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user