diff --git a/picoev_server.v b/picoev_server.v index f655caf..e54c05a 100644 --- a/picoev_server.v +++ b/picoev_server.v @@ -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 { diff --git a/streaming.v b/streaming.v index a1eac14..3fd00ac 100644 --- a/streaming.v +++ b/streaming.v @@ -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 } }