Phase 9.75-C: Convert HTTPServerBox, P2PBox, IntentBox to RwLock pattern
Co-authored-by: moe-charm <217100418+moe-charm@users.noreply.github.com>
This commit is contained in:
@ -45,7 +45,7 @@ use crate::boxes::{SocketBox, MapBox, ArrayBox};
|
|||||||
use crate::boxes::http_message_box::{HTTPRequestBox, HTTPResponseBox};
|
use crate::boxes::http_message_box::{HTTPRequestBox, HTTPResponseBox};
|
||||||
use crate::boxes::future::FutureBox;
|
use crate::boxes::future::FutureBox;
|
||||||
use std::any::Any;
|
use std::any::Any;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::RwLock;
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::thread;
|
use std::thread;
|
||||||
|
|
||||||
@ -53,26 +53,35 @@ use std::thread;
|
|||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct HTTPServerBox {
|
pub struct HTTPServerBox {
|
||||||
base: BoxBase,
|
base: BoxBase,
|
||||||
socket: Arc<Mutex<Option<SocketBox>>>,
|
socket: RwLock<Option<SocketBox>>,
|
||||||
routes: Arc<Mutex<HashMap<String, Box<dyn NyashBox>>>>,
|
routes: RwLock<HashMap<String, Box<dyn NyashBox>>>,
|
||||||
middleware: Arc<Mutex<Vec<Box<dyn NyashBox>>>>,
|
middleware: RwLock<Vec<Box<dyn NyashBox>>>,
|
||||||
running: Arc<Mutex<bool>>,
|
running: RwLock<bool>,
|
||||||
static_path: Arc<Mutex<Option<String>>>,
|
static_path: RwLock<Option<String>>,
|
||||||
timeout_seconds: Arc<Mutex<u64>>,
|
timeout_seconds: RwLock<u64>,
|
||||||
active_connections: Arc<Mutex<Vec<Box<dyn NyashBox>>>>,
|
active_connections: RwLock<Vec<Box<dyn NyashBox>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Clone for HTTPServerBox {
|
impl Clone for HTTPServerBox {
|
||||||
fn clone(&self) -> Self {
|
fn clone(&self) -> Self {
|
||||||
|
// State-preserving clone implementation following PR #87 pattern
|
||||||
|
let socket_val = self.socket.read().unwrap().clone();
|
||||||
|
let routes_val = self.routes.read().unwrap().clone();
|
||||||
|
let middleware_val = self.middleware.read().unwrap().clone();
|
||||||
|
let running_val = *self.running.read().unwrap();
|
||||||
|
let static_path_val = self.static_path.read().unwrap().clone();
|
||||||
|
let timeout_val = *self.timeout_seconds.read().unwrap();
|
||||||
|
let connections_val = self.active_connections.read().unwrap().clone();
|
||||||
|
|
||||||
Self {
|
Self {
|
||||||
base: BoxBase::new(), // New unique ID for clone
|
base: BoxBase::new(), // New unique ID for clone
|
||||||
socket: Arc::clone(&self.socket),
|
socket: RwLock::new(socket_val),
|
||||||
routes: Arc::clone(&self.routes),
|
routes: RwLock::new(routes_val),
|
||||||
middleware: Arc::clone(&self.middleware),
|
middleware: RwLock::new(middleware_val),
|
||||||
running: Arc::clone(&self.running),
|
running: RwLock::new(running_val),
|
||||||
static_path: Arc::clone(&self.static_path),
|
static_path: RwLock::new(static_path_val),
|
||||||
timeout_seconds: Arc::clone(&self.timeout_seconds),
|
timeout_seconds: RwLock::new(timeout_val),
|
||||||
active_connections: Arc::clone(&self.active_connections),
|
active_connections: RwLock::new(connections_val),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -81,13 +90,13 @@ impl HTTPServerBox {
|
|||||||
pub fn new() -> Self {
|
pub fn new() -> Self {
|
||||||
Self {
|
Self {
|
||||||
base: BoxBase::new(),
|
base: BoxBase::new(),
|
||||||
socket: Arc::new(Mutex::new(None)),
|
socket: RwLock::new(None),
|
||||||
routes: Arc::new(Mutex::new(HashMap::new())),
|
routes: RwLock::new(HashMap::new()),
|
||||||
middleware: Arc::new(Mutex::new(Vec::new())),
|
middleware: RwLock::new(Vec::new()),
|
||||||
running: Arc::new(Mutex::new(false)),
|
running: RwLock::new(false),
|
||||||
static_path: Arc::new(Mutex::new(None)),
|
static_path: RwLock::new(None),
|
||||||
timeout_seconds: Arc::new(Mutex::new(30)),
|
timeout_seconds: RwLock::new(30),
|
||||||
active_connections: Arc::new(Mutex::new(Vec::new())),
|
active_connections: RwLock::new(Vec::new()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -97,7 +106,7 @@ impl HTTPServerBox {
|
|||||||
let bind_result = socket.bind(address, port);
|
let bind_result = socket.bind(address, port);
|
||||||
|
|
||||||
if bind_result.to_string_box().value == "true" {
|
if bind_result.to_string_box().value == "true" {
|
||||||
match self.socket.lock() {
|
match self.socket.write() {
|
||||||
Ok(mut socket_guard) => {
|
Ok(mut socket_guard) => {
|
||||||
*socket_guard = Some(socket);
|
*socket_guard = Some(socket);
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
@ -113,7 +122,7 @@ impl HTTPServerBox {
|
|||||||
|
|
||||||
/// 接続待機開始
|
/// 接続待機開始
|
||||||
pub fn listen(&self, backlog: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
pub fn listen(&self, backlog: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
||||||
let socket_guard = match self.socket.lock() {
|
let socket_guard = match self.socket.read() {
|
||||||
Ok(guard) => guard,
|
Ok(guard) => guard,
|
||||||
Err(_) => return Box::new(StringBox::new("Error: Failed to acquire socket lock".to_string())),
|
Err(_) => return Box::new(StringBox::new("Error: Failed to acquire socket lock".to_string())),
|
||||||
};
|
};
|
||||||
@ -134,12 +143,12 @@ impl HTTPServerBox {
|
|||||||
/// HTTP サーバー開始(メインループ)
|
/// HTTP サーバー開始(メインループ)
|
||||||
pub fn start(&self) -> Box<dyn NyashBox> {
|
pub fn start(&self) -> Box<dyn NyashBox> {
|
||||||
// Set running state
|
// Set running state
|
||||||
match self.running.lock() {
|
match self.running.write() {
|
||||||
Ok(mut running) => *running = true,
|
Ok(mut running) => *running = true,
|
||||||
Err(_) => return Box::new(StringBox::new("Error: Failed to set running state".to_string())),
|
Err(_) => return Box::new(StringBox::new("Error: Failed to set running state".to_string())),
|
||||||
};
|
};
|
||||||
|
|
||||||
let socket_guard = match self.socket.lock() {
|
let socket_guard = match self.socket.read() {
|
||||||
Ok(guard) => guard,
|
Ok(guard) => guard,
|
||||||
Err(_) => return Box::new(StringBox::new("Error: Failed to acquire socket lock".to_string())),
|
Err(_) => return Box::new(StringBox::new("Error: Failed to acquire socket lock".to_string())),
|
||||||
};
|
};
|
||||||
@ -151,14 +160,10 @@ impl HTTPServerBox {
|
|||||||
|
|
||||||
println!("🚀 HTTP Server starting...");
|
println!("🚀 HTTP Server starting...");
|
||||||
|
|
||||||
// Main server loop
|
// Main server loop - need to handle RwLock references carefully for threading
|
||||||
let running = Arc::clone(&self.running);
|
|
||||||
let routes = Arc::clone(&self.routes);
|
|
||||||
let active_connections = Arc::clone(&self.active_connections);
|
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
// Check if server should stop
|
// Check if server should stop
|
||||||
let should_continue = match running.lock() {
|
let should_continue = match self.running.read() {
|
||||||
Ok(running_guard) => *running_guard,
|
Ok(running_guard) => *running_guard,
|
||||||
Err(_) => break, // Exit loop if we can't check running state
|
Err(_) => break, // Exit loop if we can't check running state
|
||||||
};
|
};
|
||||||
@ -177,25 +182,20 @@ impl HTTPServerBox {
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Add to active connections (with error handling)
|
// Add to active connections (with error handling)
|
||||||
if let Ok(mut connections) = active_connections.lock() {
|
if let Ok(mut connections) = self.active_connections.write() {
|
||||||
connections.push(Box::new(client_socket.clone()));
|
connections.push(Box::new(client_socket.clone()));
|
||||||
}
|
}
|
||||||
|
|
||||||
// Handle client in separate thread (simulate nowait)
|
// Handle client in separate thread (simulate nowait)
|
||||||
let routes_clone = Arc::clone(&routes);
|
// For RwLock pattern, we need to pass the data needed for the thread
|
||||||
let active_connections_clone = Arc::clone(&active_connections);
|
let routes_snapshot = match self.routes.read() {
|
||||||
|
Ok(routes_guard) => routes_guard.clone(),
|
||||||
|
Err(_) => continue, // Skip this connection if we can't read routes
|
||||||
|
};
|
||||||
|
|
||||||
thread::spawn(move || {
|
thread::spawn(move || {
|
||||||
Self::handle_client_request(client_socket, routes_clone);
|
Self::handle_client_request_with_routes(client_socket, routes_snapshot);
|
||||||
|
// Note: Connection cleanup is handled separately to avoid complex lifetime issues
|
||||||
// Remove from active connections when done
|
|
||||||
// Note: This is a simplified cleanup - real implementation would need proper tracking
|
|
||||||
let mut connections = active_connections_clone.lock().unwrap();
|
|
||||||
connections.retain(|conn| {
|
|
||||||
// Simple cleanup - remove all connections for now
|
|
||||||
// Real implementation would track by ID
|
|
||||||
false
|
|
||||||
});
|
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -207,10 +207,10 @@ impl HTTPServerBox {
|
|||||||
|
|
||||||
/// サーバー停止
|
/// サーバー停止
|
||||||
pub fn stop(&self) -> Box<dyn NyashBox> {
|
pub fn stop(&self) -> Box<dyn NyashBox> {
|
||||||
*self.running.lock().unwrap() = false;
|
*self.running.write().unwrap() = false;
|
||||||
|
|
||||||
// Close all active connections
|
// Close all active connections
|
||||||
let mut connections = self.active_connections.lock().unwrap();
|
let mut connections = self.active_connections.write().unwrap();
|
||||||
for connection in connections.iter() {
|
for connection in connections.iter() {
|
||||||
if let Some(socket) = connection.as_any().downcast_ref::<SocketBox>() {
|
if let Some(socket) = connection.as_any().downcast_ref::<SocketBox>() {
|
||||||
let _ = socket.close();
|
let _ = socket.close();
|
||||||
@ -219,7 +219,7 @@ impl HTTPServerBox {
|
|||||||
connections.clear();
|
connections.clear();
|
||||||
|
|
||||||
// Close server socket
|
// Close server socket
|
||||||
if let Some(ref socket) = *self.socket.lock().unwrap() {
|
if let Some(ref socket) = *self.socket.read().unwrap() {
|
||||||
let _ = socket.close();
|
let _ = socket.close();
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -232,7 +232,7 @@ impl HTTPServerBox {
|
|||||||
let path_str = path.to_string_box().value;
|
let path_str = path.to_string_box().value;
|
||||||
let route_key = format!("ANY {}", path_str);
|
let route_key = format!("ANY {}", path_str);
|
||||||
|
|
||||||
self.routes.lock().unwrap().insert(route_key, handler);
|
self.routes.write().unwrap().insert(route_key, handler);
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -241,7 +241,7 @@ impl HTTPServerBox {
|
|||||||
let path_str = path.to_string_box().value;
|
let path_str = path.to_string_box().value;
|
||||||
let route_key = format!("GET {}", path_str);
|
let route_key = format!("GET {}", path_str);
|
||||||
|
|
||||||
self.routes.lock().unwrap().insert(route_key, handler);
|
self.routes.write().unwrap().insert(route_key, handler);
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -250,7 +250,7 @@ impl HTTPServerBox {
|
|||||||
let path_str = path.to_string_box().value;
|
let path_str = path.to_string_box().value;
|
||||||
let route_key = format!("POST {}", path_str);
|
let route_key = format!("POST {}", path_str);
|
||||||
|
|
||||||
self.routes.lock().unwrap().insert(route_key, handler);
|
self.routes.write().unwrap().insert(route_key, handler);
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -259,7 +259,7 @@ impl HTTPServerBox {
|
|||||||
let path_str = path.to_string_box().value;
|
let path_str = path.to_string_box().value;
|
||||||
let route_key = format!("PUT {}", path_str);
|
let route_key = format!("PUT {}", path_str);
|
||||||
|
|
||||||
self.routes.lock().unwrap().insert(route_key, handler);
|
self.routes.write().unwrap().insert(route_key, handler);
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -268,28 +268,28 @@ impl HTTPServerBox {
|
|||||||
let path_str = path.to_string_box().value;
|
let path_str = path.to_string_box().value;
|
||||||
let route_key = format!("DELETE {}", path_str);
|
let route_key = format!("DELETE {}", path_str);
|
||||||
|
|
||||||
self.routes.lock().unwrap().insert(route_key, handler);
|
self.routes.write().unwrap().insert(route_key, handler);
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 静的ファイル配信パス設定
|
/// 静的ファイル配信パス設定
|
||||||
pub fn set_static_path(&self, path: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
pub fn set_static_path(&self, path: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
||||||
let path_str = path.to_string_box().value;
|
let path_str = path.to_string_box().value;
|
||||||
*self.static_path.lock().unwrap() = Some(path_str);
|
*self.static_path.write().unwrap() = Some(path_str);
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// リクエストタイムアウト設定
|
/// リクエストタイムアウト設定
|
||||||
pub fn set_timeout(&self, seconds: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
pub fn set_timeout(&self, seconds: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
||||||
let timeout_val = seconds.to_string_box().value.parse::<u64>().unwrap_or(30);
|
let timeout_val = seconds.to_string_box().value.parse::<u64>().unwrap_or(30);
|
||||||
*self.timeout_seconds.lock().unwrap() = timeout_val;
|
*self.timeout_seconds.write().unwrap() = timeout_val;
|
||||||
Box::new(BoolBox::new(true))
|
Box::new(BoolBox::new(true))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// クライアントリクエスト処理(内部メソッド)
|
/// クライアントリクエスト処理(内部メソッド)
|
||||||
fn handle_client_request(
|
fn handle_client_request_with_routes(
|
||||||
client_socket: SocketBox,
|
client_socket: SocketBox,
|
||||||
routes: Arc<Mutex<HashMap<String, Box<dyn NyashBox>>>>
|
routes: HashMap<String, Box<dyn NyashBox>>
|
||||||
) {
|
) {
|
||||||
// Read HTTP request
|
// Read HTTP request
|
||||||
let raw_request = client_socket.read_http_request();
|
let raw_request = client_socket.read_http_request();
|
||||||
@ -308,17 +308,16 @@ impl HTTPServerBox {
|
|||||||
println!("📬 {} {}", method, path);
|
println!("📬 {} {}", method, path);
|
||||||
|
|
||||||
// Find matching route
|
// Find matching route
|
||||||
let routes_guard = routes.lock().unwrap();
|
|
||||||
let route_key = format!("{} {}", method, path);
|
let route_key = format!("{} {}", method, path);
|
||||||
let fallback_key = format!("ANY {}", path);
|
let fallback_key = format!("ANY {}", path);
|
||||||
|
|
||||||
let response = if let Some(_handler) = routes_guard.get(&route_key) {
|
let response = if let Some(_handler) = routes.get(&route_key) {
|
||||||
// Found specific method route
|
// Found specific method route
|
||||||
// TODO: Actual handler invocation would need method calling infrastructure
|
// TODO: Actual handler invocation would need method calling infrastructure
|
||||||
HTTPResponseBox::create_json_response(
|
HTTPResponseBox::create_json_response(
|
||||||
Box::new(StringBox::new(r#"{"message": "Route found", "method": ""#.to_string() + &method + r#""}"#))
|
Box::new(StringBox::new(r#"{"message": "Route found", "method": ""#.to_string() + &method + r#""}"#))
|
||||||
)
|
)
|
||||||
} else if let Some(_handler) = routes_guard.get(&fallback_key) {
|
} else if let Some(_handler) = routes.get(&fallback_key) {
|
||||||
// Found generic route
|
// Found generic route
|
||||||
HTTPResponseBox::create_json_response(
|
HTTPResponseBox::create_json_response(
|
||||||
Box::new(StringBox::new(r#"{"message": "Generic route found"}"#))
|
Box::new(StringBox::new(r#"{"message": "Generic route found"}"#))
|
||||||
@ -328,8 +327,6 @@ impl HTTPServerBox {
|
|||||||
HTTPResponseBox::create_404_response()
|
HTTPResponseBox::create_404_response()
|
||||||
};
|
};
|
||||||
|
|
||||||
drop(routes_guard);
|
|
||||||
|
|
||||||
// Send response
|
// Send response
|
||||||
let response_str = response.to_http_string();
|
let response_str = response.to_http_string();
|
||||||
let _ = client_socket.write(response_str);
|
let _ = client_socket.write(response_str);
|
||||||
@ -338,13 +335,13 @@ impl HTTPServerBox {
|
|||||||
|
|
||||||
/// アクティブ接続数取得
|
/// アクティブ接続数取得
|
||||||
pub fn get_active_connections(&self) -> Box<dyn NyashBox> {
|
pub fn get_active_connections(&self) -> Box<dyn NyashBox> {
|
||||||
let connections = self.active_connections.lock().unwrap();
|
let connections = self.active_connections.read().unwrap();
|
||||||
Box::new(IntegerBox::new(connections.len() as i64))
|
Box::new(IntegerBox::new(connections.len() as i64))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// サーバー状態取得
|
/// サーバー状態取得
|
||||||
pub fn is_running(&self) -> Box<dyn NyashBox> {
|
pub fn is_running(&self) -> Box<dyn NyashBox> {
|
||||||
Box::new(BoolBox::new(*self.running.lock().unwrap()))
|
Box::new(BoolBox::new(*self.running.read().unwrap()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -354,9 +351,9 @@ impl NyashBox for HTTPServerBox {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn to_string_box(&self) -> StringBox {
|
fn to_string_box(&self) -> StringBox {
|
||||||
let running = *self.running.lock().unwrap();
|
let running = *self.running.read().unwrap();
|
||||||
let routes_count = self.routes.lock().unwrap().len();
|
let routes_count = self.routes.read().unwrap().len();
|
||||||
let connections_count = self.active_connections.lock().unwrap().len();
|
let connections_count = self.active_connections.read().unwrap().len();
|
||||||
|
|
||||||
StringBox::new(format!(
|
StringBox::new(format!(
|
||||||
"HTTPServer(id: {}, running: {}, routes: {}, connections: {})",
|
"HTTPServer(id: {}, running: {}, routes: {}, connections: {})",
|
||||||
@ -387,9 +384,9 @@ impl BoxCore for HTTPServerBox {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn fmt_box(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt_box(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
let running = *self.running.lock().unwrap();
|
let running = *self.running.read().unwrap();
|
||||||
let routes_count = self.routes.lock().unwrap().len();
|
let routes_count = self.routes.read().unwrap().len();
|
||||||
let connections_count = self.active_connections.lock().unwrap().len();
|
let connections_count = self.active_connections.read().unwrap().len();
|
||||||
|
|
||||||
write!(f, "HTTPServer(id: {}, running: {}, routes: {}, connections: {})",
|
write!(f, "HTTPServer(id: {}, running: {}, routes: {}, connections: {})",
|
||||||
self.base.id, running, routes_count, connections_count)
|
self.base.id, running, routes_count, connections_count)
|
||||||
|
|||||||
@ -34,59 +34,80 @@
|
|||||||
|
|
||||||
use crate::box_trait::{NyashBox, StringBox, BoolBox, BoxCore, BoxBase};
|
use crate::box_trait::{NyashBox, StringBox, BoolBox, BoxCore, BoxBase};
|
||||||
use std::any::Any;
|
use std::any::Any;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::RwLock;
|
||||||
use std::fmt::{self, Debug};
|
use std::fmt::{self, Debug};
|
||||||
|
|
||||||
/// IntentBox内部データ構造
|
/// IntentBox - 構造化メッセージBox (RwLock pattern)
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug)]
|
||||||
pub struct IntentBoxData {
|
pub struct IntentBox {
|
||||||
base: BoxBase,
|
base: BoxBase,
|
||||||
/// メッセージの種類 ("chat.message", "file.share"等)
|
/// メッセージの種類 ("chat.message", "file.share"等)
|
||||||
pub name: String,
|
name: RwLock<String>,
|
||||||
/// 任意のJSONデータ
|
/// 任意のJSONデータ
|
||||||
pub payload: serde_json::Value,
|
payload: RwLock<serde_json::Value>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// IntentBox - 構造化メッセージBox(Arc<Mutex>統一パターン)
|
impl Clone for IntentBox {
|
||||||
pub type IntentBox = Arc<Mutex<IntentBoxData>>;
|
fn clone(&self) -> Self {
|
||||||
|
let name_val = self.name.read().unwrap().clone();
|
||||||
|
let payload_val = self.payload.read().unwrap().clone();
|
||||||
|
|
||||||
impl IntentBoxData {
|
Self {
|
||||||
|
base: BoxBase::new(), // New unique ID for clone
|
||||||
|
name: RwLock::new(name_val),
|
||||||
|
payload: RwLock::new(payload_val),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl IntentBox {
|
||||||
/// 新しいIntentBoxを作成
|
/// 新しいIntentBoxを作成
|
||||||
pub fn new(name: String, payload: serde_json::Value) -> IntentBox {
|
pub fn new(name: String, payload: serde_json::Value) -> Self {
|
||||||
Arc::new(Mutex::new(IntentBoxData {
|
IntentBox {
|
||||||
base: BoxBase::new(),
|
base: BoxBase::new(),
|
||||||
name,
|
name: RwLock::new(name),
|
||||||
payload,
|
payload: RwLock::new(payload),
|
||||||
}))
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// メッセージ名を取得
|
/// メッセージ名を取得
|
||||||
pub fn get_name(&self) -> &str {
|
pub fn get_name(&self) -> Box<dyn NyashBox> {
|
||||||
&self.name
|
let name = self.name.read().unwrap().clone();
|
||||||
|
Box::new(StringBox::new(name))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// ペイロードを取得
|
/// ペイロードを取得
|
||||||
pub fn get_payload(&self) -> &serde_json::Value {
|
pub fn get_payload(&self) -> Box<dyn NyashBox> {
|
||||||
&self.payload
|
let payload = self.payload.read().unwrap().clone();
|
||||||
|
Box::new(StringBox::new(payload.to_string()))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// ペイロードを更新
|
/// ペイロードを更新
|
||||||
pub fn set_payload(&mut self, payload: serde_json::Value) {
|
pub fn set_payload(&self, payload: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
||||||
self.payload = payload;
|
let payload_str = payload.to_string_box().value;
|
||||||
|
match serde_json::from_str(&payload_str) {
|
||||||
|
Ok(json_val) => {
|
||||||
|
*self.payload.write().unwrap() = json_val;
|
||||||
|
Box::new(BoolBox::new(true))
|
||||||
|
},
|
||||||
|
Err(_) => Box::new(BoolBox::new(false))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl NyashBox for IntentBox {
|
impl NyashBox for IntentBox {
|
||||||
|
fn clone_box(&self) -> Box<dyn NyashBox> {
|
||||||
|
Box::new(self.clone())
|
||||||
|
}
|
||||||
|
|
||||||
fn to_string_box(&self) -> StringBox {
|
fn to_string_box(&self) -> StringBox {
|
||||||
let data = self.lock().unwrap();
|
let name = self.name.read().unwrap().clone();
|
||||||
StringBox::new(format!("IntentBox[{}]", data.name))
|
StringBox::new(format!("IntentBox[{}]", name))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn equals(&self, other: &dyn NyashBox) -> BoolBox {
|
fn equals(&self, other: &dyn NyashBox) -> BoolBox {
|
||||||
if let Some(other_intent) = other.as_any().downcast_ref::<IntentBox>() {
|
if let Some(other_intent) = other.as_any().downcast_ref::<IntentBox>() {
|
||||||
let self_data = self.lock().unwrap();
|
BoolBox::new(self.base.id == other_intent.base.id)
|
||||||
let other_data = other_intent.lock().unwrap();
|
|
||||||
BoolBox::new(self_data.base.id == other_data.base.id)
|
|
||||||
} else {
|
} else {
|
||||||
BoolBox::new(false)
|
BoolBox::new(false)
|
||||||
}
|
}
|
||||||
@ -95,25 +116,20 @@ impl NyashBox for IntentBox {
|
|||||||
fn type_name(&self) -> &'static str {
|
fn type_name(&self) -> &'static str {
|
||||||
"IntentBox"
|
"IntentBox"
|
||||||
}
|
}
|
||||||
|
|
||||||
fn clone_box(&self) -> Box<dyn NyashBox> {
|
|
||||||
let data = self.lock().unwrap();
|
|
||||||
Box::new(IntentBoxData::new(data.name.clone(), data.payload.clone()))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl BoxCore for IntentBox {
|
impl BoxCore for IntentBox {
|
||||||
fn box_id(&self) -> u64 {
|
fn box_id(&self) -> u64 {
|
||||||
self.lock().unwrap().base.id
|
self.base.id
|
||||||
}
|
}
|
||||||
|
|
||||||
fn parent_type_id(&self) -> Option<std::any::TypeId> {
|
fn parent_type_id(&self) -> Option<std::any::TypeId> {
|
||||||
self.lock().unwrap().base.parent_type_id
|
self.base.parent_type_id
|
||||||
}
|
}
|
||||||
|
|
||||||
fn fmt_box(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt_box(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
let data = self.lock().unwrap();
|
let name = self.name.read().unwrap().clone();
|
||||||
write!(f, "IntentBox[{}]", data.name)
|
write!(f, "IntentBox[{}]", name)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn as_any(&self) -> &dyn Any {
|
fn as_any(&self) -> &dyn Any {
|
||||||
@ -125,9 +141,9 @@ impl BoxCore for IntentBox {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl std::fmt::Display for IntentBoxData {
|
impl std::fmt::Display for IntentBox {
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
write!(f, "IntentBox[{}]", self.name)
|
self.fmt_box(f)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -40,29 +40,38 @@ use crate::boxes::IntentBox;
|
|||||||
use crate::transport::{Transport, InProcessTransport, TransportError};
|
use crate::transport::{Transport, InProcessTransport, TransportError};
|
||||||
use crate::messaging::IntentHandler;
|
use crate::messaging::IntentHandler;
|
||||||
use std::any::Any;
|
use std::any::Any;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::RwLock;
|
||||||
|
use std::collections::HashMap;
|
||||||
|
|
||||||
/// P2PBox内部データ構造
|
/// P2PBox - P2P通信ノード (RwLock pattern)
|
||||||
pub struct P2PBoxData {
|
#[derive(Debug)]
|
||||||
|
pub struct P2PBox {
|
||||||
base: BoxBase,
|
base: BoxBase,
|
||||||
node_id: String,
|
node_id: RwLock<String>,
|
||||||
transport: Arc<Mutex<Box<dyn Transport>>>,
|
transport: RwLock<Box<dyn Transport>>,
|
||||||
|
handlers: RwLock<HashMap<String, Box<dyn NyashBox>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl std::fmt::Debug for P2PBoxData {
|
impl Clone for P2PBox {
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn clone(&self) -> Self {
|
||||||
f.debug_struct("P2PBoxData")
|
// State-preserving clone implementation following PR #87 pattern
|
||||||
.field("base", &self.base)
|
let node_id_val = self.node_id.read().unwrap().clone();
|
||||||
.field("node_id", &self.node_id)
|
// Note: Transport cloning is complex, for now we create a new transport
|
||||||
.field("transport", &"<Transport>")
|
// In a full implementation, we'd need to properly handle transport state
|
||||||
.finish()
|
let transport_kind = TransportKind::InProcess; // Default for now
|
||||||
|
let new_transport: Box<dyn Transport> = match transport_kind {
|
||||||
|
TransportKind::InProcess => Box::new(InProcessTransport::new(node_id_val.clone())),
|
||||||
|
};
|
||||||
|
let handlers_val = HashMap::new(); // Start fresh for cloned instance
|
||||||
|
|
||||||
|
Self {
|
||||||
|
base: BoxBase::new(), // New unique ID for clone
|
||||||
|
node_id: RwLock::new(node_id_val),
|
||||||
|
transport: RwLock::new(new_transport),
|
||||||
|
handlers: RwLock::new(handlers_val),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// P2PBox - P2P通信ノード(Arc<Mutex>統一パターン)
|
|
||||||
pub type P2PBox = Arc<Mutex<P2PBoxData>>;
|
|
||||||
|
|
||||||
/// P2PBox作成時のトランスポート種類
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub enum TransportKind {
|
pub enum TransportKind {
|
||||||
InProcess,
|
InProcess,
|
||||||
@ -80,64 +89,82 @@ impl std::str::FromStr for TransportKind {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl P2PBoxData {
|
impl P2PBox {
|
||||||
/// 新しいP2PBoxを作成
|
/// 新しいP2PBoxを作成
|
||||||
pub fn new(node_id: String, transport_kind: TransportKind) -> P2PBox {
|
pub fn new(node_id: String, transport_kind: TransportKind) -> Self {
|
||||||
let transport: Box<dyn Transport> = match transport_kind {
|
let transport: Box<dyn Transport> = match transport_kind {
|
||||||
TransportKind::InProcess => Box::new(InProcessTransport::new(node_id.clone())),
|
TransportKind::InProcess => Box::new(InProcessTransport::new(node_id.clone())),
|
||||||
};
|
};
|
||||||
|
|
||||||
Arc::new(Mutex::new(P2PBoxData {
|
P2PBox {
|
||||||
base: BoxBase::new(),
|
base: BoxBase::new(),
|
||||||
node_id,
|
node_id: RwLock::new(node_id),
|
||||||
transport: Arc::new(Mutex::new(transport)),
|
transport: RwLock::new(transport),
|
||||||
}))
|
handlers: RwLock::new(HashMap::new()),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// ノードIDを取得
|
/// ノードIDを取得
|
||||||
pub fn get_node_id(&self) -> &str {
|
pub fn get_node_id(&self) -> Box<dyn NyashBox> {
|
||||||
&self.node_id
|
let node_id = self.node_id.read().unwrap().clone();
|
||||||
|
Box::new(StringBox::new(node_id))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 特定ノードにメッセージを送信
|
/// 特定ノードにメッセージを送信
|
||||||
pub fn send(&self, to: &str, intent: IntentBox) -> Result<(), TransportError> {
|
pub fn send(&self, to: Box<dyn NyashBox>, intent: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
||||||
let transport = self.transport.lock().unwrap();
|
let to_str = to.to_string_box().value;
|
||||||
transport.send(to, intent, Default::default())
|
|
||||||
|
// Extract IntentBox from the generic Box
|
||||||
|
if let Some(intent_box) = intent.as_any().downcast_ref::<IntentBox>() {
|
||||||
|
let transport = self.transport.read().unwrap();
|
||||||
|
match transport.send(&to_str, intent_box.clone(), Default::default()) {
|
||||||
|
Ok(()) => Box::new(BoolBox::new(true)),
|
||||||
|
Err(_) => Box::new(BoolBox::new(false)),
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
Box::new(BoolBox::new(false))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// イベントハンドラーを登録
|
/// イベントハンドラーを登録
|
||||||
pub fn on(&self, intent_name: &str, handler: IntentHandler) -> Result<(), String> {
|
pub fn on(&self, intent_name: Box<dyn NyashBox>, handler: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
||||||
// InProcessTransportの場合のハンドラー追加
|
let intent_str = intent_name.to_string_box().value;
|
||||||
// 現在は簡略化された実装
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
|
// For now, we'll store a simplified handler representation
|
||||||
|
// In a full implementation, this would need proper IntentHandler integration
|
||||||
|
let mut handlers = self.handlers.write().unwrap();
|
||||||
|
handlers.insert(intent_str, handler);
|
||||||
|
Box::new(BoolBox::new(true))
|
||||||
/// ノードが到達可能かチェック
|
/// ノードが到達可能かチェック
|
||||||
pub fn is_reachable(&self, node_id: &str) -> bool {
|
pub fn is_reachable(&self, node_id: Box<dyn NyashBox>) -> Box<dyn NyashBox> {
|
||||||
let transport = self.transport.lock().unwrap();
|
let node_str = node_id.to_string_box().value;
|
||||||
transport.is_reachable(node_id)
|
let transport = self.transport.read().unwrap();
|
||||||
|
Box::new(BoolBox::new(transport.is_reachable(&node_str)))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// トランスポート種類を取得
|
/// トランスポート種類を取得
|
||||||
pub fn get_transport_type(&self) -> String {
|
pub fn get_transport_type(&self) -> Box<dyn NyashBox> {
|
||||||
let transport = self.transport.lock().unwrap();
|
let transport = self.transport.read().unwrap();
|
||||||
transport.transport_type().to_string()
|
Box::new(StringBox::new(transport.transport_type().to_string()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
impl NyashBox for P2PBox {
|
impl NyashBox for P2PBox {
|
||||||
|
fn clone_box(&self) -> Box<dyn NyashBox> {
|
||||||
|
Box::new(self.clone())
|
||||||
|
}
|
||||||
|
|
||||||
fn to_string_box(&self) -> StringBox {
|
fn to_string_box(&self) -> StringBox {
|
||||||
let data = self.lock().unwrap();
|
let node_id = self.node_id.read().unwrap().clone();
|
||||||
StringBox::new(format!("P2PBox[{}:{}]", data.node_id, data.get_transport_type()))
|
let transport_type = self.transport.read().unwrap().transport_type().to_string();
|
||||||
|
StringBox::new(format!("P2PBox[{}:{}]", node_id, transport_type))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn equals(&self, other: &dyn NyashBox) -> BoolBox {
|
fn equals(&self, other: &dyn NyashBox) -> BoolBox {
|
||||||
if let Some(other_p2p) = other.as_any().downcast_ref::<P2PBox>() {
|
if let Some(other_p2p) = other.as_any().downcast_ref::<P2PBox>() {
|
||||||
let self_data = self.lock().unwrap();
|
BoolBox::new(self.base.id == other_p2p.base.id)
|
||||||
let other_data = other_p2p.lock().unwrap();
|
|
||||||
BoolBox::new(self_data.base.id == other_data.base.id)
|
|
||||||
} else {
|
} else {
|
||||||
BoolBox::new(false)
|
BoolBox::new(false)
|
||||||
}
|
}
|
||||||
@ -146,25 +173,21 @@ impl NyashBox for P2PBox {
|
|||||||
fn type_name(&self) -> &'static str {
|
fn type_name(&self) -> &'static str {
|
||||||
"P2PBox"
|
"P2PBox"
|
||||||
}
|
}
|
||||||
|
|
||||||
fn clone_box(&self) -> Box<dyn NyashBox> {
|
|
||||||
// P2PBoxは共有されるので、新しいインスタンスではなく同じ参照を返す
|
|
||||||
Box::new(self.clone())
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl BoxCore for P2PBox {
|
impl BoxCore for P2PBox {
|
||||||
fn box_id(&self) -> u64 {
|
fn box_id(&self) -> u64 {
|
||||||
self.lock().unwrap().base.id
|
self.base.id
|
||||||
}
|
}
|
||||||
|
|
||||||
fn parent_type_id(&self) -> Option<std::any::TypeId> {
|
fn parent_type_id(&self) -> Option<std::any::TypeId> {
|
||||||
self.lock().unwrap().base.parent_type_id
|
self.base.parent_type_id
|
||||||
}
|
}
|
||||||
|
|
||||||
fn fmt_box(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt_box(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
let data = self.lock().unwrap();
|
let node_id = self.node_id.read().unwrap().clone();
|
||||||
write!(f, "P2PBox[{}:{}]", data.node_id, data.get_transport_type())
|
let transport_type = self.transport.read().unwrap().transport_type().to_string();
|
||||||
|
write!(f, "P2PBox[{}:{}]", node_id, transport_type)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn as_any(&self) -> &dyn Any {
|
fn as_any(&self) -> &dyn Any {
|
||||||
@ -176,8 +199,8 @@ impl BoxCore for P2PBox {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl std::fmt::Display for P2PBoxData {
|
impl std::fmt::Display for P2PBox {
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
write!(f, "P2PBox[{}:{}]", self.node_id, self.get_transport_type())
|
self.fmt_box(f)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -529,7 +529,7 @@ impl NyashInterpreter {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
let intent_box = crate::boxes::intent_box::IntentBoxData::new(name, payload);
|
let intent_box = crate::boxes::intent_box::IntentBox::new(name, payload);
|
||||||
return Ok(Box::new(intent_box) as Box<dyn NyashBox>);
|
return Ok(Box::new(intent_box) as Box<dyn NyashBox>);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user