feat: add SSE streaming helper support
Add Server-Sent Events (SSE) and streaming response support with three streaming functions: - stream(): Basic binary streaming with chunked transfer encoding - stream_text(): Text streaming with proper Content-Type headers - stream_sse(): SSE event streaming with full event field support Features: - SSEEvent struct with data, event, id, retry fields - StreamContext for write, writeln, write_sse, sleep, pipe operations - Multi-line data handling in SSE events - Custom error handler support - Client disconnect detection via on_abort callback - Integrated with both Picoev and uSockets backends New files: - streaming.v: Core streaming implementation - examples/sse_demo.v: Comprehensive SSE demo with test page - tests/test_stream_*.v: 62 property-based tests Updated: - picoev_server.v: PicoevStreamWriter integration - usockets_server.v: UsocketsStreamWriter integration - README.md: SSE streaming documentation
This commit is contained in:
meiseayoung
+950
@@ -0,0 +1,950 @@
|
||||
// streaming.v - SSE Streaming Helper for v-hono
|
||||
// Provides Server-Sent Events (SSE) and streaming response support
|
||||
// Reference: Hono.js Streaming Helper API
|
||||
module hono
|
||||
|
||||
import net.http
|
||||
import time
|
||||
import picohttpparser
|
||||
import usockets
|
||||
|
||||
// ============================================================================
|
||||
// SSE Event Structure
|
||||
// ============================================================================
|
||||
|
||||
// SSEEvent - Server-Sent Events event structure
|
||||
// Used with stream_sse() to send formatted SSE events to clients
|
||||
pub struct SSEEvent {
|
||||
pub:
|
||||
data string // Event data (required) - supports multi-line
|
||||
event string // Event type name (optional)
|
||||
id string // Event ID for client reconnection (optional)
|
||||
retry int // Reconnection time in milliseconds (optional, 0 = not set)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// StreamWriter Interface
|
||||
// ============================================================================
|
||||
|
||||
// StreamWriter - Abstract interface for stream writing
|
||||
// Allows different server backends (picoev, uSockets) to implement their own writers
|
||||
pub interface StreamWriter {
|
||||
// Check if the connection is still active (immutable)
|
||||
is_connected() bool
|
||||
mut:
|
||||
// Write raw bytes to the stream
|
||||
write(data []u8) !
|
||||
// Write string to the stream
|
||||
write_string(data string) !
|
||||
// Flush the buffer to ensure data is sent
|
||||
flush() !
|
||||
// Close the stream connection
|
||||
close() !
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// StreamContext Structure
|
||||
// ============================================================================
|
||||
|
||||
// StreamContext - Context provided to streaming callbacks
|
||||
// Provides methods for writing data, SSE events, and managing the stream
|
||||
@[heap]
|
||||
pub struct StreamContext {
|
||||
mut:
|
||||
writer &StreamWriter
|
||||
is_closed bool
|
||||
abort_fn ?fn ()
|
||||
}
|
||||
|
||||
// StreamContext.new - Create a new StreamContext with the given writer
|
||||
pub fn StreamContext.new(writer &StreamWriter) StreamContext {
|
||||
return StreamContext{
|
||||
writer: unsafe { writer }
|
||||
is_closed: false
|
||||
abort_fn: none
|
||||
}
|
||||
}
|
||||
|
||||
// write - Write raw bytes to the stream
|
||||
pub fn (mut ctx StreamContext) write(data []u8) ! {
|
||||
if ctx.is_closed {
|
||||
return error('Stream is closed')
|
||||
}
|
||||
if !ctx.writer.is_connected() {
|
||||
ctx.is_closed = true
|
||||
return error('Connection lost')
|
||||
}
|
||||
ctx.writer.write(data)!
|
||||
}
|
||||
|
||||
// write_string - Write a string to the stream
|
||||
pub fn (mut ctx StreamContext) write_string(data string) ! {
|
||||
if ctx.is_closed {
|
||||
return error('Stream is closed')
|
||||
}
|
||||
if !ctx.writer.is_connected() {
|
||||
ctx.is_closed = true
|
||||
return error('Connection lost')
|
||||
}
|
||||
ctx.writer.write_string(data)!
|
||||
}
|
||||
|
||||
// writeln - Write a string followed by a newline character
|
||||
pub fn (mut ctx StreamContext) writeln(data string) ! {
|
||||
ctx.write_string(data)!
|
||||
ctx.write_string('\n')!
|
||||
}
|
||||
|
||||
|
||||
// write_sse - Write an SSE event to the stream
|
||||
// Formats the event according to the SSE specification:
|
||||
// - event: {type}\n (if event field is set)
|
||||
// - data: {line}\n (for each line of data)
|
||||
// - id: {id}\n (if id field is set)
|
||||
// - retry: {ms}\n (if retry > 0)
|
||||
// - \n (event separator)
|
||||
pub fn (mut ctx StreamContext) write_sse(event SSEEvent) ! {
|
||||
if ctx.is_closed {
|
||||
return error('Stream is closed')
|
||||
}
|
||||
if !ctx.writer.is_connected() {
|
||||
ctx.is_closed = true
|
||||
return error('Connection lost')
|
||||
}
|
||||
|
||||
// Write event type if specified
|
||||
if event.event.len > 0 {
|
||||
ctx.writer.write_string('event: ${event.event}\n')!
|
||||
}
|
||||
|
||||
// Write data field - handle multi-line data
|
||||
lines := event.data.split('\n')
|
||||
for line in lines {
|
||||
ctx.writer.write_string('data: ${line}\n')!
|
||||
}
|
||||
|
||||
// Write id if specified
|
||||
if event.id.len > 0 {
|
||||
ctx.writer.write_string('id: ${event.id}\n')!
|
||||
}
|
||||
|
||||
// Write retry if specified (must be > 0)
|
||||
if event.retry > 0 {
|
||||
ctx.writer.write_string('retry: ${event.retry}\n')!
|
||||
}
|
||||
|
||||
// Write event separator (empty line)
|
||||
ctx.writer.write_string('\n')!
|
||||
|
||||
// Flush to ensure event is sent immediately
|
||||
ctx.writer.flush()!
|
||||
}
|
||||
|
||||
// sleep - Pause execution for the specified milliseconds
|
||||
pub fn (ctx StreamContext) sleep(ms int) {
|
||||
time.sleep(ms * time.millisecond)
|
||||
}
|
||||
|
||||
// pipe - Pipe data from a byte array to the stream
|
||||
pub fn (mut ctx StreamContext) pipe(data []u8) ! {
|
||||
ctx.write(data)!
|
||||
}
|
||||
|
||||
// on_abort - Register a callback to be called when the client aborts the connection
|
||||
pub fn (mut ctx StreamContext) on_abort(callback fn ()) {
|
||||
ctx.abort_fn = callback
|
||||
}
|
||||
|
||||
// close - Close the stream
|
||||
pub fn (mut ctx StreamContext) close() {
|
||||
if !ctx.is_closed {
|
||||
ctx.is_closed = true
|
||||
ctx.writer.close() or {}
|
||||
}
|
||||
}
|
||||
|
||||
// is_open - Check if the stream is still open
|
||||
pub fn (ctx StreamContext) is_open() bool {
|
||||
return !ctx.is_closed && ctx.writer.is_connected()
|
||||
}
|
||||
|
||||
// trigger_abort - Internal method to trigger the abort callback
|
||||
// Called by server backends when client disconnects
|
||||
pub fn (mut ctx StreamContext) trigger_abort() {
|
||||
if abort_callback := ctx.abort_fn {
|
||||
abort_callback()
|
||||
}
|
||||
ctx.is_closed = true
|
||||
}
|
||||
|
||||
|
||||
// ============================================================================
|
||||
// Picoev StreamWriter Implementation
|
||||
// ============================================================================
|
||||
|
||||
// PicoevStreamWriter - StreamWriter implementation for picoev server
|
||||
// Supports chunked transfer encoding for streaming responses
|
||||
@[heap]
|
||||
pub struct PicoevStreamWriter {
|
||||
mut:
|
||||
res &picohttpparser.Response
|
||||
connected bool
|
||||
}
|
||||
|
||||
// PicoevStreamWriter.new - Create a new PicoevStreamWriter with the given response
|
||||
pub fn PicoevStreamWriter.new(res &picohttpparser.Response) &PicoevStreamWriter {
|
||||
return &PicoevStreamWriter{
|
||||
res: unsafe { res }
|
||||
connected: true
|
||||
}
|
||||
}
|
||||
|
||||
// write - Write raw bytes to the stream using chunked transfer encoding
|
||||
// Format: {hex_size}\r\n{data}\r\n
|
||||
pub fn (mut w PicoevStreamWriter) write(data []u8) ! {
|
||||
if !w.connected {
|
||||
return error('Connection closed')
|
||||
}
|
||||
if data.len == 0 {
|
||||
return
|
||||
}
|
||||
// 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_string - Write a string to the stream using chunked transfer encoding
|
||||
pub fn (mut w PicoevStreamWriter) write_string(data string) ! {
|
||||
if !w.connected {
|
||||
return error('Connection closed')
|
||||
}
|
||||
if data.len == 0 {
|
||||
return
|
||||
}
|
||||
// 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')
|
||||
}
|
||||
|
||||
// flush - Flush the buffer (picoev handles this automatically)
|
||||
pub fn (mut w PicoevStreamWriter) flush() ! {
|
||||
// picoev automatically flushes data
|
||||
// No explicit flush needed
|
||||
}
|
||||
|
||||
// close - Close the stream by sending the final chunk
|
||||
// Final chunk format: 0\r\n\r\n
|
||||
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.connected = false
|
||||
}
|
||||
}
|
||||
|
||||
// is_connected - Check if the connection is still active
|
||||
pub fn (w PicoevStreamWriter) is_connected() bool {
|
||||
return w.connected
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// uSockets StreamWriter Implementation
|
||||
// ============================================================================
|
||||
|
||||
// UsocketsStreamWriter - StreamWriter implementation for uSockets server
|
||||
// Supports chunked transfer encoding for streaming responses
|
||||
@[heap]
|
||||
pub struct UsocketsStreamWriter {
|
||||
mut:
|
||||
socket usockets.Socket
|
||||
connected bool
|
||||
}
|
||||
|
||||
// UsocketsStreamWriter.new - Create a new UsocketsStreamWriter with the given socket
|
||||
pub fn UsocketsStreamWriter.new(socket usockets.Socket) &UsocketsStreamWriter {
|
||||
return &UsocketsStreamWriter{
|
||||
socket: socket
|
||||
connected: true
|
||||
}
|
||||
}
|
||||
|
||||
// write - Write raw bytes to the stream using chunked transfer encoding
|
||||
// Format: {hex_size}\r\n{data}\r\n
|
||||
pub fn (mut w UsocketsStreamWriter) write(data []u8) ! {
|
||||
if !w.connected {
|
||||
return error('Connection closed')
|
||||
}
|
||||
if data.len == 0 {
|
||||
return
|
||||
}
|
||||
// Write chunked format: size in hex, CRLF, data, CRLF
|
||||
chunk_header := format_chunk_size(data.len)
|
||||
w.socket.write_bytes(chunk_header)
|
||||
w.socket.write_bytes(data.bytestr())
|
||||
w.socket.write_bytes('\r\n')
|
||||
}
|
||||
|
||||
// write_string - Write a string to the stream using chunked transfer encoding
|
||||
pub fn (mut w UsocketsStreamWriter) write_string(data string) ! {
|
||||
if !w.connected {
|
||||
return error('Connection closed')
|
||||
}
|
||||
if data.len == 0 {
|
||||
return
|
||||
}
|
||||
// Write chunked format: size in hex, CRLF, data, CRLF
|
||||
chunk_header := format_chunk_size(data.len)
|
||||
w.socket.write_bytes(chunk_header)
|
||||
w.socket.write_bytes(data)
|
||||
w.socket.write_bytes('\r\n')
|
||||
}
|
||||
|
||||
// flush - Flush the buffer (uSockets handles this automatically)
|
||||
pub fn (mut w UsocketsStreamWriter) flush() ! {
|
||||
// uSockets automatically flushes data
|
||||
// No explicit flush needed
|
||||
}
|
||||
|
||||
// close - Close the stream by sending the final chunk
|
||||
// Final chunk format: 0\r\n\r\n
|
||||
pub fn (mut w UsocketsStreamWriter) close() ! {
|
||||
if w.connected {
|
||||
// Write the final chunk (zero-length chunk) to signal end of stream
|
||||
w.socket.write_bytes('0\r\n\r\n')
|
||||
w.connected = false
|
||||
}
|
||||
}
|
||||
|
||||
// is_connected - Check if the connection is still active
|
||||
pub fn (w UsocketsStreamWriter) is_connected() bool {
|
||||
return w.connected
|
||||
}
|
||||
|
||||
// mark_disconnected - Mark the connection as disconnected
|
||||
// Called by server backends when client disconnects
|
||||
pub fn (mut w UsocketsStreamWriter) mark_disconnected() {
|
||||
w.connected = false
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Helper Functions for Chunked Encoding
|
||||
// ============================================================================
|
||||
|
||||
// format_chunk_size - Format the chunk size as hex followed by CRLF
|
||||
// This is used for HTTP chunked transfer encoding
|
||||
@[inline]
|
||||
fn format_chunk_size(size int) string {
|
||||
return '${size:x}\r\n'
|
||||
}
|
||||
|
||||
// format_chunk - Format a complete chunk with size, data, and trailing CRLF
|
||||
// Returns the full chunk in chunked transfer encoding format
|
||||
pub fn format_chunk(data []u8) string {
|
||||
if data.len == 0 {
|
||||
return ''
|
||||
}
|
||||
return '${data.len:x}\r\n${data.bytestr()}\r\n'
|
||||
}
|
||||
|
||||
// format_chunk_string - Format a string as a complete chunk
|
||||
pub fn format_chunk_string(data string) string {
|
||||
if data.len == 0 {
|
||||
return ''
|
||||
}
|
||||
return '${data.len:x}\r\n${data}\r\n'
|
||||
}
|
||||
|
||||
// format_final_chunk - Return the final chunk marker for chunked encoding
|
||||
pub fn format_final_chunk() string {
|
||||
return '0\r\n\r\n'
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Streaming Helper Functions
|
||||
// ============================================================================
|
||||
|
||||
// StreamCallback - Callback function type for streaming
|
||||
// The callback receives a mutable StreamContext and can write data to the stream
|
||||
pub type StreamCallback = fn (mut StreamContext) !
|
||||
|
||||
// StreamErrorHandler - Error handler function type for streaming
|
||||
// Called when an error occurs during streaming
|
||||
pub type StreamErrorHandler = fn (err IError, mut ctx StreamContext)
|
||||
|
||||
// StreamResponse - Response type for streaming operations
|
||||
// Contains the headers that should be sent before streaming begins
|
||||
pub struct StreamResponse {
|
||||
pub:
|
||||
status_code int = 200
|
||||
headers map[string]string
|
||||
is_stream bool = true
|
||||
}
|
||||
|
||||
// stream - Basic binary streaming function
|
||||
// Sets Transfer-Encoding: chunked and executes the callback with a StreamContext
|
||||
//
|
||||
// Parameters:
|
||||
// writer: The StreamWriter implementation (PicoevStreamWriter or UsocketsStreamWriter)
|
||||
// callback: The callback function that writes data to the stream
|
||||
// error_handler: Optional error handler called if an error occurs
|
||||
//
|
||||
// Returns:
|
||||
// StreamResponse with the appropriate headers
|
||||
//
|
||||
// Example:
|
||||
// stream(writer, fn (mut stream) ! {
|
||||
// stream.write('Hello'.bytes())!
|
||||
// stream.sleep(100)
|
||||
// stream.write('World'.bytes())!
|
||||
// })
|
||||
pub fn stream(mut writer StreamWriter, callback StreamCallback, error_handler ...StreamErrorHandler) StreamResponse {
|
||||
// Create StreamContext with the writer
|
||||
mut ctx := StreamContext.new(writer)
|
||||
|
||||
// Execute the callback and handle errors
|
||||
callback(mut ctx) or {
|
||||
// Handle error
|
||||
if error_handler.len > 0 {
|
||||
// Call custom error handler
|
||||
error_handler[0](err, mut ctx)
|
||||
} else {
|
||||
// Default: log error to console
|
||||
eprintln('[SSE Stream Error] ${err.msg()}')
|
||||
}
|
||||
}
|
||||
|
||||
// Auto-close the stream when callback completes
|
||||
ctx.close()
|
||||
|
||||
// Return response headers for streaming
|
||||
return StreamResponse{
|
||||
status_code: 200
|
||||
headers: {
|
||||
'Transfer-Encoding': 'chunked'
|
||||
'Connection': 'keep-alive'
|
||||
}
|
||||
is_stream: true
|
||||
}
|
||||
}
|
||||
|
||||
// stream_with_headers - Stream with custom initial headers
|
||||
// Allows setting additional headers before streaming begins
|
||||
pub fn stream_with_headers(mut writer StreamWriter, headers map[string]string, callback StreamCallback, error_handler ...StreamErrorHandler) StreamResponse {
|
||||
// Create StreamContext with the writer
|
||||
mut ctx := StreamContext.new(writer)
|
||||
|
||||
// Execute the callback and handle errors
|
||||
callback(mut ctx) or {
|
||||
// Handle error
|
||||
if error_handler.len > 0 {
|
||||
// Call custom error handler
|
||||
error_handler[0](err, mut ctx)
|
||||
} else {
|
||||
// Default: log error to console
|
||||
eprintln('[SSE Stream Error] ${err.msg()}')
|
||||
}
|
||||
}
|
||||
|
||||
// Auto-close the stream when callback completes
|
||||
ctx.close()
|
||||
|
||||
// Merge headers with required streaming headers
|
||||
mut response_headers := headers.clone()
|
||||
response_headers['Transfer-Encoding'] = 'chunked'
|
||||
if 'Connection' !in response_headers {
|
||||
response_headers['Connection'] = 'keep-alive'
|
||||
}
|
||||
|
||||
return StreamResponse{
|
||||
status_code: 200
|
||||
headers: response_headers
|
||||
is_stream: true
|
||||
}
|
||||
}
|
||||
|
||||
// get_stream_headers - Get the headers required for basic streaming
|
||||
// Returns a map of headers that should be set for chunked transfer encoding
|
||||
pub fn get_stream_headers() map[string]string {
|
||||
return {
|
||||
'Transfer-Encoding': 'chunked'
|
||||
'Connection': 'keep-alive'
|
||||
}
|
||||
}
|
||||
|
||||
// write_stream_headers - Write HTTP headers for streaming response
|
||||
// This is a helper function to write the initial HTTP response line and headers
|
||||
// before starting chunked data transfer
|
||||
//
|
||||
// Parameters:
|
||||
// writer: The StreamWriter to write headers to
|
||||
// status_code: HTTP status code (default 200)
|
||||
// headers: Map of header name to value
|
||||
pub fn write_stream_headers(mut writer StreamWriter, status_code int, headers map[string]string) ! {
|
||||
// Build HTTP response line
|
||||
status_text := get_status_text(status_code)
|
||||
mut response := 'HTTP/1.1 ${status_code} ${status_text}\r\n'
|
||||
|
||||
// Add headers
|
||||
for key, value in headers {
|
||||
response += '${key}: ${value}\r\n'
|
||||
}
|
||||
|
||||
// Add required streaming headers if not present
|
||||
if 'Transfer-Encoding' !in headers {
|
||||
response += 'Transfer-Encoding: chunked\r\n'
|
||||
}
|
||||
if 'Connection' !in headers {
|
||||
response += 'Connection: keep-alive\r\n'
|
||||
}
|
||||
|
||||
// End headers section
|
||||
response += '\r\n'
|
||||
|
||||
// Write headers directly (not chunked)
|
||||
// Note: Headers are written raw, not using chunked encoding
|
||||
// The chunked encoding starts after the headers
|
||||
writer.write_string(response)!
|
||||
}
|
||||
|
||||
// stream_text - Text streaming function
|
||||
// Sets Content-Type: text/plain; charset=utf-8, Transfer-Encoding: chunked,
|
||||
// and X-Content-Type-Options: nosniff headers, then executes the callback
|
||||
// with a StreamContext for text streaming.
|
||||
//
|
||||
// Parameters:
|
||||
// writer: The StreamWriter implementation (PicoevStreamWriter or UsocketsStreamWriter)
|
||||
// callback: The callback function that writes text data to the stream
|
||||
// error_handler: Optional error handler called if an error occurs
|
||||
//
|
||||
// Returns:
|
||||
// StreamResponse with the appropriate headers for text streaming
|
||||
//
|
||||
// Example:
|
||||
// stream_text(writer, fn (mut stream) ! {
|
||||
// stream.write_string('Hello ')!
|
||||
// stream.sleep(100)
|
||||
// stream.writeln('World')!
|
||||
// })
|
||||
pub fn stream_text(mut writer StreamWriter, callback StreamCallback, error_handler ...StreamErrorHandler) StreamResponse {
|
||||
// Create StreamContext with the writer
|
||||
mut ctx := StreamContext.new(writer)
|
||||
|
||||
// Execute the callback and handle errors
|
||||
callback(mut ctx) or {
|
||||
// Handle error
|
||||
if error_handler.len > 0 {
|
||||
// Call custom error handler
|
||||
error_handler[0](err, mut ctx)
|
||||
} else {
|
||||
// Default: log error to console
|
||||
eprintln('[SSE StreamText Error] ${err.msg()}')
|
||||
}
|
||||
}
|
||||
|
||||
// Auto-close the stream when callback completes
|
||||
ctx.close()
|
||||
|
||||
// Return response headers for text streaming
|
||||
return StreamResponse{
|
||||
status_code: 200
|
||||
headers: {
|
||||
'Content-Type': 'text/plain; charset=utf-8'
|
||||
'Transfer-Encoding': 'chunked'
|
||||
'X-Content-Type-Options': 'nosniff'
|
||||
'Connection': 'keep-alive'
|
||||
}
|
||||
is_stream: true
|
||||
}
|
||||
}
|
||||
|
||||
// get_stream_text_headers - Get the headers required for text streaming
|
||||
// Returns a map of headers that should be set for text streaming with chunked transfer encoding
|
||||
pub fn get_stream_text_headers() map[string]string {
|
||||
return {
|
||||
'Content-Type': 'text/plain; charset=utf-8'
|
||||
'Transfer-Encoding': 'chunked'
|
||||
'X-Content-Type-Options': 'nosniff'
|
||||
'Connection': 'keep-alive'
|
||||
}
|
||||
}
|
||||
|
||||
// stream_sse - Server-Sent Events streaming function
|
||||
// Sets Content-Type: text/event-stream, Cache-Control: no-cache,
|
||||
// and Connection: keep-alive headers, then executes the callback
|
||||
// with a StreamContext for SSE streaming.
|
||||
//
|
||||
// Parameters:
|
||||
// writer: The StreamWriter implementation (PicoevStreamWriter or UsocketsStreamWriter)
|
||||
// callback: The callback function that writes SSE events to the stream
|
||||
// error_handler: Optional error handler called if an error occurs
|
||||
//
|
||||
// Returns:
|
||||
// StreamResponse with the appropriate headers for SSE streaming
|
||||
//
|
||||
// Example:
|
||||
// stream_sse(writer, fn (mut stream) ! {
|
||||
// stream.write_sse(SSEEvent{
|
||||
// data: 'Hello World'
|
||||
// event: 'message'
|
||||
// id: '1'
|
||||
// })!
|
||||
// stream.sleep(1000)
|
||||
// stream.write_sse(SSEEvent{
|
||||
// data: 'Another message'
|
||||
// event: 'update'
|
||||
// id: '2'
|
||||
// })!
|
||||
// })
|
||||
pub fn stream_sse(mut writer StreamWriter, callback StreamCallback, error_handler ...StreamErrorHandler) StreamResponse {
|
||||
// Create StreamContext with the writer
|
||||
mut ctx := StreamContext.new(writer)
|
||||
|
||||
// Execute the callback and handle errors
|
||||
callback(mut ctx) or {
|
||||
// Handle error
|
||||
if error_handler.len > 0 {
|
||||
// Call custom error handler
|
||||
error_handler[0](err, mut ctx)
|
||||
} else {
|
||||
// Default: log error to console
|
||||
eprintln('[SSE StreamSSE Error] ${err.msg()}')
|
||||
}
|
||||
}
|
||||
|
||||
// Auto-close the stream when callback completes
|
||||
ctx.close()
|
||||
|
||||
// Return response headers for SSE streaming
|
||||
// Requirements 3.1, 3.2, 3.3:
|
||||
// - Content-Type: text/event-stream
|
||||
// - Cache-Control: no-cache
|
||||
// - Connection: keep-alive
|
||||
return StreamResponse{
|
||||
status_code: 200
|
||||
headers: {
|
||||
'Content-Type': 'text/event-stream'
|
||||
'Cache-Control': 'no-cache'
|
||||
'Connection': 'keep-alive'
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
is_stream: true
|
||||
}
|
||||
}
|
||||
|
||||
// get_stream_sse_headers - Get the headers required for SSE streaming
|
||||
// Returns a map of headers that should be set for Server-Sent Events streaming
|
||||
pub fn get_stream_sse_headers() map[string]string {
|
||||
return {
|
||||
'Content-Type': 'text/event-stream'
|
||||
'Cache-Control': 'no-cache'
|
||||
'Connection': 'keep-alive'
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// ============================================================================
|
||||
// Context-Based Streaming Functions
|
||||
// ============================================================================
|
||||
// These functions are designed to be used in route handlers and integrate
|
||||
// with the picoev and uSockets server backends.
|
||||
|
||||
// StreamType - Type of streaming response
|
||||
pub enum StreamType {
|
||||
basic // Basic binary streaming
|
||||
text // Text streaming with text/plain content type
|
||||
sse // Server-Sent Events streaming
|
||||
}
|
||||
|
||||
// StreamConfig - Configuration stored in Context for streaming responses
|
||||
// This is used internally by the server backends to detect and handle streaming
|
||||
pub struct StreamConfig {
|
||||
pub:
|
||||
stream_type StreamType
|
||||
callback StreamCallback = unsafe { nil }
|
||||
error_handler ?StreamErrorHandler
|
||||
}
|
||||
|
||||
// Global storage for stream callbacks (workaround for V's limitations with storing functions in maps)
|
||||
// Key is a unique identifier stored in Context.store['_stream_id']
|
||||
// Using shared map for thread-safe access
|
||||
struct StreamCallbackStorage {
|
||||
mut:
|
||||
callbacks shared map[string]StreamConfig
|
||||
}
|
||||
|
||||
// Module-level storage instance
|
||||
const stream_storage = &StreamCallbackStorage{}
|
||||
|
||||
// store_stream_config - Store a stream configuration
|
||||
fn store_stream_config(id string, config StreamConfig) {
|
||||
lock stream_storage.callbacks {
|
||||
stream_storage.callbacks[id] = config
|
||||
}
|
||||
}
|
||||
|
||||
// retrieve_stream_config - Retrieve a stream configuration by ID
|
||||
fn retrieve_stream_config(id string) ?StreamConfig {
|
||||
rlock stream_storage.callbacks {
|
||||
if id in stream_storage.callbacks {
|
||||
return stream_storage.callbacks[id]
|
||||
}
|
||||
}
|
||||
return none
|
||||
}
|
||||
|
||||
// remove_stream_config - Remove a stream configuration by ID
|
||||
fn remove_stream_config(id string) {
|
||||
lock stream_storage.callbacks {
|
||||
stream_storage.callbacks.delete(id)
|
||||
}
|
||||
}
|
||||
|
||||
// generate_stream_id - Generate a unique stream ID
|
||||
fn generate_stream_id() string {
|
||||
return '${time.now().unix_milli()}_${rand_int()}'
|
||||
}
|
||||
|
||||
// rand_int - Simple random integer generator
|
||||
fn rand_int() int {
|
||||
return int(time.now().unix_nano() % 1000000)
|
||||
}
|
||||
|
||||
// c_stream - Basic binary streaming function for use in route handlers
|
||||
// Sets Transfer-Encoding: chunked and stores the callback for execution by the server backend
|
||||
//
|
||||
// Parameters:
|
||||
// c: The request Context
|
||||
// callback: The callback function that writes data to the stream
|
||||
// error_handler: Optional error handler called if an error occurs
|
||||
//
|
||||
// Returns:
|
||||
// http.Response with streaming marker headers
|
||||
//
|
||||
// Example:
|
||||
// app.get('/stream', fn (mut c hono.Context) http.Response {
|
||||
// return hono.c_stream(mut c, fn (mut stream hono.StreamContext) ! {
|
||||
// stream.write('Hello'.bytes())!
|
||||
// stream.sleep(100)
|
||||
// stream.write('World'.bytes())!
|
||||
// })
|
||||
// })
|
||||
pub fn c_stream(mut c Context, callback StreamCallback, error_handler ...StreamErrorHandler) http.Response {
|
||||
// Generate unique stream ID
|
||||
stream_id := generate_stream_id()
|
||||
|
||||
// Store stream configuration
|
||||
config := StreamConfig{
|
||||
stream_type: .basic
|
||||
callback: callback
|
||||
error_handler: if error_handler.len > 0 { error_handler[0] } else { none }
|
||||
}
|
||||
store_stream_config(stream_id, config)
|
||||
|
||||
// Mark context as streaming
|
||||
c.store['_stream'] = 'true'
|
||||
c.store['_stream_id'] = stream_id
|
||||
c.store['_stream_type'] = 'basic'
|
||||
|
||||
// Set streaming headers
|
||||
c.headers['Transfer-Encoding'] = 'chunked'
|
||||
c.headers['Connection'] = 'keep-alive'
|
||||
|
||||
// Return a marker response (body will be ignored, streaming will be handled by server)
|
||||
mut headers := http.new_header()
|
||||
headers.add_custom('Transfer-Encoding', 'chunked') or {}
|
||||
headers.add_custom('Connection', 'keep-alive') or {}
|
||||
|
||||
return http.Response{
|
||||
status_code: 200
|
||||
header: headers
|
||||
body: ''
|
||||
}
|
||||
}
|
||||
|
||||
// c_stream_text - Text streaming function for use in route handlers
|
||||
// Sets Content-Type: text/plain; charset=utf-8, Transfer-Encoding: chunked,
|
||||
// and X-Content-Type-Options: nosniff headers
|
||||
//
|
||||
// Parameters:
|
||||
// c: The request Context
|
||||
// callback: The callback function that writes text data to the stream
|
||||
// error_handler: Optional error handler called if an error occurs
|
||||
//
|
||||
// Returns:
|
||||
// http.Response with streaming marker headers
|
||||
//
|
||||
// Example:
|
||||
// app.get('/stream-text', fn (mut c hono.Context) http.Response {
|
||||
// return hono.c_stream_text(mut c, fn (mut stream hono.StreamContext) ! {
|
||||
// stream.write_string('Hello ')!
|
||||
// stream.sleep(100)
|
||||
// stream.writeln('World')!
|
||||
// })
|
||||
// })
|
||||
pub fn c_stream_text(mut c Context, callback StreamCallback, error_handler ...StreamErrorHandler) http.Response {
|
||||
// Generate unique stream ID
|
||||
stream_id := generate_stream_id()
|
||||
|
||||
// Store stream configuration
|
||||
config := StreamConfig{
|
||||
stream_type: .text
|
||||
callback: callback
|
||||
error_handler: if error_handler.len > 0 { error_handler[0] } else { none }
|
||||
}
|
||||
store_stream_config(stream_id, config)
|
||||
|
||||
// Mark context as streaming
|
||||
c.store['_stream'] = 'true'
|
||||
c.store['_stream_id'] = stream_id
|
||||
c.store['_stream_type'] = 'text'
|
||||
|
||||
// Set streaming headers
|
||||
c.headers['Content-Type'] = 'text/plain; charset=utf-8'
|
||||
c.headers['Transfer-Encoding'] = 'chunked'
|
||||
c.headers['X-Content-Type-Options'] = 'nosniff'
|
||||
c.headers['Connection'] = 'keep-alive'
|
||||
|
||||
// Return a marker response
|
||||
mut headers := http.new_header()
|
||||
headers.add_custom('Content-Type', 'text/plain; charset=utf-8') or {}
|
||||
headers.add_custom('Transfer-Encoding', 'chunked') or {}
|
||||
headers.add_custom('X-Content-Type-Options', 'nosniff') or {}
|
||||
headers.add_custom('Connection', 'keep-alive') or {}
|
||||
|
||||
return http.Response{
|
||||
status_code: 200
|
||||
header: headers
|
||||
body: ''
|
||||
}
|
||||
}
|
||||
|
||||
// c_stream_sse - Server-Sent Events streaming function for use in route handlers
|
||||
// Sets Content-Type: text/event-stream, Cache-Control: no-cache,
|
||||
// and Connection: keep-alive headers
|
||||
//
|
||||
// Parameters:
|
||||
// c: The request Context
|
||||
// callback: The callback function that writes SSE events to the stream
|
||||
// error_handler: Optional error handler called if an error occurs
|
||||
//
|
||||
// Returns:
|
||||
// http.Response with streaming marker headers
|
||||
//
|
||||
// Example:
|
||||
// app.get('/sse', fn (mut c hono.Context) http.Response {
|
||||
// return hono.c_stream_sse(mut c, fn (mut stream hono.StreamContext) ! {
|
||||
// stream.write_sse(hono.SSEEvent{
|
||||
// data: 'Hello World'
|
||||
// event: 'message'
|
||||
// id: '1'
|
||||
// })!
|
||||
// stream.sleep(1000)
|
||||
// stream.write_sse(hono.SSEEvent{
|
||||
// data: 'Another message'
|
||||
// event: 'update'
|
||||
// id: '2'
|
||||
// })!
|
||||
// })
|
||||
// })
|
||||
pub fn c_stream_sse(mut c Context, callback StreamCallback, error_handler ...StreamErrorHandler) http.Response {
|
||||
// Generate unique stream ID
|
||||
stream_id := generate_stream_id()
|
||||
|
||||
// Store stream configuration
|
||||
config := StreamConfig{
|
||||
stream_type: .sse
|
||||
callback: callback
|
||||
error_handler: if error_handler.len > 0 { error_handler[0] } else { none }
|
||||
}
|
||||
store_stream_config(stream_id, config)
|
||||
|
||||
// Mark context as streaming
|
||||
c.store['_stream'] = 'true'
|
||||
c.store['_stream_id'] = stream_id
|
||||
c.store['_stream_type'] = 'sse'
|
||||
|
||||
// Set SSE headers (Requirements 3.1, 3.2, 3.3)
|
||||
c.headers['Content-Type'] = 'text/event-stream'
|
||||
c.headers['Cache-Control'] = 'no-cache'
|
||||
c.headers['Connection'] = 'keep-alive'
|
||||
c.headers['Transfer-Encoding'] = 'chunked'
|
||||
|
||||
// Return a marker response
|
||||
mut headers := http.new_header()
|
||||
headers.add_custom('Content-Type', 'text/event-stream') or {}
|
||||
headers.add_custom('Cache-Control', 'no-cache') or {}
|
||||
headers.add_custom('Connection', 'keep-alive') or {}
|
||||
headers.add_custom('Transfer-Encoding', 'chunked') or {}
|
||||
|
||||
return http.Response{
|
||||
status_code: 200
|
||||
header: headers
|
||||
body: ''
|
||||
}
|
||||
}
|
||||
|
||||
// is_streaming_response - Check if a Context represents a streaming response
|
||||
// Used by server backends to detect streaming responses
|
||||
pub fn is_streaming_response(ctx Context) bool {
|
||||
return '_stream' in ctx.store && ctx.store['_stream'] == 'true'
|
||||
}
|
||||
|
||||
// get_stream_config - Get the stream configuration from Context
|
||||
// Returns the StreamConfig if this is a streaming response, none otherwise
|
||||
pub fn get_stream_config(ctx Context) ?StreamConfig {
|
||||
if !is_streaming_response(ctx) {
|
||||
return none
|
||||
}
|
||||
|
||||
stream_id := ctx.store['_stream_id'] or { return none }
|
||||
return retrieve_stream_config(stream_id)
|
||||
}
|
||||
|
||||
// cleanup_stream_config - Remove stream configuration after use
|
||||
// Should be called by server backends after streaming is complete
|
||||
pub fn cleanup_stream_config(ctx Context) {
|
||||
if stream_id := ctx.store['_stream_id'] {
|
||||
remove_stream_config(stream_id)
|
||||
}
|
||||
}
|
||||
|
||||
// execute_stream - Execute a streaming response with the given writer
|
||||
// This is the main entry point for server backends to handle streaming
|
||||
//
|
||||
// Parameters:
|
||||
// ctx: The request Context with streaming configuration
|
||||
// writer: The StreamWriter implementation for the server backend
|
||||
//
|
||||
// Returns:
|
||||
// true if streaming was executed, false if not a streaming response
|
||||
pub fn execute_stream(ctx Context, mut writer StreamWriter) bool {
|
||||
config := get_stream_config(ctx) or { return false }
|
||||
|
||||
// Create StreamContext with the writer
|
||||
mut stream_ctx := StreamContext.new(writer)
|
||||
|
||||
// Execute the callback and handle errors
|
||||
config.callback(mut stream_ctx) or {
|
||||
// Handle error
|
||||
if error_handler := config.error_handler {
|
||||
error_handler(err, mut stream_ctx)
|
||||
} else {
|
||||
// Default: log error to console
|
||||
eprintln('[SSE Stream Error] ${err.msg()}')
|
||||
}
|
||||
}
|
||||
|
||||
// Auto-close the stream when callback completes
|
||||
stream_ctx.close()
|
||||
|
||||
// Cleanup the stored configuration
|
||||
cleanup_stream_config(ctx)
|
||||
|
||||
return true
|
||||
}
|
||||
Reference in New Issue
Block a user