fluo10/caretta-sync
1
0
Fork 0
Code Issues 10 Pull requests Projects Releases Packages Wiki Activity Actions

#16 Enable RPC to listen http:127.0.0.1 for for Windows and iOS sim

Browse source
This commit is contained in:
fluo10 2025-08-30 10:30:25 +09:00
parent 312b91fedb
commit 069ce417df
10 changed files with 62 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.toml
View file

@ -50,6 +50,7 @@ serde = { version = "1.0.219", features = ["derive"] }
thiserror = "2.0.12"
tokio = { version = "1.45.0", features = ["macros", "rt", "rt-multi-thread"] }
tonic = "0.14.0"
url = { version = "2.5.7", features = ["serde"] }
uuid = { version = "1.17.0", features = ["v7"] }
[profile.dev]

4
bevy/src/peer.rs
View file

@ -17,8 +17,8 @@ pub struct PeerAddress(String);
#[tokio::main]
async fn add_cached_peers(mut commands: Commands) {
let config = CONFIG.get_unchecked();
let path = String::from("unix://") + config.rpc.socket_path.as_os_str().to_str().expect("Invalid string");
let mut client = caretta_sync_core::proto::cached_peer_service_client::CachedPeerServiceClient::connect(path).await.expect("Unix socket should be accessible");
let url = config.rpc.endpoint_url.to_string();
let mut client = caretta_sync_core::proto::cached_peer_service_client::CachedPeerServiceClient::connect(url).await.expect("Unix socket should be accessible");
let request = tonic::Request::new(CachedPeerListRequest {});
let response = client.list(request).await.expect("Faild to request/response");
let peers = response.into_inner().peers;

4
cli/src/cli/peer/list.rs
View file

@ -15,8 +15,8 @@ impl Runnable for PeerListCommandArgs {
#[tokio::main]
async fn run(self, app_name: &'static str) {
let config = self.config.into_config(app_name).await;
let path = String::from("unix://") + config.rpc.socket_path.as_os_str().to_str().expect("Invalid string");
let mut client = caretta_sync_core::proto::cached_peer_service_client::CachedPeerServiceClient::connect(path).await.expect("Unix socket should be accessible");
let url = config.rpc.endpoint_url.to_string();
let mut client = caretta_sync_core::proto::cached_peer_service_client::CachedPeerServiceClient::connect(url).await.expect("Target endpoint should be accessible");
let request = tonic::Request::new(CachedPeerListRequest {});
let response = client.list(request).await.expect("Faild to request/response");
println!("{:?}", response);

3
core/Cargo.toml
View file

@ -23,6 +23,7 @@ libp2p.workspace = true
libp2p-core = { version = "0.43.0", features = ["serde"] }
libp2p-identity = { version = "0.2.11", features = ["ed25519", "peerid", "rand", "serde"] }
prost = "0.14.1"
prost-types = "0.14.1"
sea-orm.workspace = true
sea-orm-migration.workspace = true
serde.workspace = true
@ -35,7 +36,7 @@ tonic-prost = "0.14.0"
tracing = "0.1.41"
tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
uuid.workspace = true
prost-types = "0.14.1"
url.workspace = true
sysinfo = "0.37.0"
whoami = "1.6.1"

6
core/src/config/error.rs
View file

@ -1,3 +1,5 @@
use url::Url;
#[derive(thiserror::Error, Debug)]
pub enum ConfigError {
#[error("missing config: {0}")]
@ -7,5 +9,7 @@ pub enum ConfigError {
#[error("Toml Deserialization Error")]
TomlDerialization(#[from] toml::de::Error),
#[error("Toml Serialization Error")]
TomlSerialization(#[from] toml::ser::Error),
TomlSerialization(#[from] toml::ser::Error),
#[error("Invalid url: {0}")]
InvalidUrl(Url)
}

29
core/src/config/rpc.rs
View file

@ -1,25 +1,28 @@
use std::{net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener}, path::PathBuf, str::FromStr};
#[cfg(feature="cli")]
use clap::Args;
use url::Url;
use crate::{config::PartialConfig, utils::{emptiable::Emptiable, mergeable::Mergeable}};
use libp2p::mdns::Config;
use serde::{Deserialize, Serialize};
use crate::config::error::ConfigError;
#[cfg(unix)]
static DEFAULT_SOCKET_PATH: &str = "caretta.sock";
static DEFAULT_PORT: u16 = 54321;
#[derive(Clone, Debug)]
pub struct RpcConfig {
pub socket_path: PathBuf,
pub endpoint_url: Url,
}
impl TryFrom<PartialRpcConfig> for RpcConfig {
type Error = ConfigError;
fn try_from(config: PartialRpcConfig) -> Result<Self, Self::Error> {
Ok(Self{
socket_path: config.socket_path.ok_or(ConfigError::MissingConfig("port".to_string()))?,
endpoint_url: config.endpoint_url.ok_or(ConfigError::MissingConfig("endpoint".to_string()))?,
})
}
}
@ -27,21 +30,21 @@ impl TryFrom<PartialRpcConfig> for RpcConfig {
#[cfg_attr(feature="cli", derive(Args))]
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
pub struct PartialRpcConfig {
pub socket_path: Option<PathBuf>,
pub endpoint_url: Option<Url>,
}
impl PartialRpcConfig {
#[cfg(not(target_os="ios"))]
#[cfg(not(any(all(target_os="ios", target_abi="sim"), target_os="windows")))]
pub fn default(app_name: &'static str) -> Self {
let username = whoami::username();
Self{
socket_path: Some(std::env::temp_dir().join(username).join(String::from(app_name) + ".sock")),
endpoint_url: Some(Url::parse(&(String::from("unix://") + std::env::temp_dir().join(username).join(String::from(app_name) + ".sock").to_str().unwrap())).unwrap()),
}
}
#[cfg(target_os="ios")]
#[cfg(any(all(target_os="ios", target_abi="sim"), target_os="windows"))]
pub fn default(app_name: &'static str) -> Self {
Self{
socket_path: Some(std::env::temp_dir().join(String::from(app_name) + ".sock")),
endpoint_url: Some(Url::parse("http://127.0.0.1:54321").unwrap()),
}
}
}
@ -49,26 +52,26 @@ impl PartialRpcConfig {
impl Emptiable for PartialRpcConfig {
fn empty() -> Self {
Self {
socket_path: None,
endpoint_url: None,
}
}
fn is_empty(&self) -> bool {
self.socket_path.is_none()
self.endpoint_url.is_none()
}
}
impl From<RpcConfig> for PartialRpcConfig {
fn from(source: RpcConfig) -> Self {
Self {
socket_path: Some(source.socket_path),
endpoint_url: Some(source.endpoint_url),
}
}
}
impl Mergeable for PartialRpcConfig {
fn merge(&mut self, other: Self) {
if let Some(x) = other.socket_path {
self.socket_path = Some(x);
if let Some(x) = other.endpoint_url {
self.endpoint_url = Some(x);
}
}
}

3
core/src/tests.rs
View file

@ -3,6 +3,7 @@ use std::{path::PathBuf, sync::LazyLock};
use sea_orm::{sea_query::{FromValueTuple, IntoValueTuple, ValueType}, ActiveModelBehavior, ActiveModelTrait, ColumnTrait, Condition, DatabaseConnection, EntityTrait, IntoActiveModel, ModelTrait, PrimaryKeyToColumn, PrimaryKeyTrait, Value};
use sea_orm::QueryFilter;
use tempfile::TempDir;
use url::Url;
use crate::{ config::{Config, PartialConfig, PartialP2pConfig, PartialRpcConfig, RpcConfig, StorageConfig}, message::Message};
use serde::{de::DeserializeOwned, Deserialize, Serialize};
@ -20,7 +21,7 @@ pub static TEST_CONFIG: LazyLock<Config> = LazyLock::new(|| {
cache_directory: cache_dir,
},
rpc: RpcConfig{
socket_path: test_dir.join("socket.sock"),
endpoint_url: Url::parse(&(String::from("unix://") + test_dir.join("socket.sock").to_str().unwrap())).unwrap(),
},
}
});

1
examples/core/Cargo.toml
View file

@ -13,3 +13,4 @@ libp2p.workspace = true
tokio.workspace = true
tokio-stream = { version = "0.1.17", features = ["net"] }
tonic.workspace = true
url.workspace = true

42
examples/core/src/server.rs
View file

@ -1,3 +1,5 @@
use std::path::PathBuf;
use caretta_sync::{
config::P2pConfig,
proto::cached_peer_service_server::CachedPeerServiceServer,
@ -43,21 +45,35 @@ impl ServerTrait for Server {
async fn serve_rpc<T>(config: &T) -> Result<(), caretta_sync::error::Error>
where T: AsRef<caretta_sync::config::RpcConfig> {
let path = config.as_ref().socket_path.clone();
if let Some(x) = path.parent() {
if !x.exists() {
std::fs::create_dir_all(x).expect("Failed to create directory for socket file!");
let url = config.as_ref().endpoint_url.clone();
let router = tonic::transport::Server::builder()
.add_service(CachedPeerServiceServer::new(CachedPeerService::default()));
match url.scheme() {
"unix" => {
let path = PathBuf::from(url.path());
if let Some(x) = path.parent() {
if !x.exists() {
std::fs::create_dir_all(x).expect("Failed to create directory for socket file!");
}
}
if path.exists() {
std::fs::remove_file(&path).expect("Failed to remove existing socket file!")
}
let uds = UnixListener::bind(path).unwrap();
let uds_stream = UnixListenerStream::new(uds);
router.serve_with_incoming(uds_stream)
.await.unwrap();
},
"http" => {
let host = url.socket_addrs(|| None).expect("http endpoint should have host address and port").pop().unwrap();
router.serve(host).await.unwrap();
},
_ => {
Err(caretta_sync::error::Error::Config(caretta_sync::config::ConfigError::InvalidUrl(url)))?;
}
}
if path.exists() {
std::fs::remove_file(&path).expect("Failed to remove existing socket file!")
}
let uds = UnixListener::bind(path).unwrap();
let uds_stream = UnixListenerStream::new(uds);
tonic::transport::Server::builder()
.add_service(CachedPeerServiceServer::new(CachedPeerService::default()))
.serve_with_incoming(uds_stream)
.await.unwrap();
Ok(())
}
}

6
examples/mobile/src/lib.rs
View file

@ -30,16 +30,14 @@ pub async fn init_config() {
let mut default = PartialConfig::default(APP_NAME);
default.merge(config);
let config2 : Config = default.try_into().unwrap();
Server::serve_all(&config2).await;
Server::serve_all(&config2);
}
#[bevy_main]
pub fn main() {
//init_config();
init_config();
let mut app = App::new();
app.add_plugins(