1880 lines
68 KiB
Rust
1880 lines
68 KiB
Rust
//! Nyash Net Plugin (HTTP minimal) — TypeBox v2
|
|
//! Provides ServerBox/RequestBox/ResponseBox/ClientBox and socket variants.
|
|
//! Pure in-process HTTP over localhost for E2E of BoxRef args/returns.
|
|
|
|
use once_cell::sync::Lazy;
|
|
use std::collections::{HashMap, VecDeque};
|
|
use std::io::Write as IoWrite;
|
|
use std::io::{Read, Write};
|
|
use std::net::{TcpListener, TcpStream};
|
|
use std::sync::{
|
|
atomic::{AtomicBool, AtomicU32, Ordering},
|
|
Arc, Mutex,
|
|
};
|
|
use std::time::Duration;
|
|
|
|
// ===== Simple logger (enabled when NYASH_NET_LOG=1) =====
|
|
static LOG_ON: Lazy<bool> = Lazy::new(|| std::env::var("NYASH_NET_LOG").unwrap_or_default() == "1");
|
|
static LOG_PATH: Lazy<String> = Lazy::new(|| {
|
|
std::env::var("NYASH_NET_LOG_FILE").unwrap_or_else(|_| "net_plugin.log".to_string())
|
|
});
|
|
static LOG_MTX: Lazy<Mutex<()>> = Lazy::new(|| Mutex::new(()));
|
|
|
|
fn net_log(msg: &str) {
|
|
if !*LOG_ON {
|
|
return;
|
|
}
|
|
// Always mirror to stderr for visibility
|
|
eprintln!("[net] {}", msg);
|
|
let _g = LOG_MTX.lock().unwrap();
|
|
if let Ok(mut f) = std::fs::OpenOptions::new()
|
|
.create(true)
|
|
.append(true)
|
|
.open(&*LOG_PATH)
|
|
{
|
|
let _ = writeln!(f, "[{:?}] {}", std::time::SystemTime::now(), msg);
|
|
}
|
|
}
|
|
|
|
macro_rules! netlog {
|
|
($($arg:tt)*) => {{ let s = format!($($arg)*); net_log(&s); }}
|
|
}
|
|
|
|
// Error codes
|
|
const OK: i32 = 0;
|
|
const E_SHORT: i32 = -1;
|
|
const E_INV_TYPE: i32 = -2;
|
|
const E_INV_METHOD: i32 = -3;
|
|
const E_INV_ARGS: i32 = -4;
|
|
const E_ERR: i32 = -5;
|
|
const E_INV_HANDLE: i32 = -8;
|
|
|
|
// Type IDs
|
|
const T_SERVER: u32 = 20;
|
|
const T_REQUEST: u32 = 21;
|
|
const T_RESPONSE: u32 = 22;
|
|
const T_CLIENT: u32 = 23;
|
|
// Socket
|
|
const T_SOCK_SERVER: u32 = 30;
|
|
const T_SOCK_CONN: u32 = 31;
|
|
const T_SOCK_CLIENT: u32 = 32;
|
|
|
|
// Methods
|
|
const M_BIRTH: u32 = 0;
|
|
|
|
// Server
|
|
const M_SERVER_START: u32 = 1;
|
|
const M_SERVER_STOP: u32 = 2;
|
|
const M_SERVER_ACCEPT: u32 = 3; // -> Handle(Request)
|
|
|
|
// Request
|
|
const M_REQ_PATH: u32 = 1; // -> String
|
|
const M_REQ_READ_BODY: u32 = 2; // -> Bytes (optional)
|
|
const M_REQ_RESPOND: u32 = 3; // arg: Handle(Response)
|
|
|
|
// Response
|
|
const M_RESP_SET_STATUS: u32 = 1; // arg: i32
|
|
const M_RESP_SET_HEADER: u32 = 2; // args: name, value (string)
|
|
const M_RESP_WRITE: u32 = 3; // arg: bytes/string
|
|
const M_RESP_READ_BODY: u32 = 4; // -> Bytes
|
|
const M_RESP_GET_STATUS: u32 = 5; // -> i32
|
|
const M_RESP_GET_HEADER: u32 = 6; // arg: name -> string (or empty)
|
|
|
|
// Client
|
|
const M_CLIENT_GET: u32 = 1; // arg: url -> Handle(Response)
|
|
const M_CLIENT_POST: u32 = 2; // args: url, body(bytes/string) -> Handle(Response)
|
|
|
|
// Socket Server
|
|
const M_SRV_BIRTH: u32 = 0;
|
|
const M_SRV_START: u32 = 1; // port
|
|
const M_SRV_STOP: u32 = 2;
|
|
const M_SRV_ACCEPT: u32 = 3; // -> Handle(T_SOCK_CONN)
|
|
const M_SRV_ACCEPT_TIMEOUT: u32 = 4; // ms -> Handle(T_SOCK_CONN) or void
|
|
|
|
// Socket Client
|
|
const M_SC_BIRTH: u32 = 0;
|
|
const M_SC_CONNECT: u32 = 1; // host, port -> Handle(T_SOCK_CONN)
|
|
|
|
// Socket Conn
|
|
const M_CONN_BIRTH: u32 = 0;
|
|
const M_CONN_SEND: u32 = 1; // bytes/string -> void
|
|
const M_CONN_RECV: u32 = 2; // -> bytes
|
|
const M_CONN_CLOSE: u32 = 3; // -> void
|
|
const M_CONN_RECV_TIMEOUT: u32 = 4; // ms -> bytes (empty if timeout)
|
|
|
|
// Global State
|
|
static SERVER_INSTANCES: Lazy<Mutex<HashMap<u32, ServerState>>> =
|
|
Lazy::new(|| Mutex::new(HashMap::new()));
|
|
static SERVER_START_SEQ: AtomicU32 = AtomicU32::new(1);
|
|
static ACTIVE_SERVER_ID: Lazy<Mutex<Option<u32>>> = Lazy::new(|| Mutex::new(None));
|
|
static LAST_ACCEPTED_REQ: Lazy<Mutex<Option<u32>>> = Lazy::new(|| Mutex::new(None));
|
|
static REQUESTS: Lazy<Mutex<HashMap<u32, RequestState>>> = Lazy::new(|| Mutex::new(HashMap::new()));
|
|
static RESPONSES: Lazy<Mutex<HashMap<u32, ResponseState>>> =
|
|
Lazy::new(|| Mutex::new(HashMap::new()));
|
|
static CLIENTS: Lazy<Mutex<HashMap<u32, ClientState>>> = Lazy::new(|| Mutex::new(HashMap::new()));
|
|
|
|
static SERVER_ID: AtomicU32 = AtomicU32::new(1);
|
|
static REQUEST_ID: AtomicU32 = AtomicU32::new(1);
|
|
static RESPONSE_ID: AtomicU32 = AtomicU32::new(1);
|
|
static CLIENT_ID: AtomicU32 = AtomicU32::new(1);
|
|
static SOCK_SERVER_ID: AtomicU32 = AtomicU32::new(1);
|
|
static SOCK_CONN_ID: AtomicU32 = AtomicU32::new(1);
|
|
static SOCK_CLIENT_ID: AtomicU32 = AtomicU32::new(1);
|
|
|
|
struct ServerState {
|
|
running: Arc<AtomicBool>,
|
|
port: i32,
|
|
pending: Arc<Mutex<VecDeque<u32>>>, // queue of request ids
|
|
handle: Mutex<Option<std::thread::JoinHandle<()>>>,
|
|
start_seq: u32,
|
|
}
|
|
|
|
struct RequestState {
|
|
path: String,
|
|
body: Vec<u8>,
|
|
response_id: Option<u32>,
|
|
// For HTTP-over-TCP server: map to an active accepted socket to respond on
|
|
server_conn_id: Option<u32>,
|
|
responded: bool,
|
|
}
|
|
|
|
struct ResponseState {
|
|
status: i32,
|
|
headers: HashMap<String, String>,
|
|
body: Vec<u8>,
|
|
// For HTTP-over-TCP client: associated socket connection id to read from
|
|
client_conn_id: Option<u32>,
|
|
parsed: bool,
|
|
}
|
|
|
|
struct ClientState;
|
|
|
|
// Socket types
|
|
struct SockServerState {
|
|
running: Arc<AtomicBool>,
|
|
pending: Arc<Mutex<VecDeque<u32>>>,
|
|
handle: Mutex<Option<std::thread::JoinHandle<()>>>,
|
|
}
|
|
|
|
struct SockConnState {
|
|
stream: Mutex<TcpStream>,
|
|
}
|
|
|
|
struct SockClientState;
|
|
|
|
#[no_mangle]
|
|
pub extern "C" fn nyash_plugin_abi() -> u32 {
|
|
1
|
|
}
|
|
|
|
#[no_mangle]
|
|
pub extern "C" fn nyash_plugin_init() -> i32 {
|
|
// Force initialize logging
|
|
let _ = *LOG_ON;
|
|
let _ = &*LOG_PATH;
|
|
netlog!(
|
|
"Net plugin initialized, LOG_ON={}, LOG_PATH={}",
|
|
*LOG_ON,
|
|
*LOG_PATH
|
|
);
|
|
eprintln!("Net plugin: LOG_ON={}, LOG_PATH={}", *LOG_ON, *LOG_PATH);
|
|
OK
|
|
}
|
|
|
|
#[no_mangle]
|
|
pub extern "C" fn nyash_plugin_invoke(
|
|
type_id: u32,
|
|
method_id: u32,
|
|
instance_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe {
|
|
match type_id {
|
|
T_SERVER => server_invoke(method_id, instance_id, args, args_len, result, result_len),
|
|
T_REQUEST => request_invoke(method_id, instance_id, args, args_len, result, result_len),
|
|
T_RESPONSE => {
|
|
response_invoke(method_id, instance_id, args, args_len, result, result_len)
|
|
}
|
|
T_CLIENT => client_invoke(method_id, instance_id, args, args_len, result, result_len),
|
|
T_SOCK_SERVER => {
|
|
sock_server_invoke(method_id, instance_id, args, args_len, result, result_len)
|
|
}
|
|
T_SOCK_CLIENT => {
|
|
sock_client_invoke(method_id, instance_id, args, args_len, result, result_len)
|
|
}
|
|
T_SOCK_CONN => {
|
|
sock_conn_invoke(method_id, instance_id, args, args_len, result, result_len)
|
|
}
|
|
_ => E_INV_TYPE,
|
|
}
|
|
}
|
|
}
|
|
|
|
// ===== TypeBox ABI v2 (per-Box resolve/invoke_id) =====
|
|
#[repr(C)]
|
|
pub struct NyashTypeBoxFfi {
|
|
pub abi_tag: u32, // 'TYBX'
|
|
pub version: u16, // 1
|
|
pub struct_size: u16, // sizeof(NyashTypeBoxFfi)
|
|
pub name: *const std::os::raw::c_char,
|
|
pub resolve: Option<extern "C" fn(*const std::os::raw::c_char) -> u32>,
|
|
pub invoke_id: Option<extern "C" fn(u32, u32, *const u8, usize, *mut u8, *mut usize) -> i32>,
|
|
pub capabilities: u64,
|
|
}
|
|
unsafe impl Sync for NyashTypeBoxFfi {}
|
|
|
|
use std::ffi::CStr;
|
|
extern "C" fn responsebox_resolve(name: *const std::os::raw::c_char) -> u32 {
|
|
if name.is_null() { return 0; }
|
|
let s = unsafe { CStr::from_ptr(name) }.to_string_lossy();
|
|
match s.as_ref() {
|
|
"setStatus" => M_RESP_SET_STATUS,
|
|
"setHeader" => M_RESP_SET_HEADER,
|
|
"write" => M_RESP_WRITE,
|
|
"readBody" => M_RESP_READ_BODY,
|
|
"getStatus" => M_RESP_GET_STATUS,
|
|
"getHeader" => M_RESP_GET_HEADER,
|
|
"birth" => M_BIRTH,
|
|
"fini" => u32::MAX,
|
|
_ => 0,
|
|
}
|
|
}
|
|
extern "C" fn clientbox_resolve(name: *const std::os::raw::c_char) -> u32 {
|
|
if name.is_null() { return 0; }
|
|
let s = unsafe { CStr::from_ptr(name) }.to_string_lossy();
|
|
match s.as_ref() {
|
|
"get" => M_CLIENT_GET,
|
|
"post" => M_CLIENT_POST,
|
|
"birth" => M_BIRTH,
|
|
"fini" => u32::MAX,
|
|
_ => 0,
|
|
}
|
|
}
|
|
|
|
extern "C" fn responsebox_invoke_id(
|
|
instance_id: u32,
|
|
method_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe { response_invoke(method_id, instance_id, args, args_len, result, result_len) }
|
|
}
|
|
|
|
extern "C" fn clientbox_invoke_id(
|
|
instance_id: u32,
|
|
method_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe { client_invoke(method_id, instance_id, args, args_len, result, result_len) }
|
|
}
|
|
|
|
#[no_mangle]
|
|
pub static nyash_typebox_ResponseBox: NyashTypeBoxFfi = NyashTypeBoxFfi {
|
|
abi_tag: 0x54594258,
|
|
version: 1,
|
|
struct_size: std::mem::size_of::<NyashTypeBoxFfi>() as u16,
|
|
name: b"ResponseBox\0".as_ptr() as *const std::os::raw::c_char,
|
|
resolve: Some(responsebox_resolve),
|
|
invoke_id: Some(responsebox_invoke_id),
|
|
capabilities: 0,
|
|
};
|
|
|
|
#[no_mangle]
|
|
pub static nyash_typebox_ClientBox: NyashTypeBoxFfi = NyashTypeBoxFfi {
|
|
abi_tag: 0x54594258,
|
|
version: 1,
|
|
struct_size: std::mem::size_of::<NyashTypeBoxFfi>() as u16,
|
|
name: b"ClientBox\0".as_ptr() as *const std::os::raw::c_char,
|
|
resolve: Some(clientbox_resolve),
|
|
invoke_id: Some(clientbox_invoke_id),
|
|
capabilities: 0,
|
|
};
|
|
|
|
// --- ServerBox ---
|
|
extern "C" fn serverbox_resolve(name: *const std::os::raw::c_char) -> u32 {
|
|
if name.is_null() { return 0; }
|
|
let s = unsafe { CStr::from_ptr(name) }.to_string_lossy();
|
|
match s.as_ref() {
|
|
"start" => M_SERVER_START,
|
|
"stop" => M_SERVER_STOP,
|
|
"accept" => M_SERVER_ACCEPT,
|
|
"birth" => M_BIRTH,
|
|
"fini" => u32::MAX,
|
|
_ => 0,
|
|
}
|
|
}
|
|
extern "C" fn serverbox_invoke_id(
|
|
instance_id: u32,
|
|
method_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe { server_invoke(method_id, instance_id, args, args_len, result, result_len) }
|
|
}
|
|
#[no_mangle]
|
|
pub static nyash_typebox_ServerBox: NyashTypeBoxFfi = NyashTypeBoxFfi {
|
|
abi_tag: 0x54594258,
|
|
version: 1,
|
|
struct_size: std::mem::size_of::<NyashTypeBoxFfi>() as u16,
|
|
name: b"ServerBox\0".as_ptr() as *const std::os::raw::c_char,
|
|
resolve: Some(serverbox_resolve),
|
|
invoke_id: Some(serverbox_invoke_id),
|
|
capabilities: 0,
|
|
};
|
|
|
|
// --- SockServerBox ---
|
|
extern "C" fn sockserver_resolve(name: *const std::os::raw::c_char) -> u32 {
|
|
if name.is_null() { return 0; }
|
|
let s = unsafe { CStr::from_ptr(name) }.to_string_lossy();
|
|
match s.as_ref() {
|
|
"start" => M_SRV_START,
|
|
"stop" => M_SRV_STOP,
|
|
"accept" => M_SRV_ACCEPT,
|
|
"acceptTimeout" => M_SRV_ACCEPT_TIMEOUT,
|
|
"birth" => M_SRV_BIRTH,
|
|
"fini" => u32::MAX,
|
|
_ => 0,
|
|
}
|
|
}
|
|
extern "C" fn sockserver_invoke_id(
|
|
instance_id: u32,
|
|
method_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe { sock_server_invoke(method_id, instance_id, args, args_len, result, result_len) }
|
|
}
|
|
#[no_mangle]
|
|
pub static nyash_typebox_SockServerBox: NyashTypeBoxFfi = NyashTypeBoxFfi {
|
|
abi_tag: 0x54594258,
|
|
version: 1,
|
|
struct_size: std::mem::size_of::<NyashTypeBoxFfi>() as u16,
|
|
name: b"SockServerBox\0".as_ptr() as *const std::os::raw::c_char,
|
|
resolve: Some(sockserver_resolve),
|
|
invoke_id: Some(sockserver_invoke_id),
|
|
capabilities: 0,
|
|
};
|
|
|
|
// --- SockClientBox ---
|
|
extern "C" fn sockclient_resolve(name: *const std::os::raw::c_char) -> u32 {
|
|
if name.is_null() { return 0; }
|
|
let s = unsafe { CStr::from_ptr(name) }.to_string_lossy();
|
|
match s.as_ref() {
|
|
"connect" => M_SC_CONNECT,
|
|
"birth" => M_SC_BIRTH,
|
|
"fini" => u32::MAX,
|
|
_ => 0,
|
|
}
|
|
}
|
|
extern "C" fn sockclient_invoke_id(
|
|
instance_id: u32,
|
|
method_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe { sock_client_invoke(method_id, instance_id, args, args_len, result, result_len) }
|
|
}
|
|
#[no_mangle]
|
|
pub static nyash_typebox_SockClientBox: NyashTypeBoxFfi = NyashTypeBoxFfi {
|
|
abi_tag: 0x54594258,
|
|
version: 1,
|
|
struct_size: std::mem::size_of::<NyashTypeBoxFfi>() as u16,
|
|
name: b"SockClientBox\0".as_ptr() as *const std::os::raw::c_char,
|
|
resolve: Some(sockclient_resolve),
|
|
invoke_id: Some(sockclient_invoke_id),
|
|
capabilities: 0,
|
|
};
|
|
|
|
// --- SockConnBox ---
|
|
extern "C" fn sockconn_resolve(name: *const std::os::raw::c_char) -> u32 {
|
|
if name.is_null() { return 0; }
|
|
let s = unsafe { CStr::from_ptr(name) }.to_string_lossy();
|
|
match s.as_ref() {
|
|
"send" => M_CONN_SEND,
|
|
"recv" => M_CONN_RECV,
|
|
"close" => M_CONN_CLOSE,
|
|
"recvTimeout" => M_CONN_RECV_TIMEOUT,
|
|
"birth" => M_CONN_BIRTH,
|
|
"fini" => u32::MAX,
|
|
_ => 0,
|
|
}
|
|
}
|
|
extern "C" fn sockconn_invoke_id(
|
|
instance_id: u32,
|
|
method_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe { sock_conn_invoke(method_id, instance_id, args, args_len, result, result_len) }
|
|
}
|
|
#[no_mangle]
|
|
pub static nyash_typebox_SockConnBox: NyashTypeBoxFfi = NyashTypeBoxFfi {
|
|
abi_tag: 0x54594258,
|
|
version: 1,
|
|
struct_size: std::mem::size_of::<NyashTypeBoxFfi>() as u16,
|
|
name: b"SockConnBox\0".as_ptr() as *const std::os::raw::c_char,
|
|
resolve: Some(sockconn_resolve),
|
|
invoke_id: Some(sockconn_invoke_id),
|
|
capabilities: 0,
|
|
};
|
|
extern "C" fn requestbox_resolve(name: *const std::os::raw::c_char) -> u32 {
|
|
if name.is_null() { return 0; }
|
|
let s = unsafe { CStr::from_ptr(name) }.to_string_lossy();
|
|
match s.as_ref() {
|
|
"path" => M_REQ_PATH,
|
|
"readBody" => M_REQ_READ_BODY,
|
|
"respond" => M_REQ_RESPOND,
|
|
"birth" => M_BIRTH,
|
|
"fini" => u32::MAX,
|
|
_ => 0,
|
|
}
|
|
}
|
|
extern "C" fn requestbox_invoke_id(
|
|
instance_id: u32,
|
|
method_id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
result: *mut u8,
|
|
result_len: *mut usize,
|
|
) -> i32 {
|
|
unsafe { request_invoke(method_id, instance_id, args, args_len, result, result_len) }
|
|
}
|
|
|
|
#[no_mangle]
|
|
pub static nyash_typebox_RequestBox: NyashTypeBoxFfi = NyashTypeBoxFfi {
|
|
abi_tag: 0x54594258,
|
|
version: 1,
|
|
struct_size: std::mem::size_of::<NyashTypeBoxFfi>() as u16,
|
|
name: b"RequestBox\0".as_ptr() as *const std::os::raw::c_char,
|
|
resolve: Some(requestbox_resolve),
|
|
invoke_id: Some(requestbox_invoke_id),
|
|
capabilities: 0,
|
|
};
|
|
|
|
unsafe fn server_invoke(
|
|
m: u32,
|
|
id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
res: *mut u8,
|
|
res_len: *mut usize,
|
|
) -> i32 {
|
|
match m {
|
|
M_BIRTH => {
|
|
let id = SERVER_ID.fetch_add(1, Ordering::Relaxed);
|
|
SERVER_INSTANCES.lock().unwrap().insert(
|
|
id,
|
|
ServerState {
|
|
running: Arc::new(AtomicBool::new(false)),
|
|
port: 0,
|
|
pending: Arc::new(Mutex::new(VecDeque::new())),
|
|
handle: Mutex::new(None),
|
|
start_seq: 0,
|
|
},
|
|
);
|
|
write_u32(id, res, res_len)
|
|
}
|
|
M_SERVER_START => {
|
|
// args: TLV string/int (port)
|
|
let port = tlv_parse_i32(slice(args, args_len)).unwrap_or(0);
|
|
if let Some(s) = SERVER_INSTANCES.lock().unwrap().get_mut(&id) {
|
|
s.port = port;
|
|
s.start_seq = SERVER_START_SEQ.fetch_add(1, Ordering::Relaxed);
|
|
let running = s.running.clone();
|
|
let pending = s.pending.clone();
|
|
running.store(true, Ordering::SeqCst);
|
|
// Bind listener synchronously to avoid race with client connect
|
|
let addr = format!("127.0.0.1:{}", port);
|
|
let listener = match TcpListener::bind(&addr) {
|
|
Ok(l) => {
|
|
netlog!("http:listener bound {}", addr);
|
|
l
|
|
}
|
|
Err(e) => {
|
|
netlog!("http:bind error {} err={:?}", addr, e);
|
|
running.store(false, Ordering::SeqCst);
|
|
return write_tlv_void(res, res_len);
|
|
}
|
|
};
|
|
// Spawn HTTP listener thread (real TCP)
|
|
let server_id_copy = id;
|
|
let handle = std::thread::spawn(move || {
|
|
let _ = listener.set_nonblocking(true);
|
|
loop {
|
|
if !running.load(Ordering::SeqCst) {
|
|
break;
|
|
}
|
|
match listener.accept() {
|
|
Ok((mut stream, _)) => {
|
|
// Parse minimal HTTP request (GET/POST)
|
|
let _ = stream.set_read_timeout(Some(Duration::from_millis(2000)));
|
|
if let Some((path, body, resp_hint)) =
|
|
read_http_request(&mut stream)
|
|
{
|
|
// Store stream for later respond()
|
|
let conn_id = SOCK_CONN_ID.fetch_add(1, Ordering::Relaxed);
|
|
SOCK_CONNS.lock().unwrap().insert(
|
|
conn_id,
|
|
SockConnState {
|
|
stream: Mutex::new(stream),
|
|
},
|
|
);
|
|
|
|
let req_id = REQUEST_ID.fetch_add(1, Ordering::Relaxed);
|
|
REQUESTS.lock().unwrap().insert(
|
|
req_id,
|
|
RequestState {
|
|
path,
|
|
body,
|
|
response_id: resp_hint,
|
|
server_conn_id: Some(conn_id),
|
|
responded: false,
|
|
},
|
|
);
|
|
if let Some(h) = resp_hint {
|
|
netlog!("http:accept linked resp_id hint={} for req_id={} conn_id={}", h, req_id, conn_id);
|
|
}
|
|
pending.lock().unwrap().push_back(req_id);
|
|
} else {
|
|
// Malformed; drop connection
|
|
}
|
|
}
|
|
Err(_) => {
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
}
|
|
}
|
|
});
|
|
*s.handle.lock().unwrap() = Some(handle);
|
|
}
|
|
// mark active server
|
|
*ACTIVE_SERVER_ID.lock().unwrap() = Some(id);
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
M_SERVER_STOP => {
|
|
if let Some(s) = SERVER_INSTANCES.lock().unwrap().get_mut(&id) {
|
|
s.running.store(false, Ordering::SeqCst);
|
|
if let Some(h) = s.handle.lock().unwrap().take() {
|
|
let _ = h.join();
|
|
}
|
|
}
|
|
// clear active if this server was active
|
|
let mut active = ACTIVE_SERVER_ID.lock().unwrap();
|
|
if active.map(|v| v == id).unwrap_or(false) {
|
|
*active = None;
|
|
}
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
M_SERVER_ACCEPT => {
|
|
// wait up to ~5000ms for a request to arrive
|
|
for _ in 0..1000 {
|
|
// Prefer TCP-backed requests (server_conn_id=Some) over stub ones
|
|
if let Some(req_id) = {
|
|
let mut map = SERVER_INSTANCES.lock().unwrap();
|
|
if let Some(s) = map.get_mut(&id) {
|
|
let mut q = s.pending.lock().unwrap();
|
|
// Find first index with TCP backing
|
|
let mut chosen: Option<usize> = None;
|
|
for i in 0..q.len() {
|
|
if let Some(rid) = q.get(i).copied() {
|
|
if let Some(rq) = REQUESTS.lock().unwrap().get(&rid) {
|
|
if rq.server_conn_id.is_some() {
|
|
chosen = Some(i);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if let Some(idx) = chosen {
|
|
q.remove(idx)
|
|
} else {
|
|
q.pop_front()
|
|
}
|
|
} else {
|
|
None
|
|
}
|
|
} {
|
|
netlog!("server.accept: return req_id={} srv_id={}", req_id, id);
|
|
*LAST_ACCEPTED_REQ.lock().unwrap() = Some(req_id);
|
|
return write_tlv_handle(T_REQUEST, req_id, res, res_len);
|
|
}
|
|
std::thread::sleep(Duration::from_millis(5));
|
|
}
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
_ => E_INV_METHOD,
|
|
}
|
|
}
|
|
|
|
unsafe fn request_invoke(
|
|
m: u32,
|
|
id: u32,
|
|
_args: *const u8,
|
|
_args_len: usize,
|
|
res: *mut u8,
|
|
res_len: *mut usize,
|
|
) -> i32 {
|
|
match m {
|
|
M_BIRTH => {
|
|
let id = REQUEST_ID.fetch_add(1, Ordering::Relaxed);
|
|
REQUESTS.lock().unwrap().insert(
|
|
id,
|
|
RequestState {
|
|
path: String::new(),
|
|
body: vec![],
|
|
response_id: None,
|
|
server_conn_id: None,
|
|
responded: false,
|
|
},
|
|
);
|
|
write_u32(id, res, res_len)
|
|
}
|
|
M_REQ_PATH => {
|
|
if let Some(rq) = REQUESTS.lock().unwrap().get(&id) {
|
|
write_tlv_string(&rq.path, res, res_len)
|
|
} else {
|
|
E_INV_HANDLE
|
|
}
|
|
}
|
|
M_REQ_READ_BODY => {
|
|
if let Some(rq) = REQUESTS.lock().unwrap().get(&id) {
|
|
write_tlv_bytes(&rq.body, res, res_len)
|
|
} else {
|
|
E_INV_HANDLE
|
|
}
|
|
}
|
|
M_REQ_RESPOND => {
|
|
// args: TLV Handle(Response)
|
|
let (t, provided_resp_id) = tlv_parse_handle(slice(_args, _args_len))
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or((0, 0));
|
|
if t != T_RESPONSE {
|
|
return E_INV_ARGS;
|
|
}
|
|
// Acquire request
|
|
let mut rq_map = REQUESTS.lock().unwrap();
|
|
if let Some(rq) = rq_map.get_mut(&id) {
|
|
netlog!(
|
|
"Request.respond: req_id={} provided_resp_id={} server_conn_id={:?} response_id_hint={:?}",
|
|
id, provided_resp_id, rq.server_conn_id, rq.response_id
|
|
);
|
|
// If request is backed by a real socket, write HTTP over that socket
|
|
if let Some(conn_id) = rq.server_conn_id {
|
|
drop(rq_map);
|
|
// Read response content from provided response handle
|
|
let (status, headers, body) = {
|
|
let resp_map = RESPONSES.lock().unwrap();
|
|
if let Some(src) = resp_map.get(&provided_resp_id) {
|
|
netlog!(
|
|
"Request.respond: Reading response id={}, status={}, body_len={}",
|
|
provided_resp_id,
|
|
src.status,
|
|
src.body.len()
|
|
);
|
|
(src.status, src.headers.clone(), src.body.clone())
|
|
} else {
|
|
netlog!(
|
|
"Request.respond: Response id={} not found!",
|
|
provided_resp_id
|
|
);
|
|
return E_INV_HANDLE;
|
|
}
|
|
};
|
|
// Build minimal HTTP/1.1 response
|
|
let reason = match status {
|
|
200 => "OK",
|
|
201 => "Created",
|
|
204 => "No Content",
|
|
400 => "Bad Request",
|
|
404 => "Not Found",
|
|
500 => "Internal Server Error",
|
|
_ => "OK",
|
|
};
|
|
let mut buf = Vec::new();
|
|
buf.extend_from_slice(format!("HTTP/1.1 {} {}\r\n", status, reason).as_bytes());
|
|
let mut has_len = false;
|
|
for (k, v) in &headers {
|
|
if k.eq_ignore_ascii_case("Content-Length") {
|
|
has_len = true;
|
|
}
|
|
buf.extend_from_slice(format!("{}: {}\r\n", k, v).as_bytes());
|
|
}
|
|
if !has_len {
|
|
buf.extend_from_slice(
|
|
format!("Content-Length: {}\r\n", body.len()).as_bytes(),
|
|
);
|
|
}
|
|
buf.extend_from_slice(b"Connection: close\r\n");
|
|
buf.extend_from_slice(b"\r\n");
|
|
buf.extend_from_slice(&body);
|
|
// Write and close
|
|
netlog!(
|
|
"Request.respond: Sending HTTP response, buf_len={}",
|
|
buf.len()
|
|
);
|
|
if let Some(conn) = SOCK_CONNS.lock().unwrap().remove(&conn_id) {
|
|
if let Ok(mut s) = conn.stream.lock() {
|
|
let _ = s.write_all(&buf);
|
|
let _ = s.flush();
|
|
netlog!(
|
|
"Request.respond: HTTP response sent to socket conn_id={}",
|
|
conn_id
|
|
);
|
|
}
|
|
} else {
|
|
netlog!("Request.respond: Socket conn_id={} not found!", conn_id);
|
|
}
|
|
// Also mirror to paired client Response handle to avoid race on immediate read
|
|
if let Some(target_id) = {
|
|
let rq_map2 = REQUESTS.lock().unwrap();
|
|
rq_map2.get(&id).and_then(|rq2| rq2.response_id)
|
|
} {
|
|
let mut resp_map = RESPONSES.lock().unwrap();
|
|
let dst = resp_map.entry(target_id).or_insert(ResponseState {
|
|
status: 200,
|
|
headers: HashMap::new(),
|
|
body: vec![],
|
|
client_conn_id: None,
|
|
parsed: true,
|
|
});
|
|
dst.status = status;
|
|
dst.headers = headers.clone();
|
|
dst.body = body.clone();
|
|
netlog!("Request.respond: mirrored client handle id={} body_len={} headers={} status={}", target_id, dst.body.len(), dst.headers.len(), dst.status);
|
|
}
|
|
// mark responded
|
|
{
|
|
let mut rq_map3 = REQUESTS.lock().unwrap();
|
|
if let Some(rq3) = rq_map3.get_mut(&id) {
|
|
rq3.responded = true;
|
|
}
|
|
}
|
|
return write_tlv_void(res, res_len);
|
|
}
|
|
|
|
// Not backed by a socket: attempt reroute to last accepted or latest TCP-backed unresponded request
|
|
drop(rq_map);
|
|
let candidate_req = {
|
|
if let Some(last_id) = *LAST_ACCEPTED_REQ.lock().unwrap() {
|
|
if let Some(r) = REQUESTS.lock().unwrap().get(&last_id) {
|
|
if r.server_conn_id.is_some() && !r.responded {
|
|
Some(last_id)
|
|
} else {
|
|
None
|
|
}
|
|
} else {
|
|
None
|
|
}
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
.or_else(|| {
|
|
REQUESTS
|
|
.lock()
|
|
.unwrap()
|
|
.iter()
|
|
.filter_map(|(rid, rqs)| {
|
|
if rqs.server_conn_id.is_some() && !rqs.responded {
|
|
Some(*rid)
|
|
} else {
|
|
None
|
|
}
|
|
})
|
|
.max()
|
|
});
|
|
if let Some(target_req_id) = candidate_req {
|
|
let (conn_id_alt, resp_hint_alt) = {
|
|
let map = REQUESTS.lock().unwrap();
|
|
let r = map.get(&target_req_id).unwrap();
|
|
(r.server_conn_id.unwrap(), r.response_id)
|
|
};
|
|
let (status, headers, body) = {
|
|
let resp_map = RESPONSES.lock().unwrap();
|
|
if let Some(src) = resp_map.get(&provided_resp_id) {
|
|
(src.status, src.headers.clone(), src.body.clone())
|
|
} else {
|
|
return E_INV_HANDLE;
|
|
}
|
|
};
|
|
let reason = match status {
|
|
200 => "OK",
|
|
201 => "Created",
|
|
204 => "No Content",
|
|
400 => "Bad Request",
|
|
404 => "Not Found",
|
|
500 => "Internal Server Error",
|
|
_ => "OK",
|
|
};
|
|
let mut buf = Vec::new();
|
|
buf.extend_from_slice(format!("HTTP/1.1 {} {}\r\n", status, reason).as_bytes());
|
|
let mut has_len = false;
|
|
for (k, v) in &headers {
|
|
if k.eq_ignore_ascii_case("Content-Length") {
|
|
has_len = true;
|
|
}
|
|
buf.extend_from_slice(format!("{}: {}\r\n", k, v).as_bytes());
|
|
}
|
|
if !has_len {
|
|
buf.extend_from_slice(
|
|
format!("Content-Length: {}\r\n", body.len()).as_bytes(),
|
|
);
|
|
}
|
|
buf.extend_from_slice(b"Connection: close\r\n\r\n");
|
|
buf.extend_from_slice(&body);
|
|
netlog!(
|
|
"Request.respond: reroute TCP send via req_id={} conn_id={}",
|
|
target_req_id,
|
|
conn_id_alt
|
|
);
|
|
if let Some(conn) = SOCK_CONNS.lock().unwrap().remove(&conn_id_alt) {
|
|
if let Ok(mut s) = conn.stream.lock() {
|
|
let _ = s.write_all(&buf);
|
|
let _ = s.flush();
|
|
}
|
|
}
|
|
if let Some(target_id) = resp_hint_alt {
|
|
let mut resp_map = RESPONSES.lock().unwrap();
|
|
let dst = resp_map.entry(target_id).or_insert(ResponseState {
|
|
status: 200,
|
|
headers: HashMap::new(),
|
|
body: vec![],
|
|
client_conn_id: None,
|
|
parsed: true,
|
|
});
|
|
dst.status = status;
|
|
dst.headers = headers.clone();
|
|
dst.body = body.clone();
|
|
netlog!("Request.respond: mirrored client handle id={} body_len={} headers={} status={}", target_id, dst.body.len(), dst.headers.len(), dst.status);
|
|
}
|
|
if let Some(rq4) = REQUESTS.lock().unwrap().get_mut(&target_req_id) {
|
|
rq4.responded = true;
|
|
}
|
|
return write_tlv_void(res, res_len);
|
|
}
|
|
netlog!("Request.respond: no suitable TCP-backed request found for reroute; invalid handle");
|
|
return E_INV_HANDLE;
|
|
}
|
|
E_INV_HANDLE
|
|
}
|
|
_ => E_INV_METHOD,
|
|
}
|
|
}
|
|
|
|
unsafe fn response_invoke(
|
|
m: u32,
|
|
id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
res: *mut u8,
|
|
res_len: *mut usize,
|
|
) -> i32 {
|
|
match m {
|
|
M_BIRTH => {
|
|
let id = RESPONSE_ID.fetch_add(1, Ordering::Relaxed);
|
|
RESPONSES.lock().unwrap().insert(
|
|
id,
|
|
ResponseState {
|
|
status: 200,
|
|
headers: HashMap::new(),
|
|
body: vec![],
|
|
client_conn_id: None,
|
|
parsed: false,
|
|
},
|
|
);
|
|
netlog!("Response.birth: new id={}", id);
|
|
write_u32(id, res, res_len)
|
|
}
|
|
M_RESP_SET_STATUS => {
|
|
let code = tlv_parse_i32(slice(args, args_len)).unwrap_or(200);
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get_mut(&id) {
|
|
rp.status = code;
|
|
}
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
M_RESP_SET_HEADER => {
|
|
if let Ok((name, value)) = tlv_parse_two_strings(slice(args, args_len)) {
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get_mut(&id) {
|
|
rp.headers.insert(name, value);
|
|
}
|
|
return write_tlv_void(res, res_len);
|
|
}
|
|
E_INV_ARGS
|
|
}
|
|
M_RESP_WRITE => {
|
|
// Accept String or Bytes
|
|
let bytes = tlv_parse_bytes(slice(args, args_len)).unwrap_or_default();
|
|
netlog!("HttpResponse.write: id={} bytes_len={}", id, bytes.len());
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get_mut(&id) {
|
|
rp.body.extend_from_slice(&bytes);
|
|
netlog!("HttpResponse.write: body now has {} bytes", rp.body.len());
|
|
}
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
M_RESP_READ_BODY => {
|
|
netlog!("HttpResponse.readBody: enter id={}", id);
|
|
// If bound to a client connection, lazily read and parse (with short retries)
|
|
for _ in 0..50 {
|
|
let need_parse = {
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get(&id) {
|
|
rp.client_conn_id
|
|
} else {
|
|
return E_INV_HANDLE;
|
|
}
|
|
};
|
|
if let Some(conn_id) = need_parse {
|
|
parse_client_response_into(id, conn_id);
|
|
std::thread::sleep(Duration::from_millis(5));
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get(&id) {
|
|
netlog!(
|
|
"HttpResponse.readBody: id={} body_len={}",
|
|
id,
|
|
rp.body.len()
|
|
);
|
|
write_tlv_bytes(&rp.body, res, res_len)
|
|
} else {
|
|
E_INV_HANDLE
|
|
}
|
|
}
|
|
M_RESP_GET_STATUS => {
|
|
for _ in 0..50 {
|
|
let need_parse = {
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get(&id) {
|
|
rp.client_conn_id
|
|
} else {
|
|
return E_INV_HANDLE;
|
|
}
|
|
};
|
|
if let Some(conn_id) = need_parse {
|
|
parse_client_response_into(id, conn_id);
|
|
std::thread::sleep(Duration::from_millis(5));
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get(&id) {
|
|
write_tlv_i32(rp.status, res, res_len)
|
|
} else {
|
|
E_INV_HANDLE
|
|
}
|
|
}
|
|
M_RESP_GET_HEADER => {
|
|
if let Ok(name) = tlv_parse_string(slice(args, args_len)) {
|
|
for _ in 0..50 {
|
|
let need_parse = {
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get(&id) {
|
|
rp.client_conn_id
|
|
} else {
|
|
return E_INV_HANDLE;
|
|
}
|
|
};
|
|
if let Some(conn_id) = need_parse {
|
|
parse_client_response_into(id, conn_id);
|
|
std::thread::sleep(Duration::from_millis(5));
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get(&id) {
|
|
let v = rp.headers.get(&name).cloned().unwrap_or_default();
|
|
return write_tlv_string(&v, res, res_len);
|
|
} else {
|
|
return E_INV_HANDLE;
|
|
}
|
|
}
|
|
E_INV_ARGS
|
|
}
|
|
_ => E_INV_METHOD,
|
|
}
|
|
}
|
|
|
|
unsafe fn client_invoke(
|
|
m: u32,
|
|
id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
res: *mut u8,
|
|
res_len: *mut usize,
|
|
) -> i32 {
|
|
match m {
|
|
M_BIRTH => {
|
|
let id = CLIENT_ID.fetch_add(1, Ordering::Relaxed);
|
|
CLIENTS.lock().unwrap().insert(id, ClientState);
|
|
write_u32(id, res, res_len)
|
|
}
|
|
M_CLIENT_GET => {
|
|
// args: TLV String(url)
|
|
let url = tlv_parse_string(slice(args, args_len)).unwrap_or_default();
|
|
let port = parse_port(&url).unwrap_or(80);
|
|
let host = parse_host(&url).unwrap_or_else(|| "127.0.0.1".to_string());
|
|
let path = parse_path(&url);
|
|
// Create client response handle first, so we can include it in header
|
|
let resp_id = RESPONSE_ID.fetch_add(1, Ordering::Relaxed);
|
|
let (_h, _p, req_bytes) = build_http_request("GET", &url, None, resp_id);
|
|
// Try TCP connect (best effort)
|
|
let mut tcp_ok = false;
|
|
if let Ok(mut stream) = TcpStream::connect(format!("{}:{}", host, port)) {
|
|
let _ = stream.write_all(&req_bytes);
|
|
let _ = stream.flush();
|
|
let conn_id = SOCK_CONN_ID.fetch_add(1, Ordering::Relaxed);
|
|
SOCK_CONNS.lock().unwrap().insert(
|
|
conn_id,
|
|
SockConnState {
|
|
stream: Mutex::new(stream),
|
|
},
|
|
);
|
|
// Map to server_id by port if available
|
|
let server_id_for_port = {
|
|
let servers = SERVER_INSTANCES.lock().unwrap();
|
|
servers
|
|
.iter()
|
|
.find(|(_, s)| s.port == port)
|
|
.map(|(sid, _)| *sid)
|
|
};
|
|
RESPONSES.lock().unwrap().insert(
|
|
resp_id,
|
|
ResponseState {
|
|
status: 0,
|
|
headers: HashMap::new(),
|
|
body: vec![],
|
|
client_conn_id: Some(conn_id),
|
|
parsed: false,
|
|
},
|
|
);
|
|
tcp_ok = true;
|
|
netlog!(
|
|
"client.get: url={} resp_id={} tcp_ok=true conn_id={}",
|
|
url,
|
|
resp_id,
|
|
conn_id
|
|
);
|
|
} else {
|
|
let server_id_for_port = {
|
|
let servers = SERVER_INSTANCES.lock().unwrap();
|
|
servers
|
|
.iter()
|
|
.find(|(_, s)| s.port == port)
|
|
.map(|(sid, _)| *sid)
|
|
};
|
|
RESPONSES.lock().unwrap().insert(
|
|
resp_id,
|
|
ResponseState {
|
|
status: 0,
|
|
headers: HashMap::new(),
|
|
body: vec![],
|
|
client_conn_id: None,
|
|
parsed: false,
|
|
},
|
|
);
|
|
netlog!("client.get: url={} resp_id={} tcp_ok=false", url, resp_id);
|
|
}
|
|
// No stub enqueue in TCP-only design
|
|
if tcp_ok {
|
|
write_tlv_handle(T_RESPONSE, resp_id, res, res_len)
|
|
} else {
|
|
// Encode error string; loader interprets returns_result=true methods' string payload as Err
|
|
let msg = format!(
|
|
"connect failed for {}:{}{}",
|
|
host,
|
|
port,
|
|
if path.is_empty() { "" } else { &path }
|
|
);
|
|
write_tlv_string(&msg, res, res_len)
|
|
}
|
|
}
|
|
M_CLIENT_POST => {
|
|
// args: TLV String(url), Bytes body
|
|
let data = slice(args, args_len);
|
|
let (_, argc, mut pos) = tlv_parse_header(data)
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or((1, 0, 4));
|
|
if argc < 2 {
|
|
return E_INV_ARGS;
|
|
}
|
|
let (_t1, s1, p1) = tlv_parse_entry_hdr(data, pos)
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or((0, 0, 0));
|
|
if data[pos] != 6 {
|
|
return E_INV_ARGS;
|
|
}
|
|
let url = std::str::from_utf8(&data[p1..p1 + s1])
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or("")
|
|
.to_string();
|
|
pos = p1 + s1;
|
|
let (t2, s2, p2) = tlv_parse_entry_hdr(data, pos)
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or((0, 0, 0));
|
|
if t2 != 6 && t2 != 7 {
|
|
return E_INV_ARGS;
|
|
}
|
|
let body = data[p2..p2 + s2].to_vec();
|
|
let port = parse_port(&url).unwrap_or(80);
|
|
let host = parse_host(&url).unwrap_or_else(|| "127.0.0.1".to_string());
|
|
let path = parse_path(&url);
|
|
let body_len = body.len();
|
|
// Create client response handle
|
|
let resp_id = RESPONSE_ID.fetch_add(1, Ordering::Relaxed);
|
|
let (_h, _p, req_bytes) = build_http_request("POST", &url, Some(&body), resp_id);
|
|
let mut tcp_ok = false;
|
|
if let Ok(mut stream) = TcpStream::connect(format!("{}:{}", host, port)) {
|
|
let _ = stream.write_all(&req_bytes);
|
|
let _ = stream.flush();
|
|
let conn_id = SOCK_CONN_ID.fetch_add(1, Ordering::Relaxed);
|
|
SOCK_CONNS.lock().unwrap().insert(
|
|
conn_id,
|
|
SockConnState {
|
|
stream: Mutex::new(stream),
|
|
},
|
|
);
|
|
let server_id_for_port = {
|
|
let servers = SERVER_INSTANCES.lock().unwrap();
|
|
servers
|
|
.iter()
|
|
.find(|(_, s)| s.port == port)
|
|
.map(|(sid, _)| *sid)
|
|
};
|
|
RESPONSES.lock().unwrap().insert(
|
|
resp_id,
|
|
ResponseState {
|
|
status: 0,
|
|
headers: HashMap::new(),
|
|
body: vec![],
|
|
client_conn_id: Some(conn_id),
|
|
parsed: false,
|
|
},
|
|
);
|
|
tcp_ok = true;
|
|
netlog!(
|
|
"client.post: url={} resp_id={} tcp_ok=true conn_id={} body_len={}",
|
|
url,
|
|
resp_id,
|
|
conn_id,
|
|
body.len()
|
|
);
|
|
} else {
|
|
let server_id_for_port = {
|
|
let servers = SERVER_INSTANCES.lock().unwrap();
|
|
servers
|
|
.iter()
|
|
.find(|(_, s)| s.port == port)
|
|
.map(|(sid, _)| *sid)
|
|
};
|
|
RESPONSES.lock().unwrap().insert(
|
|
resp_id,
|
|
ResponseState {
|
|
status: 0,
|
|
headers: HashMap::new(),
|
|
body: vec![],
|
|
client_conn_id: None,
|
|
parsed: false,
|
|
},
|
|
);
|
|
netlog!(
|
|
"client.post: url={} resp_id={} tcp_ok=false body_len={}",
|
|
url,
|
|
resp_id,
|
|
body.len()
|
|
);
|
|
}
|
|
// No stub enqueue in TCP-only design
|
|
if tcp_ok {
|
|
write_tlv_handle(T_RESPONSE, resp_id, res, res_len)
|
|
} else {
|
|
let msg = format!(
|
|
"connect failed for {}:{}{} (body_len={})",
|
|
host,
|
|
port,
|
|
if path.is_empty() { "" } else { &path },
|
|
body_len
|
|
);
|
|
write_tlv_string(&msg, res, res_len)
|
|
}
|
|
}
|
|
_ => E_INV_METHOD,
|
|
}
|
|
}
|
|
|
|
fn parse_path(url: &str) -> String {
|
|
// Robust-ish path extraction:
|
|
// - http://host:port/path -> "/path"
|
|
// - https://host/path -> "/path"
|
|
// - /relative -> as-is
|
|
// - otherwise -> "/"
|
|
if url.starts_with('/') {
|
|
return url.to_string();
|
|
}
|
|
if let Some(scheme_pos) = url.find("//") {
|
|
let after_scheme = &url[scheme_pos + 2..];
|
|
if let Some(slash) = after_scheme.find('/') {
|
|
return after_scheme[slash..].to_string();
|
|
} else {
|
|
return "/".to_string();
|
|
}
|
|
}
|
|
"/".to_string()
|
|
}
|
|
|
|
fn parse_port(url: &str) -> Option<i32> {
|
|
// match patterns like http://host:PORT/ or :PORT/
|
|
if let Some(pat) = url.split("//").nth(1) {
|
|
if let Some(after_host) = pat.split('/').next() {
|
|
if let Some(colon) = after_host.rfind(':') {
|
|
return after_host[colon + 1..].parse::<i32>().ok();
|
|
}
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
// ===== Helpers =====
|
|
unsafe fn slice<'a>(p: *const u8, len: usize) -> &'a [u8] {
|
|
std::slice::from_raw_parts(p, len)
|
|
}
|
|
|
|
fn write_u32(v: u32, res: *mut u8, res_len: *mut usize) -> i32 {
|
|
unsafe {
|
|
if res_len.is_null() {
|
|
return E_INV_ARGS;
|
|
}
|
|
if res.is_null() || *res_len < 4 {
|
|
*res_len = 4;
|
|
return E_SHORT;
|
|
}
|
|
let b = v.to_le_bytes();
|
|
std::ptr::copy_nonoverlapping(b.as_ptr(), res, 4);
|
|
*res_len = 4;
|
|
}
|
|
OK
|
|
}
|
|
|
|
fn write_tlv_result(payloads: &[(u8, &[u8])], res: *mut u8, res_len: *mut usize) -> i32 {
|
|
if res_len.is_null() {
|
|
return E_INV_ARGS;
|
|
}
|
|
let mut buf = Vec::with_capacity(4 + payloads.iter().map(|(_, p)| 4 + p.len()).sum::<usize>());
|
|
buf.extend_from_slice(&1u16.to_le_bytes());
|
|
buf.extend_from_slice(&(payloads.len() as u16).to_le_bytes());
|
|
for (tag, p) in payloads {
|
|
buf.push(*tag);
|
|
buf.push(0);
|
|
buf.extend_from_slice(&(p.len() as u16).to_le_bytes());
|
|
buf.extend_from_slice(p);
|
|
}
|
|
unsafe {
|
|
let need = buf.len();
|
|
if res.is_null() || *res_len < need {
|
|
*res_len = need;
|
|
return E_SHORT;
|
|
}
|
|
std::ptr::copy_nonoverlapping(buf.as_ptr(), res, need);
|
|
*res_len = need;
|
|
}
|
|
OK
|
|
}
|
|
|
|
fn write_tlv_void(res: *mut u8, res_len: *mut usize) -> i32 {
|
|
write_tlv_result(&[(9u8, &[])], res, res_len)
|
|
}
|
|
fn write_tlv_string(s: &str, res: *mut u8, res_len: *mut usize) -> i32 {
|
|
write_tlv_result(&[(6u8, s.as_bytes())], res, res_len)
|
|
}
|
|
fn write_tlv_bytes(b: &[u8], res: *mut u8, res_len: *mut usize) -> i32 {
|
|
write_tlv_result(&[(7u8, b)], res, res_len)
|
|
}
|
|
fn write_tlv_i32(v: i32, res: *mut u8, res_len: *mut usize) -> i32 {
|
|
write_tlv_result(&[(2u8, &v.to_le_bytes())], res, res_len)
|
|
}
|
|
fn write_tlv_handle(t: u32, id: u32, res: *mut u8, res_len: *mut usize) -> i32 {
|
|
let mut payload = [0u8; 8];
|
|
payload[0..4].copy_from_slice(&t.to_le_bytes());
|
|
payload[4..8].copy_from_slice(&id.to_le_bytes());
|
|
write_tlv_result(&[(8u8, &payload)], res, res_len)
|
|
}
|
|
|
|
fn tlv_parse_header(data: &[u8]) -> Result<(u16, u16, usize), ()> {
|
|
if data.len() < 4 {
|
|
return Err(());
|
|
}
|
|
let ver = u16::from_le_bytes([data[0], data[1]]);
|
|
let argc = u16::from_le_bytes([data[2], data[3]]);
|
|
if ver != 1 {
|
|
return Err(());
|
|
}
|
|
Ok((ver, argc, 4))
|
|
}
|
|
fn tlv_parse_string(data: &[u8]) -> Result<String, ()> {
|
|
let (_, argc, mut pos) = tlv_parse_header(data)?;
|
|
if argc < 1 {
|
|
return Err(());
|
|
}
|
|
let (tag, size, p) = tlv_parse_entry_hdr(data, pos)?;
|
|
if tag != 6 {
|
|
return Err(());
|
|
}
|
|
Ok(std::str::from_utf8(&data[p..p + size])
|
|
.map_err(|_| ())?
|
|
.to_string())
|
|
}
|
|
fn tlv_parse_two_strings(data: &[u8]) -> Result<(String, String), ()> {
|
|
let (_, argc, mut pos) = tlv_parse_header(data)?;
|
|
if argc < 2 {
|
|
return Err(());
|
|
}
|
|
let (tag1, size1, p1) = tlv_parse_entry_hdr(data, pos)?;
|
|
if tag1 != 6 {
|
|
return Err(());
|
|
}
|
|
let s1 = std::str::from_utf8(&data[p1..p1 + size1])
|
|
.map_err(|_| ())?
|
|
.to_string();
|
|
pos = p1 + size1;
|
|
let (tag2, size2, p2) = tlv_parse_entry_hdr(data, pos)?;
|
|
if tag2 != 6 {
|
|
return Err(());
|
|
}
|
|
let s2 = std::str::from_utf8(&data[p2..p2 + size2])
|
|
.map_err(|_| ())?
|
|
.to_string();
|
|
Ok((s1, s2))
|
|
}
|
|
fn tlv_parse_bytes(data: &[u8]) -> Result<Vec<u8>, ()> {
|
|
let (_, argc, mut pos) = tlv_parse_header(data)?;
|
|
if argc < 1 {
|
|
return Err(());
|
|
}
|
|
let (tag, size, p) = tlv_parse_entry_hdr(data, pos)?;
|
|
if tag != 6 && tag != 7 {
|
|
return Err(());
|
|
}
|
|
Ok(data[p..p + size].to_vec())
|
|
}
|
|
fn tlv_parse_i32(data: &[u8]) -> Result<i32, ()> {
|
|
let (_, argc, mut pos) = tlv_parse_header(data)?;
|
|
if argc < 1 {
|
|
return Err(());
|
|
}
|
|
let (tag, size, p) = tlv_parse_entry_hdr(data, pos)?;
|
|
match (tag, size) {
|
|
(2, 4) => {
|
|
let mut b = [0u8; 4];
|
|
b.copy_from_slice(&data[p..p + 4]);
|
|
Ok(i32::from_le_bytes(b))
|
|
}
|
|
(5, 8) => {
|
|
// accept i64
|
|
let mut b = [0u8; 8];
|
|
b.copy_from_slice(&data[p..p + 8]);
|
|
Ok(i64::from_le_bytes(b) as i32)
|
|
}
|
|
_ => Err(()),
|
|
}
|
|
}
|
|
fn tlv_parse_handle(data: &[u8]) -> Result<(u32, u32), ()> {
|
|
let (_, argc, mut pos) = tlv_parse_header(data)?;
|
|
if argc < 1 {
|
|
return Err(());
|
|
}
|
|
let (tag, size, p) = tlv_parse_entry_hdr(data, pos)?;
|
|
if tag != 8 || size != 8 {
|
|
return Err(());
|
|
}
|
|
let mut t = [0u8; 4];
|
|
let mut i = [0u8; 4];
|
|
t.copy_from_slice(&data[p..p + 4]);
|
|
i.copy_from_slice(&data[p + 4..p + 8]);
|
|
Ok((u32::from_le_bytes(t), u32::from_le_bytes(i)))
|
|
}
|
|
fn tlv_parse_entry_hdr(data: &[u8], pos: usize) -> Result<(u8, usize, usize), ()> {
|
|
if pos + 4 > data.len() {
|
|
return Err(());
|
|
}
|
|
let tag = data[pos];
|
|
let _rsv = data[pos + 1];
|
|
let size = u16::from_le_bytes([data[pos + 2], data[pos + 3]]) as usize;
|
|
let p = pos + 4;
|
|
if p + size > data.len() {
|
|
return Err(());
|
|
}
|
|
Ok((tag, size, p))
|
|
}
|
|
|
|
// ===== HTTP helpers =====
|
|
fn parse_host(url: &str) -> Option<String> {
|
|
// http://host[:port]/...
|
|
if let Some(rest) = url.split("//").nth(1) {
|
|
let host_port = rest.split('/').next().unwrap_or("");
|
|
let host = host_port.split(':').next().unwrap_or("");
|
|
if !host.is_empty() {
|
|
return Some(host.to_string());
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
fn build_http_request(
|
|
method: &str,
|
|
url: &str,
|
|
body: Option<&[u8]>,
|
|
resp_id: u32,
|
|
) -> (String, String, Vec<u8>) {
|
|
let host = parse_host(url).unwrap_or_else(|| "127.0.0.1".to_string());
|
|
let path = parse_path(url);
|
|
let mut buf = Vec::new();
|
|
buf.extend_from_slice(format!("{} {} HTTP/1.1\r\n", method, &path).as_bytes());
|
|
buf.extend_from_slice(format!("Host: {}\r\n", host).as_bytes());
|
|
buf.extend_from_slice(b"User-Agent: nyash-net-plugin/0.1\r\n");
|
|
// Embed client response handle id so server can mirror
|
|
buf.extend_from_slice(format!("X-Nyash-Resp-Id: {}\r\n", resp_id).as_bytes());
|
|
match body {
|
|
Some(b) => {
|
|
buf.extend_from_slice(format!("Content-Length: {}\r\n", b.len()).as_bytes());
|
|
buf.extend_from_slice(b"Content-Type: application/octet-stream\r\n");
|
|
buf.extend_from_slice(b"Connection: close\r\n\r\n");
|
|
buf.extend_from_slice(b);
|
|
}
|
|
None => {
|
|
buf.extend_from_slice(b"Connection: close\r\n\r\n");
|
|
}
|
|
}
|
|
(host, path, buf)
|
|
}
|
|
|
|
fn read_http_request(stream: &mut TcpStream) -> Option<(String, Vec<u8>, Option<u32>)> {
|
|
let mut buf = Vec::with_capacity(1024);
|
|
let mut tmp = [0u8; 1024];
|
|
// Read until we see CRLFCRLF
|
|
let header_end;
|
|
loop {
|
|
match stream.read(&mut tmp) {
|
|
Ok(0) => return None, // EOF without finding header end
|
|
Ok(n) => {
|
|
buf.extend_from_slice(&tmp[..n]);
|
|
if let Some(pos) = find_header_end(&buf) {
|
|
header_end = pos;
|
|
break;
|
|
}
|
|
if buf.len() > 64 * 1024 {
|
|
return None;
|
|
}
|
|
}
|
|
Err(_) => return None,
|
|
}
|
|
}
|
|
// Parse request line and headers
|
|
let header = &buf[..header_end];
|
|
let after = &buf[header_end + 4..];
|
|
let header_str = String::from_utf8_lossy(header);
|
|
let mut lines = header_str.split("\r\n");
|
|
let request_line = lines.next().unwrap_or("");
|
|
let mut parts = request_line.split_whitespace();
|
|
let method = parts.next().unwrap_or("");
|
|
let path = parts.next().unwrap_or("/").to_string();
|
|
let mut content_length: usize = 0;
|
|
let mut resp_handle_id: Option<u32> = None;
|
|
for line in lines {
|
|
if let Some((k, v)) = line.split_once(':') {
|
|
if k.eq_ignore_ascii_case("Content-Length") {
|
|
content_length = v.trim().parse().unwrap_or(0);
|
|
}
|
|
if k.eq_ignore_ascii_case("X-Nyash-Resp-Id") {
|
|
resp_handle_id = v.trim().parse::<u32>().ok();
|
|
}
|
|
}
|
|
}
|
|
let mut body = after.to_vec();
|
|
while body.len() < content_length {
|
|
match stream.read(&mut tmp) {
|
|
Ok(0) => break,
|
|
Ok(n) => body.extend_from_slice(&tmp[..n]),
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
if method == "GET" || method == "POST" {
|
|
Some((path, body, resp_handle_id))
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
fn find_header_end(buf: &[u8]) -> Option<usize> {
|
|
if buf.len() < 4 {
|
|
return None;
|
|
}
|
|
for i in 0..=buf.len() - 4 {
|
|
if &buf[i..i + 4] == b"\r\n\r\n" {
|
|
return Some(i);
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
fn parse_client_response_into(resp_id: u32, conn_id: u32) {
|
|
// Read full response from socket and fill ResponseState
|
|
let mut status: i32 = 200;
|
|
let mut headers: HashMap<String, String> = HashMap::new();
|
|
let mut body: Vec<u8> = Vec::new();
|
|
// Keep the connection until parsing succeeds; do not remove up front
|
|
let mut should_remove = false;
|
|
if let Ok(mut map) = SOCK_CONNS.lock() {
|
|
if let Some(conn) = map.get(&conn_id) {
|
|
if let Ok(mut s) = conn.stream.lock() {
|
|
let _ = s.set_read_timeout(Some(Duration::from_millis(4000)));
|
|
let mut buf = Vec::with_capacity(2048);
|
|
let mut tmp = [0u8; 2048];
|
|
loop {
|
|
match s.read(&mut tmp) {
|
|
Ok(0) => {
|
|
// EOF without header; keep connection for retry
|
|
return;
|
|
}
|
|
Ok(n) => {
|
|
buf.extend_from_slice(&tmp[..n]);
|
|
if find_header_end(&buf).is_some() {
|
|
break;
|
|
}
|
|
if buf.len() > 256 * 1024 {
|
|
break;
|
|
}
|
|
}
|
|
Err(_) => return,
|
|
}
|
|
}
|
|
if let Some(pos) = find_header_end(&buf) {
|
|
let header = &buf[..pos];
|
|
let after = &buf[pos + 4..];
|
|
// Parse status line and headers
|
|
let header_str = String::from_utf8_lossy(header);
|
|
let mut lines = header_str.split("\r\n");
|
|
if let Some(status_line) = lines.next() {
|
|
let mut sp = status_line.split_whitespace();
|
|
let _ver = sp.next();
|
|
if let Some(code_str) = sp.next() {
|
|
status = code_str.parse::<i32>().unwrap_or(200);
|
|
}
|
|
}
|
|
for line in lines {
|
|
if let Some((k, v)) = line.split_once(':') {
|
|
headers.insert(k.trim().to_string(), v.trim().to_string());
|
|
}
|
|
}
|
|
body.extend_from_slice(after);
|
|
let need = headers
|
|
.get("Content-Length")
|
|
.and_then(|v| v.parse::<usize>().ok())
|
|
.unwrap_or(0);
|
|
while body.len() < need {
|
|
match s.read(&mut tmp) {
|
|
Ok(0) => break,
|
|
Ok(n) => body.extend_from_slice(&tmp[..n]),
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
// Parsing succeeded; mark for removal
|
|
should_remove = true;
|
|
}
|
|
}
|
|
}
|
|
if should_remove {
|
|
map.remove(&conn_id);
|
|
}
|
|
}
|
|
if let Some(rp) = RESPONSES.lock().unwrap().get_mut(&resp_id) {
|
|
rp.status = status;
|
|
rp.headers = headers;
|
|
rp.body = body;
|
|
rp.parsed = true;
|
|
rp.client_conn_id = None;
|
|
}
|
|
}
|
|
|
|
// ===== Socket implementation =====
|
|
static SOCK_SERVERS: Lazy<Mutex<HashMap<u32, SockServerState>>> =
|
|
Lazy::new(|| Mutex::new(HashMap::new()));
|
|
static SOCK_CONNS: Lazy<Mutex<HashMap<u32, SockConnState>>> =
|
|
Lazy::new(|| Mutex::new(HashMap::new()));
|
|
static SOCK_CLIENTS: Lazy<Mutex<HashMap<u32, SockClientState>>> =
|
|
Lazy::new(|| Mutex::new(HashMap::new()));
|
|
|
|
unsafe fn sock_server_invoke(
|
|
m: u32,
|
|
id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
res: *mut u8,
|
|
res_len: *mut usize,
|
|
) -> i32 {
|
|
match m {
|
|
M_SRV_BIRTH => {
|
|
netlog!("sock:birth server");
|
|
let id = SOCK_SERVER_ID.fetch_add(1, Ordering::Relaxed);
|
|
SOCK_SERVERS.lock().unwrap().insert(
|
|
id,
|
|
SockServerState {
|
|
running: Arc::new(AtomicBool::new(false)),
|
|
pending: Arc::new(Mutex::new(VecDeque::new())),
|
|
handle: Mutex::new(None),
|
|
},
|
|
);
|
|
write_u32(id, res, res_len)
|
|
}
|
|
M_SRV_START => {
|
|
let port = tlv_parse_i32(slice(args, args_len)).unwrap_or(0);
|
|
netlog!("sock:start server id={} port={}", id, port);
|
|
if let Some(ss) = SOCK_SERVERS.lock().unwrap().get(&id) {
|
|
let running = ss.running.clone();
|
|
let pending = ss.pending.clone();
|
|
running.store(true, Ordering::SeqCst);
|
|
let handle = std::thread::spawn(move || {
|
|
let addr = format!("127.0.0.1:{}", port);
|
|
let listener = TcpListener::bind(addr);
|
|
if let Ok(listener) = listener {
|
|
listener.set_nonblocking(true).ok();
|
|
while running.load(Ordering::SeqCst) {
|
|
match listener.accept() {
|
|
Ok((stream, _)) => {
|
|
stream.set_nonblocking(false).ok();
|
|
let conn_id = SOCK_CONN_ID.fetch_add(1, Ordering::Relaxed);
|
|
SOCK_CONNS.lock().unwrap().insert(
|
|
conn_id,
|
|
SockConnState {
|
|
stream: Mutex::new(stream),
|
|
},
|
|
);
|
|
netlog!("sock:accept conn_id={}", conn_id);
|
|
pending.lock().unwrap().push_back(conn_id);
|
|
}
|
|
Err(_) => {
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
}
|
|
}
|
|
}
|
|
netlog!("sock:listener exit port={}", port);
|
|
}
|
|
});
|
|
*ss.handle.lock().unwrap() = Some(handle);
|
|
}
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
M_SRV_STOP => {
|
|
netlog!("sock:stop server id={}", id);
|
|
if let Some(ss) = SOCK_SERVERS.lock().unwrap().get(&id) {
|
|
ss.running.store(false, Ordering::SeqCst);
|
|
if let Some(h) = ss.handle.lock().unwrap().take() {
|
|
let _ = h.join();
|
|
}
|
|
}
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
M_SRV_ACCEPT => {
|
|
if let Some(ss) = SOCK_SERVERS.lock().unwrap().get(&id) {
|
|
// wait up to ~5000ms
|
|
for _ in 0..1000 {
|
|
if let Some(cid) = ss.pending.lock().unwrap().pop_front() {
|
|
netlog!("sock:accept returned conn_id={}", cid);
|
|
return write_tlv_handle(T_SOCK_CONN, cid, res, res_len);
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(5));
|
|
}
|
|
}
|
|
netlog!("sock:accept timeout id={}", id);
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
M_SRV_ACCEPT_TIMEOUT => {
|
|
let timeout_ms = tlv_parse_i32(slice(args, args_len)).unwrap_or(0).max(0) as u64;
|
|
if let Some(ss) = SOCK_SERVERS.lock().unwrap().get(&id) {
|
|
let deadline = std::time::Instant::now() + Duration::from_millis(timeout_ms);
|
|
loop {
|
|
if let Some(cid) = ss.pending.lock().unwrap().pop_front() {
|
|
netlog!("sock:acceptTimeout returned conn_id={}", cid);
|
|
return write_tlv_handle(T_SOCK_CONN, cid, res, res_len);
|
|
}
|
|
if std::time::Instant::now() >= deadline {
|
|
break;
|
|
}
|
|
std::thread::sleep(Duration::from_millis(5));
|
|
}
|
|
}
|
|
netlog!("sock:acceptTimeout timeout id={} ms={}", id, timeout_ms);
|
|
// Signal timeout as error for Result normalization
|
|
E_ERR
|
|
}
|
|
_ => E_INV_METHOD,
|
|
}
|
|
}
|
|
|
|
unsafe fn sock_client_invoke(
|
|
m: u32,
|
|
id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
res: *mut u8,
|
|
res_len: *mut usize,
|
|
) -> i32 {
|
|
match m {
|
|
M_SC_BIRTH => {
|
|
let id = SOCK_CLIENT_ID.fetch_add(1, Ordering::Relaxed);
|
|
SOCK_CLIENTS.lock().unwrap().insert(id, SockClientState);
|
|
write_u32(id, res, res_len)
|
|
}
|
|
M_SC_CONNECT => {
|
|
// args: host(string), port(i32)
|
|
let data = slice(args, args_len);
|
|
let (_, argc, mut pos) = tlv_parse_header(data)
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or((1, 0, 4));
|
|
if argc < 2 {
|
|
return E_INV_ARGS;
|
|
}
|
|
let (_t1, s1, p1) = tlv_parse_entry_hdr(data, pos)
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or((0, 0, 0));
|
|
if data[pos] != 6 {
|
|
return E_INV_ARGS;
|
|
}
|
|
let host = std::str::from_utf8(&data[p1..p1 + s1])
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or("")
|
|
.to_string();
|
|
pos = p1 + s1;
|
|
let (_t2, _s2, p2) = tlv_parse_entry_hdr(data, pos)
|
|
.map_err(|_| ())
|
|
.or(Err(()))
|
|
.unwrap_or((0, 0, 0));
|
|
let port = if data[pos] == 2 {
|
|
// i32
|
|
let mut b = [0u8; 4];
|
|
b.copy_from_slice(&data[p2..p2 + 4]);
|
|
i32::from_le_bytes(b)
|
|
} else {
|
|
return E_INV_ARGS;
|
|
};
|
|
let addr = format!("{}:{}", host, port);
|
|
match TcpStream::connect(addr) {
|
|
Ok(mut stream) => {
|
|
stream.set_nonblocking(false).ok();
|
|
let conn_id = SOCK_CONN_ID.fetch_add(1, Ordering::Relaxed);
|
|
SOCK_CONNS.lock().unwrap().insert(
|
|
conn_id,
|
|
SockConnState {
|
|
stream: Mutex::new(stream),
|
|
},
|
|
);
|
|
netlog!("sock:connect ok conn_id={}", conn_id);
|
|
write_tlv_handle(T_SOCK_CONN, conn_id, res, res_len)
|
|
}
|
|
Err(e) => {
|
|
netlog!("sock:connect error: {:?}", e);
|
|
E_ERR
|
|
}
|
|
}
|
|
}
|
|
_ => E_INV_METHOD,
|
|
}
|
|
}
|
|
|
|
unsafe fn sock_conn_invoke(
|
|
m: u32,
|
|
id: u32,
|
|
args: *const u8,
|
|
args_len: usize,
|
|
res: *mut u8,
|
|
res_len: *mut usize,
|
|
) -> i32 {
|
|
match m {
|
|
M_CONN_BIRTH => {
|
|
// not used directly
|
|
write_u32(0, res, res_len)
|
|
}
|
|
M_CONN_SEND => {
|
|
let bytes = tlv_parse_bytes(slice(args, args_len)).unwrap_or_default();
|
|
if let Some(conn) = SOCK_CONNS.lock().unwrap().get(&id) {
|
|
if let Ok(mut s) = conn.stream.lock() {
|
|
let _ = s.write_all(&bytes);
|
|
}
|
|
netlog!("sock:send id={} n={}", id, bytes.len());
|
|
return write_tlv_void(res, res_len);
|
|
}
|
|
E_INV_HANDLE
|
|
}
|
|
M_CONN_RECV => {
|
|
if let Some(conn) = SOCK_CONNS.lock().unwrap().get(&id) {
|
|
if let Ok(mut s) = conn.stream.lock() {
|
|
let mut buf = vec![0u8; 4096];
|
|
match s.read(&mut buf) {
|
|
Ok(n) => {
|
|
buf.truncate(n);
|
|
netlog!("sock:recv id={} n={}", id, n);
|
|
return write_tlv_bytes(&buf, res, res_len);
|
|
}
|
|
Err(_) => return write_tlv_bytes(&[], res, res_len),
|
|
}
|
|
}
|
|
}
|
|
E_INV_HANDLE
|
|
}
|
|
M_CONN_RECV_TIMEOUT => {
|
|
let timeout_ms = tlv_parse_i32(slice(args, args_len)).unwrap_or(0).max(0) as u64;
|
|
if let Some(conn) = SOCK_CONNS.lock().unwrap().get(&id) {
|
|
if let Ok(mut s) = conn.stream.lock() {
|
|
let _ = s.set_read_timeout(Some(Duration::from_millis(timeout_ms)));
|
|
let mut buf = vec![0u8; 4096];
|
|
let resv = s.read(&mut buf);
|
|
let _ = s.set_read_timeout(None);
|
|
match resv {
|
|
Ok(n) => {
|
|
buf.truncate(n);
|
|
netlog!("sock:recvTimeout id={} n={} ms={}", id, n, timeout_ms);
|
|
return write_tlv_bytes(&buf, res, res_len);
|
|
}
|
|
Err(e) => {
|
|
netlog!(
|
|
"sock:recvTimeout error id={} ms={} err={:?}",
|
|
id,
|
|
timeout_ms,
|
|
e
|
|
);
|
|
return E_ERR;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
E_INV_HANDLE
|
|
}
|
|
M_CONN_CLOSE => {
|
|
// Drop the stream by removing entry
|
|
SOCK_CONNS.lock().unwrap().remove(&id);
|
|
write_tlv_void(res, res_len)
|
|
}
|
|
_ => E_INV_METHOD,
|
|
}
|
|
}
|