diff --git a/hyperprocess_macro/src/lib.rs b/hyperprocess_macro/src/lib.rs index c1a0be6..a0dbbf0 100644 --- a/hyperprocess_macro/src/lib.rs +++ b/hyperprocess_macro/src/lib.rs @@ -5,7 +5,6 @@ use syn::{ parse_macro_input, punctuated::Punctuated, spanned::Spanned, token::Comma, Expr, ItemImpl, Meta, ReturnType, }; -use uuid; //------------------------------------------------------------------------------ // Type Definitions @@ -914,18 +913,14 @@ fn generate_response_handling( // Instead of wrapping in HPMResponse enum, directly serialize the result let response_bytes = serde_json::to_vec(&result).unwrap(); - // Get headers from the new request context - let headers_opt = hyperware_app_common::APP_HELPERS.with(|helper_ctx| { - helper_ctx.borrow().current_http_request_id.as_ref().and_then(|id| { - hyperware_app_common::HTTP_REQUEST_CONTEXTS.with(|contexts| { - contexts.borrow().get(id).and_then(|req_ctx| { - if req_ctx.response_headers.is_empty() { - None - } else { - Some(req_ctx.response_headers.clone()) - } - }) - }) + // Get headers from the current HTTP context + let headers_opt = hyperware_app_common::APP_HELPERS.with(|helpers| { + helpers.borrow().current_http_context.as_ref().and_then(|ctx| { + if ctx.response_headers.is_empty() { + None + } else { + Some(ctx.response_headers.clone()) + } }) }); @@ -934,7 +929,7 @@ fn generate_response_handling( headers_opt, response_bytes ); - + // Clear HTTP context immediately after sending the response hyperware_app_common::clear_http_request_context(); } @@ -950,19 +945,13 @@ fn generate_async_handler_arm( variant_name: &syn::Ident, response_handling: proc_macro2::TokenStream, ) -> proc_macro2::TokenStream { - let hyper_macro = if func.is_http { - quote! { hyperware_app_common::hyper_http } - } else { - quote! { hyperware_app_common::hyper } - }; - if func.params.is_empty() { // Updated pattern to match struct variant with no fields quote! { HPMRequest::#variant_name{} => { // Create a raw pointer to state for use in the async block let state_ptr: *mut #self_ty = state; - #hyper_macro! { + hyperware_app_common::hyper! { // Inside the async block, use the pointer to access state let result = unsafe { (*state_ptr).#fn_name().await }; #response_handling @@ -976,7 +965,7 @@ fn generate_async_handler_arm( let param_captured = param; // Capture param before moving into async block // Create a raw pointer to state for use in the async block let state_ptr: *mut #self_ty = state; - #hyper_macro! { + hyperware_app_common::hyper! { // Inside the async block, use the pointer to access state let result = unsafe { (*state_ptr).#fn_name(param_captured).await }; #response_handling @@ -1000,7 +989,7 @@ fn generate_async_handler_arm( #(#capture_statements)* // Create a raw pointer to state for use in the async block let state_ptr: *mut #self_ty = state; - #hyper_macro! { + hyperware_app_common::hyper! { // Inside the async block, use the pointer to access state let result = unsafe { (*state_ptr).#fn_name(#(#captured_names),*).await }; #response_handling @@ -1096,101 +1085,96 @@ fn ws_method_opt_to_call(ws_method: &Option) -> proc_macro2::TokenSt } } -/// Generate message handler functions for message types -fn generate_message_handlers( - self_ty: &Box, - handler_arms: &HandlerDispatch, - ws_method_call: &proc_macro2::TokenStream, - http_handlers: &[&FunctionMetadata], -) -> proc_macro2::TokenStream { - let http_request_match_arms = &handler_arms.http; - let local_request_match_arms = &handler_arms.local; - let remote_request_match_arms = &handler_arms.remote; - // We now use the combined local_and_remote handlers for local messages - let local_and_remote_request_match_arms = &handler_arms.local_and_remote; - - // Collect all specific paths from ALL handlers (parameter-less AND parameter-based) - let specific_paths: Vec<_> = http_handlers - .iter() - .filter_map(|h| h.http_path.as_ref()) - .collect(); +//------------------------------------------------------------------------------ +// HTTP Helper Functions +//------------------------------------------------------------------------------ - // Generate method checking for HTTP handlers with parameters (Phase 2 only) - let http_handlers_with_params: Vec<_> = http_handlers - .iter() - .filter(|h| !h.params.is_empty()) - .collect(); +/// Generate HTTP context setup code +fn generate_http_context_setup() -> proc_macro2::TokenStream { + quote! { + hyperware_app_common::APP_HELPERS.with(|helpers| { + helpers.borrow_mut().current_http_context = Some(hyperware_app_common::HttpRequestContext { + request: http_request, + response_headers: std::collections::HashMap::new(), + }); + }); + hyperware_process_lib::logging::debug!("HTTP context established"); + } +} - let http_method_checks: Vec<_> = http_handlers_with_params - .iter() - .map(|handler| { - let variant_name = format_ident!("{}", &handler.variant_name); - let methods = &handler.http_methods; - let path = handler.http_path.as_deref().unwrap_or(""); +/// Generate HTTP context cleanup code +fn generate_http_context_cleanup() -> proc_macro2::TokenStream { + quote! { + hyperware_app_common::clear_http_request_context(); + } +} - quote! { - (stringify!(#variant_name), (vec![#(#methods),*], #path)) - } - }) - .collect(); +/// Generate HTTP error response handling +fn generate_http_error_response( + status: &str, + message: proc_macro2::TokenStream, +) -> proc_macro2::TokenStream { + let status_ident = format_ident!("{}", status); + let cleanup = generate_http_context_cleanup(); + quote! { + hyperware_process_lib::http::server::send_response( + hyperware_process_lib::http::StatusCode::#status_ident, + None, + #message.into_bytes() + ); + #cleanup + } +} - // Generate path checking for HTTP handlers with parameters (Phase 2 only) - let http_path_checks = http_handlers_with_params.iter().map(|handler| { - let variant_name = format_ident!("{}", &handler.variant_name); +/// Generate HTTP method and path parsing code +fn generate_http_request_parsing() -> proc_macro2::TokenStream { + quote! { + let http_method = hyperware_app_common::get_http_method() + .unwrap_or_else(|e| { + hyperware_process_lib::logging::warn!("Failed to get HTTP method from request context: {}", e); + "UNKNOWN".to_string() + }); - if let Some(path) = &handler.http_path { - quote! { - (stringify!(#variant_name), Some(#path)) - } - } else { - quote! { - (stringify!(#variant_name), None::<&str>) + let current_path = match hyperware_app_common::get_path() { + Ok(path) => { + hyperware_process_lib::logging::debug!("Successfully parsed HTTP path: '{}'", path); + path + }, + Err(e) => { + hyperware_process_lib::logging::error!("Failed to get HTTP path: {}", e); + hyperware_process_lib::http::server::send_response( + hyperware_process_lib::http::StatusCode::BAD_REQUEST, + None, + format!("Invalid path: {}", e).into_bytes(), + ); + hyperware_app_common::clear_http_request_context(); + return; } - } - }); - - // Generate variant names for pattern matching (Phase 2 only) - let http_variants_with_params: Vec<_> = http_handlers_with_params - .iter() - .map(|handler| format_ident!("{}", &handler.variant_name)) - .collect(); - - // --- Handler Dispatch --- - // Generate dispatch logic for all HTTP handlers. - // Priority logic: When there's a request body, prioritize parameterized handlers. - // When there's no body, prioritize parameter-less handlers. - - // First, separate handlers by parameter presence - let parameterized_handlers: Vec<_> = http_handlers - .iter() - .filter(|h| !h.params.is_empty()) - .collect(); - let parameterless_handlers: Vec<_> = http_handlers - .iter() - .filter(|h| h.params.is_empty()) - .collect(); - - // Sort each group by path specificity (specific paths before dynamic) - let mut sorted_parameterized: Vec<_> = parameterized_handlers; - sorted_parameterized.sort_by_key(|handler| handler.http_path.is_none()); + }; + } +} - let mut sorted_parameterless: Vec<_> = parameterless_handlers; - sorted_parameterless.sort_by_key(|handler| handler.http_path.is_none()); +/// Generate parameterized handler dispatch arms +fn generate_parameterized_handler_dispatch( + parameterized_handlers: &[&&FunctionMetadata], + self_ty: &Box, + http_request_match_arms: &proc_macro2::TokenStream, + specific_paths: &[&String], +) -> proc_macro2::TokenStream { + let mut sorted_handlers = parameterized_handlers.to_vec(); + sorted_handlers.sort_by_key(|handler| handler.http_path.is_none()); - // Create dispatch arms with proper priority logic - let parameterized_dispatch_arms: Vec<_> = sorted_parameterized.iter().map(|handler| { + let dispatch_arms: Vec<_> = sorted_handlers.iter().map(|handler| { let fn_name = &handler.name; let variant_name = format_ident!("{}", &handler.variant_name); let path_check = if let Some(path) = &handler.http_path { quote! { ¤t_path == #path } } else { - // Dynamic routing: match any path EXCEPT those with specific handlers quote! { ![#(#specific_paths),*].contains(¤t_path.as_str()) } }; let methods = &handler.http_methods; let method_check = quote! { [#(#methods),*].contains(&http_method.as_str()) }; - // Handler with parameters - need request body quote! { hyperware_process_lib::logging::debug!("Checking parameterized handler {} for {} {} - path_check: {}, method_check: {}", stringify!(#fn_name), http_method, current_path, (#path_check), (#method_check)); @@ -1203,14 +1187,12 @@ fn generate_message_handlers( Ok(request) => { match request { HPMRequest::#variant_name(..) => { - // Correct variant - dispatch through the match unsafe { #http_request_match_arms hyperware_app_common::maybe_save_state(&mut *state); } }, _ => { - // Wrong variant name in request hyperware_process_lib::logging::error!("Request body contains wrong handler name for {} {}", http_method, current_path); hyperware_process_lib::http::server::send_response( hyperware_process_lib::http::StatusCode::BAD_REQUEST, @@ -1237,7 +1219,6 @@ fn generate_message_handlers( hyperware_process_lib::logging::error!("Failed to parse request body for {} {}: {}", http_method, current_path, error_details); - // Send appropriate error response instead of falling through hyperware_process_lib::http::server::send_response( hyperware_process_lib::http::StatusCode::BAD_REQUEST, None, @@ -1260,7 +1241,19 @@ fn generate_message_handlers( } }).collect(); - let parameterless_dispatch_arms: Vec<_> = sorted_parameterless.iter().map(|handler| { + quote! { #(#dispatch_arms)* } +} + +/// Generate parameterless handler dispatch arms +fn generate_parameterless_handler_dispatch( + parameterless_handlers: &[&&FunctionMetadata], + self_ty: &Box, + specific_paths: &[&String], +) -> proc_macro2::TokenStream { + let mut sorted_handlers = parameterless_handlers.to_vec(); + sorted_handlers.sort_by_key(|handler| handler.http_path.is_none()); + + let dispatch_arms: Vec<_> = sorted_handlers.iter().map(|handler| { let fn_name = &handler.name; let path_check = if let Some(path) = &handler.http_path { quote! { ¤t_path == #path } @@ -1270,7 +1263,6 @@ fn generate_message_handlers( let methods = &handler.http_methods; let method_check = quote! { [#(#methods),*].contains(&http_method.as_str()) }; - // Parameterless handler - direct dispatch let response_handling = quote! { let response_bytes = match serde_json::to_vec(&result) { Ok(bytes) => bytes, @@ -1285,21 +1277,14 @@ fn generate_message_handlers( } }; - // Get headers from HTTP_REQUEST_CONTEXTS if any are set - let headers_opt = hyperware_app_common::APP_HELPERS.with(|helper_ctx| { - if let Some(id) = &helper_ctx.borrow().current_http_request_id { - hyperware_app_common::HTTP_REQUEST_CONTEXTS.with(|contexts| { - contexts.borrow().get(id).and_then(|req_ctx| { - if req_ctx.response_headers.is_empty() { - None - } else { - Some(req_ctx.response_headers.clone()) - } - }) - }) - } else { - None - } + let headers_opt = hyperware_app_common::APP_HELPERS.with(|helpers| { + helpers.borrow().current_http_context.as_ref().and_then(|ctx| { + if ctx.response_headers.is_empty() { + None + } else { + Some(ctx.response_headers.clone()) + } + }) }); hyperware_process_lib::http::server::send_response( @@ -1308,15 +1293,13 @@ fn generate_message_handlers( response_bytes ); - // Clear headers and HTTP context after sending response - hyperware_app_common::clear_response_headers(); hyperware_app_common::clear_http_request_context(); }; let handler_body = if handler.is_async { quote! { let state_ptr: *mut #self_ty = state; - hyperware_app_common::hyper_http! { + hyperware_app_common::hyper! { let result = unsafe { (*state_ptr).#fn_name().await }; #response_handling } @@ -1341,185 +1324,160 @@ fn generate_message_handlers( } }).collect(); - quote! { - /// Handle messages from the HTTP server - fn handle_http_server_message(state: *mut #self_ty, message: hyperware_process_lib::Message) { - // Get the blob early for all message types - HTTP and WebSocket both might need it - let blob_opt = message.blob(); - - // Parse HTTP server request - match serde_json::from_slice::(message.body()) { - Ok(http_server_request) => { - match http_server_request { - hyperware_process_lib::http::server::HttpServerRequest::Http(http_request) => { - // Use the already captured blob for HTTP requests - - // Debug the message structure - hyperware_process_lib::logging::debug!("Processing HTTP request, message has blob: {}", blob_opt.is_some()); - if let Some(ref blob) = blob_opt { - hyperware_process_lib::logging::debug!("Blob size: {} bytes, content: {}", blob.bytes.len(), String::from_utf8_lossy(&blob.bytes[..std::cmp::min(200, blob.bytes.len())])); - } - - // Create a unique ID for this request and set up its context - let request_id = uuid::Uuid::new_v4().to_string(); - hyperware_process_lib::logging::debug!("Setting up HTTP context with id: {}", request_id); - hyperware_app_common::HTTP_REQUEST_CONTEXTS.with(|contexts| { - contexts.borrow_mut().insert(request_id.clone(), hyperware_app_common::HttpRequestContext { - request: http_request, - response_headers: std::collections::HashMap::new(), - }); - }); - hyperware_app_common::APP_HELPERS.with(|helpers| { - helpers.borrow_mut().current_http_request_id = Some(request_id.clone()); - }); - hyperware_process_lib::logging::debug!("HTTP context established, id: {}", request_id); + quote! { #(#dispatch_arms)* } +} - // Get the HTTP method and path, handling potential errors - let http_method = hyperware_app_common::get_http_method() - .unwrap_or_else(|e| { - hyperware_process_lib::logging::warn!("Failed to get HTTP method from request context: {}", e); - "UNKNOWN".to_string() - }); +/// Generate HTTP handler dispatcher +fn generate_http_handler_dispatcher( + http_handlers: &[&FunctionMetadata], + self_ty: &Box, + http_request_match_arms: &proc_macro2::TokenStream, +) -> proc_macro2::TokenStream { + let specific_paths: Vec<_> = http_handlers + .iter() + .filter_map(|h| h.http_path.as_ref()) + .collect(); - let current_path = match hyperware_app_common::get_path() { - Ok(path) => { - hyperware_process_lib::logging::debug!("Successfully parsed HTTP path: '{}'", path); - path - }, - Err(e) => { - hyperware_process_lib::logging::error!("Failed to get HTTP path: {}", e); - hyperware_process_lib::http::server::send_response( - hyperware_process_lib::http::StatusCode::BAD_REQUEST, - None, - format!("Invalid path: {}", e).into_bytes(), - ); - hyperware_app_common::clear_http_request_context(); - return; - } - }; + let parameterized_handlers: Vec<_> = http_handlers + .iter() + .filter(|h| !h.params.is_empty()) + .collect(); - // Match request to handler based on method and path - hyperware_process_lib::logging::debug!("Starting handler matching for {} {}", http_method, current_path); - - // Priority logic: if there's a request body, try parameterized handlers first - if blob_opt.is_some() && !blob_opt.as_ref().unwrap().bytes.is_empty() { - hyperware_process_lib::logging::debug!("Request has body, using two-phase matching"); - - // Phase 1: Try to deserialize the body to get the variant name - if let Some(ref blob) = blob_opt { - match serde_json::from_slice::(&blob.bytes) { - Ok(request) => { - hyperware_process_lib::logging::debug!("Successfully parsed request body, dispatching to specific handler"); - // Phase 2: Dispatch to the specific handler based on the variant - unsafe { - #http_request_match_arms - hyperware_app_common::maybe_save_state(&mut *state); - } - // NOTE: Context cleanup happens after response is sent in the response handling code - return; - }, - Err(e) => { - let error_details = if blob.bytes.is_empty() { - "Request body is empty but was expected to contain handler parameters.".to_string() - } else if let Ok(json_value) = serde_json::from_slice::(&blob.bytes) { - format!( - "Invalid request format. Expected one of the parameterized handler formats, but got: {}", - serde_json::to_string(&json_value).unwrap_or_else(|_| "invalid JSON".to_string()) - ) - } else { - format!( - "Invalid JSON in request body. Parse error: {}", - e - ) - }; - - hyperware_process_lib::logging::error!("Failed to parse request body for {} {}: {}", http_method, current_path, error_details); - - // Send appropriate error response instead of falling through - hyperware_process_lib::http::server::send_response( - hyperware_process_lib::http::StatusCode::BAD_REQUEST, - None, - error_details.into_bytes() - ); - hyperware_app_common::clear_http_request_context(); - return; - } - } - } - } else { - hyperware_process_lib::logging::debug!("Request has no body, trying parameter-less handlers first"); - // If no body, try parameter-less handlers first - #(#parameterless_dispatch_arms)* - } + let parameterless_handlers: Vec<_> = http_handlers + .iter() + .filter(|h| h.params.is_empty()) + .collect(); - // If we reach here, no handler matched - return 404 - hyperware_process_lib::logging::error!("No handler found for {} {} - all handlers checked", http_method, current_path); - hyperware_process_lib::http::server::send_response( - hyperware_process_lib::http::StatusCode::NOT_FOUND, - None, - format!("No handler found for {} {}", http_method, current_path).into_bytes(), - ); - hyperware_app_common::clear_http_request_context(); - }, - hyperware_process_lib::http::server::HttpServerRequest::WebSocketPush { channel_id, message_type } => { - hyperware_process_lib::logging::debug!("Received WebSocket message on channel {}, type: {:?}", channel_id, message_type); - - // Use the already captured blob - let Some(blob) = blob_opt else { - hyperware_process_lib::logging::error!( - "Failed to get blob for WebSocketPush on channel {}. This indicates a malformed WebSocket message.", - channel_id - ); - return; - }; + let parameterized_dispatch = generate_parameterized_handler_dispatch( + ¶meterized_handlers, + self_ty, + http_request_match_arms, + &specific_paths, + ); - hyperware_process_lib::logging::debug!("Processing WebSocket message with {} bytes", blob.bytes.len()); - // Call the websocket handler if it exists - #ws_method_call + let parameterless_dispatch = + generate_parameterless_handler_dispatch(¶meterless_handlers, self_ty, &specific_paths); - // Save state if needed - unsafe { - hyperware_app_common::maybe_save_state(&mut *state); - } - }, - hyperware_process_lib::http::server::HttpServerRequest::WebSocketOpen { path, channel_id } => { - hyperware_process_lib::logging::debug!("WebSocket connection opened on path '{}' with channel {}", path, channel_id); - match hyperware_app_common::get_server() { - Some(server) => server.handle_websocket_open(&path, channel_id), - None => hyperware_process_lib::logging::error!("Failed to get server instance for WebSocket open event") - } - }, - hyperware_process_lib::http::server::HttpServerRequest::WebSocketClose(channel_id) => { - hyperware_process_lib::logging::debug!("WebSocket connection closed on channel {}", channel_id); - match hyperware_app_common::get_server() { - Some(server) => server.handle_websocket_close(channel_id), - None => hyperware_process_lib::logging::error!("Failed to get server instance for WebSocket close event") - } + quote! { + hyperware_process_lib::logging::debug!("Starting handler matching for {} {}", http_method, current_path); + + if blob_opt.is_some() && !blob_opt.as_ref().unwrap().bytes.is_empty() { + hyperware_process_lib::logging::debug!("Request has body, using two-phase matching"); + + if let Some(ref blob) = blob_opt { + match serde_json::from_slice::(&blob.bytes) { + Ok(request) => { + hyperware_process_lib::logging::debug!("Successfully parsed request body, dispatching to specific handler"); + unsafe { + #http_request_match_arms + hyperware_app_common::maybe_save_state(&mut *state); } + return; + }, + Err(e) => { + let error_details = if blob.bytes.is_empty() { + "Request body is empty but was expected to contain handler parameters.".to_string() + } else if let Ok(json_value) = serde_json::from_slice::(&blob.bytes) { + format!( + "Invalid request format. Expected one of the parameterized handler formats, but got: {}", + serde_json::to_string(&json_value).unwrap_or_else(|_| "invalid JSON".to_string()) + ) + } else { + format!( + "Invalid JSON in request body. Parse error: {}", + e + ) + }; + + hyperware_process_lib::logging::error!("Failed to parse request body for {} {}: {}", http_method, current_path, error_details); + + hyperware_process_lib::http::server::send_response( + hyperware_process_lib::http::StatusCode::BAD_REQUEST, + None, + error_details.into_bytes() + ); + hyperware_app_common::clear_http_request_context(); + return; } - }, - Err(e) => { - hyperware_process_lib::logging::error!( - "Failed to parse HTTP server request: {}\n\ - This usually indicates a malformed message to the HTTP server.", - e - ); } } + } else { + hyperware_process_lib::logging::debug!("Request has no body, trying parameter-less handlers first"); + #parameterless_dispatch } + hyperware_process_lib::logging::error!("No handler found for {} {} - all handlers checked", http_method, current_path); + hyperware_process_lib::http::server::send_response( + hyperware_process_lib::http::StatusCode::NOT_FOUND, + None, + format!("No handler found for {} {}", http_method, current_path).into_bytes(), + ); + hyperware_app_common::clear_http_request_context(); + } +} + +//------------------------------------------------------------------------------ +// WebSocket Helper Functions +//------------------------------------------------------------------------------ + +/// Generate WebSocket message handler +fn generate_websocket_handler( + ws_method_call: &proc_macro2::TokenStream, + self_ty: &Box, +) -> proc_macro2::TokenStream { + quote! { + hyperware_process_lib::http::server::HttpServerRequest::WebSocketPush { channel_id, message_type } => { + hyperware_process_lib::logging::debug!("Received WebSocket message on channel {}, type: {:?}", channel_id, message_type); + + let Some(blob) = blob_opt else { + hyperware_process_lib::logging::error!( + "Failed to get blob for WebSocketPush on channel {}. This indicates a malformed WebSocket message.", + channel_id + ); + return; + }; + + hyperware_process_lib::logging::debug!("Processing WebSocket message with {} bytes", blob.bytes.len()); + #ws_method_call + + unsafe { + hyperware_app_common::maybe_save_state(&mut *state); + } + }, + hyperware_process_lib::http::server::HttpServerRequest::WebSocketOpen { path, channel_id } => { + hyperware_process_lib::logging::debug!("WebSocket connection opened on path '{}' with channel {}", path, channel_id); + match hyperware_app_common::get_server() { + Some(server) => server.handle_websocket_open(&path, channel_id), + None => hyperware_process_lib::logging::error!("Failed to get server instance for WebSocket open event") + } + }, + hyperware_process_lib::http::server::HttpServerRequest::WebSocketClose(channel_id) => { + hyperware_process_lib::logging::debug!("WebSocket connection closed on channel {}", channel_id); + match hyperware_app_common::get_server() { + Some(server) => server.handle_websocket_close(channel_id), + None => hyperware_process_lib::logging::error!("Failed to get server instance for WebSocket close event") + } + } + } +} + +//------------------------------------------------------------------------------ +// Local/Remote Message Helper Functions +//------------------------------------------------------------------------------ + +/// Generate local message handler +fn generate_local_message_handler( + self_ty: &Box, + match_arms: &proc_macro2::TokenStream, +) -> proc_macro2::TokenStream { + quote! { /// Handle local messages fn handle_local_message(state: *mut #self_ty, message: hyperware_process_lib::Message) { hyperware_process_lib::logging::debug!("Processing local message from: {:?}", message.source()); - // Process the local request based on our handlers (now including both local and remote handlers) match serde_json::from_slice::(message.body()) { Ok(request) => { unsafe { - // Match on the request variant and call the appropriate handler - // Now using combined local_and_remote handlers - #local_and_remote_request_match_arms - - // Save state if needed + #match_arms hyperware_app_common::maybe_save_state(&mut *state); } }, @@ -1537,22 +1495,24 @@ fn generate_message_handlers( } } } + } +} +/// Generate remote message handler +fn generate_remote_message_handler( + self_ty: &Box, + match_arms: &proc_macro2::TokenStream, +) -> proc_macro2::TokenStream { + quote! { /// Handle remote messages fn handle_remote_message(state: *mut #self_ty, message: hyperware_process_lib::Message) { hyperware_process_lib::logging::debug!("Processing remote message from: {:?}", message.source()); - // Process the remote request based on our handlers match serde_json::from_slice::(message.body()) { Ok(request) => { - hyperware_process_lib::logging::debug!("Successfully deserialized remote request"); unsafe { - // Match on the request variant and call the appropriate handler - #remote_request_match_arms - - // Save state if needed + #match_arms hyperware_app_common::maybe_save_state(&mut *state); } - hyperware_process_lib::logging::debug!("Remote message processed successfully"); }, Err(e) => { let raw_body = String::from_utf8_lossy(message.body()); @@ -1571,6 +1531,63 @@ fn generate_message_handlers( } } +/// Generate message handler functions for message types +fn generate_message_handlers( + self_ty: &Box, + handler_arms: &HandlerDispatch, + ws_method_call: &proc_macro2::TokenStream, + http_handlers: &[&FunctionMetadata], +) -> proc_macro2::TokenStream { + let http_request_match_arms = &handler_arms.http; + let local_and_remote_request_match_arms = &handler_arms.local_and_remote; + let remote_request_match_arms = &handler_arms.remote; + + let http_context_setup = generate_http_context_setup(); + let http_request_parsing = generate_http_request_parsing(); + let http_dispatcher = + generate_http_handler_dispatcher(http_handlers, self_ty, http_request_match_arms); + let websocket_handlers = generate_websocket_handler(ws_method_call, self_ty); + let local_message_handler = + generate_local_message_handler(self_ty, local_and_remote_request_match_arms); + let remote_message_handler = + generate_remote_message_handler(self_ty, remote_request_match_arms); + + quote! { + /// Handle messages from the HTTP server + fn handle_http_server_message(state: *mut #self_ty, message: hyperware_process_lib::Message) { + let blob_opt = message.blob(); + + match serde_json::from_slice::(message.body()) { + Ok(http_server_request) => { + match http_server_request { + hyperware_process_lib::http::server::HttpServerRequest::Http(http_request) => { + hyperware_process_lib::logging::debug!("Processing HTTP request, message has blob: {}", blob_opt.is_some()); + if let Some(ref blob) = blob_opt { + hyperware_process_lib::logging::debug!("Blob size: {} bytes, content: {}", blob.bytes.len(), String::from_utf8_lossy(&blob.bytes[..std::cmp::min(200, blob.bytes.len())])); + } + + #http_context_setup + #http_request_parsing + #http_dispatcher + }, + #websocket_handlers + } + }, + Err(e) => { + hyperware_process_lib::logging::error!( + "Failed to parse HTTP server request: {}\n\ + This usually indicates a malformed message to the HTTP server.", + e + ); + } + } + } + + #local_message_handler + #remote_message_handler + } +} + /// Helper function to determine if an Expr is "None" fn is_none_literal(expr: &Expr) -> bool { if let Expr::Path(expr_path) = expr { diff --git a/hyperware_app_common/src/lib.rs b/hyperware_app_common/src/lib.rs index 3799919..9ad22d5 100644 --- a/hyperware_app_common/src/lib.rs +++ b/hyperware_app_common/src/lib.rs @@ -31,15 +31,14 @@ thread_local! { pub static RESPONSE_REGISTRY: RefCell>> = RefCell::new(HashMap::new()); - pub static HTTP_REQUEST_CONTEXTS: RefCell> = RefCell::new(HashMap::new()); - pub static APP_HELPERS: RefCell = RefCell::new(AppHelpers { current_server: None, current_message: None, - current_http_request_id: None, + current_http_context: None, }); } +#[derive(Clone)] pub struct HttpRequestContext { pub request: IncomingHttpRequest, pub response_headers: HashMap, @@ -53,34 +52,15 @@ pub struct AppContext { pub struct AppHelpers { pub current_server: Option<*mut HttpServer>, pub current_message: Option, - pub current_http_request_id: Option, + pub current_http_context: Option, } // Access function for the current path pub fn get_path() -> Result { - APP_HELPERS.with(|ctx| { - let helpers = ctx.borrow(); - match helpers.current_http_request_id.as_ref() { - Some(id) => { - HTTP_REQUEST_CONTEXTS.with(|contexts| { - let contexts = contexts.borrow(); - match contexts.get(id) { - Some(req_ctx) => { - let path_result = req_ctx.request.path() - .map_err(|e| e.to_string()); - path_result - }, - None => { - hyperware_process_lib::logging::debug!("No HTTP context found for id: {}", id); - Err("No HTTP context available".to_string()) - }, - } - }) - }, - None => { - hyperware_process_lib::logging::debug!("No current_http_request_id set"); - Err("No HTTP context available".to_string()) - }, + APP_HELPERS.with(|helpers| { + match &helpers.borrow().current_http_context { + Some(ctx) => ctx.request.path().map_err(|e| e.to_string()), + None => Err("No HTTP context available".to_string()), } }) } @@ -91,83 +71,38 @@ pub fn get_server() -> Option<&'static mut HttpServer> { } pub fn get_http_method() -> Result { - APP_HELPERS.with(|ctx| { - let helpers = ctx.borrow(); - hyperware_process_lib::logging::debug!("get_http_method called, current_http_request_id: {:?}", helpers.current_http_request_id); - match helpers.current_http_request_id.as_ref() { - Some(id) => { - HTTP_REQUEST_CONTEXTS.with(|contexts| { - let contexts = contexts.borrow(); - hyperware_process_lib::logging::debug!("Looking for method context with id: {}", id); - match contexts.get(id) { - Some(req_ctx) => { - let method_result = req_ctx.request.method() - .map(|m| m.to_string()) - .map_err(|e| e.to_string()); - hyperware_process_lib::logging::debug!("get_http_method result: {:?}", method_result); - method_result - }, - None => { - hyperware_process_lib::logging::debug!("No HTTP context found for method id: {}", id); - Err("No HTTP context available".to_string()) - }, - } - }) - }, - None => { - hyperware_process_lib::logging::debug!("No current_http_request_id set for method"); - Err("No HTTP context available".to_string()) - }, + APP_HELPERS.with(|helpers| { + match &helpers.borrow().current_http_context { + Some(ctx) => ctx.request.method() + .map(|m| m.to_string()) + .map_err(|e| e.to_string()), + None => Err("No HTTP context available".to_string()), } }) } // Set response headers that will be included in the HTTP response pub fn set_response_headers(headers: HashMap) { - APP_HELPERS.with(|helper_ctx| { - if let Some(id) = &helper_ctx.borrow().current_http_request_id { - HTTP_REQUEST_CONTEXTS.with(|contexts| { - if let Some(req_ctx) = contexts.borrow_mut().get_mut(id) { - req_ctx.response_headers = headers; - } - }) + APP_HELPERS.with(|helpers| { + if let Some(ctx) = &mut helpers.borrow_mut().current_http_context { + ctx.response_headers = headers; } }) } // Add a single response header pub fn add_response_header(key: String, value: String) { - APP_HELPERS.with(|helper_ctx| { - if let Some(id) = &helper_ctx.borrow().current_http_request_id { - HTTP_REQUEST_CONTEXTS.with(|contexts| { - if let Some(req_ctx) = contexts.borrow_mut().get_mut(id) { - req_ctx.response_headers.insert(key, value); - } - }) + APP_HELPERS.with(|helpers| { + if let Some(ctx) = &mut helpers.borrow_mut().current_http_context { + ctx.response_headers.insert(key, value); } }) } -// Clear all response headers -pub fn clear_response_headers() { - APP_HELPERS.with(|helper_ctx| { - if let Some(id) = &helper_ctx.borrow().current_http_request_id { - HTTP_REQUEST_CONTEXTS.with(|contexts| { - if let Some(req_ctx) = contexts.borrow_mut().get_mut(id) { - req_ctx.response_headers.clear(); - } - }) - } - }) -} pub fn clear_http_request_context() { - APP_HELPERS.with(|helper_ctx| { - if let Some(id) = helper_ctx.borrow_mut().current_http_request_id.take() { - HTTP_REQUEST_CONTEXTS.with(|contexts| { - contexts.borrow_mut().remove(&id); - }) - } + APP_HELPERS.with(|helpers| { + helpers.borrow_mut().current_http_context = None; }) } @@ -203,12 +138,7 @@ pub fn get_query_params() -> Option> { } pub struct Executor { - tasks: Vec, -} - -struct Task { - future: Pin>>, - http_request_id: Option, + tasks: Vec>>>, } impl Executor { @@ -217,19 +147,7 @@ impl Executor { } pub fn spawn(&mut self, fut: impl Future + 'static) { - self.tasks.push(Task { - future: Box::pin(fut), - http_request_id: None, - }); - } - - pub fn spawn_with_http_context(&mut self, fut: impl Future + 'static) { - let http_request_id = - APP_HELPERS.with(|helpers| helpers.borrow().current_http_request_id.clone()); - self.tasks.push(Task { - future: Box::pin(fut), - http_request_id, - }); + self.tasks.push(Box::pin(fut)); } pub fn poll_all_tasks(&mut self) { @@ -237,22 +155,9 @@ impl Executor { let mut completed = Vec::new(); for i in 0..self.tasks.len() { - let task_id = self.tasks[i].http_request_id.clone(); - - let original_id = APP_HELPERS.with(|helpers| { - let mut helpers = helpers.borrow_mut(); - let original_id = helpers.current_http_request_id.clone(); - helpers.current_http_request_id = task_id; - original_id - }); - - if let Poll::Ready(()) = self.tasks[i].future.as_mut().poll(&mut ctx) { + if let Poll::Ready(()) = self.tasks[i].as_mut().poll(&mut ctx) { completed.push(i); } - - APP_HELPERS.with(|helpers| { - helpers.borrow_mut().current_http_request_id = original_id; - }); } for idx in completed.into_iter().rev() { @@ -262,11 +167,21 @@ impl Executor { } struct ResponseFuture { correlation_id: String, + // Capture HTTP context at creation time + http_context: Option, } impl ResponseFuture { fn new(correlation_id: String) -> Self { - Self { correlation_id } + // Capture current HTTP context when future is created (at .await point) + let http_context = APP_HELPERS.with(|helpers| { + helpers.borrow().current_http_context.clone() + }); + + Self { + correlation_id, + http_context, + } } } @@ -282,6 +197,13 @@ impl Future for ResponseFuture { }); if let Some(bytes) = maybe_bytes { + // Restore this future's captured context + if let Some(ref context) = self.http_context { + APP_HELPERS.with(|helpers| { + helpers.borrow_mut().current_http_context = Some(context.clone()); + }); + } + Poll::Ready(bytes) } else { Poll::Pending @@ -373,16 +295,6 @@ macro_rules! hyper { }; } -#[macro_export] -macro_rules! hyper_http { - ($($code:tt)*) => { - $crate::APP_CONTEXT.with(|ctx| { - ctx.borrow_mut().executor.spawn_with_http_context(async move { - $($code)* - }) - }) - }; -} // Enum defining the state persistance behaviour #[derive(Clone)]