summaryrefslogtreecommitdiff
path: root/crates/vcs_actions
diff options
context:
space:
mode:
author魏曹先生 <1992414357@qq.com>2025-10-13 13:34:39 +0800
committer魏曹先生 <1992414357@qq.com>2025-10-13 13:34:39 +0800
commit67fb8ec01b351c6c9fd2af321166bb92250b1218 (patch)
tree4509509ef932e1009dcd2da25928fcff71a5f29b /crates/vcs_actions
parent860fb317bca61ce66a2c98df933aa666dae0a43f (diff)
feat: Implement JSON-based type-erased action invocation
- Add process_json method to ActionPool for type-agnostic calls using JSON serialization - Extend ActionContext with action_name and action_args fields and setter methods - Update action_gen macro to use process_json instead of typed process method - Implement remote action invocation framework in client_registry and action_service - Add protocol definitions for remote action communication - Enable flexible action execution without explicit type specifications
Diffstat (limited to 'crates/vcs_actions')
-rw-r--r--crates/vcs_actions/src/actions/local_actions.rs2
-rw-r--r--crates/vcs_actions/src/connection.rs1
-rw-r--r--crates/vcs_actions/src/connection/action_service.rs29
-rw-r--r--crates/vcs_actions/src/connection/protocol.rs7
-rw-r--r--crates/vcs_actions/src/registry/client_registry.rs28
5 files changed, 56 insertions, 11 deletions
diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs
index b230c6f..0e210a7 100644
--- a/crates/vcs_actions/src/actions/local_actions.rs
+++ b/crates/vcs_actions/src/actions/local_actions.rs
@@ -6,7 +6,7 @@ use tcp_connection::error::TcpTargetError;
#[action_gen(local)]
pub async fn set_upstream_vault_action(
ctx: ActionContext,
- upstream: SocketAddr,
+ _upstream: SocketAddr,
) -> Result<(), TcpTargetError> {
if ctx.is_remote() {
return Err(TcpTargetError::NotLocal(
diff --git a/crates/vcs_actions/src/connection.rs b/crates/vcs_actions/src/connection.rs
index dabbd44..918f93c 100644
--- a/crates/vcs_actions/src/connection.rs
+++ b/crates/vcs_actions/src/connection.rs
@@ -1,2 +1,3 @@
pub mod action_service;
pub mod error;
+pub mod protocol;
diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs
index 8d3a03d..9ea5957 100644
--- a/crates/vcs_actions/src/connection/action_service.rs
+++ b/crates/vcs_actions/src/connection/action_service.rs
@@ -1,6 +1,6 @@
use std::{net::SocketAddr, path::PathBuf, sync::Arc};
-use action_system::action_pool::ActionPool;
+use action_system::{action::ActionContext, action_pool::ActionPool};
use cfg_file::config::ConfigFile;
use tcp_connection::{error::TcpTargetError, instance::ConnectionInstance};
use tokio::{
@@ -10,10 +10,12 @@ use tokio::{
};
use vcs_data::data::vault::{Vault, config::VaultConfig};
-use crate::registry::server_registry::server_action_pool;
+use crate::{
+ connection::protocol::RemoteActionInvoke, registry::server_registry::server_action_pool,
+};
// Start the server with a Vault using the specified directory
-pub async fn server_entry(path: impl Into<PathBuf>) -> Result<(), TcpTargetError> {
+pub async fn server_entry(vault_path: impl Into<PathBuf>) -> Result<(), TcpTargetError> {
// Read the vault cfg
let vault_cfg = VaultConfig::read().await?;
@@ -21,7 +23,7 @@ pub async fn server_entry(path: impl Into<PathBuf>) -> Result<(), TcpTargetError
let listener = create_tcp_listener(&vault_cfg).await?;
// Initialize the vault
- let vault: Arc<Vault> = init_vault(vault_cfg, path.into()).await?;
+ let vault: Arc<Vault> = init_vault(vault_cfg, vault_path.into()).await?;
// Create ActionPool
let action_pool: Arc<ActionPool> = Arc::new(server_action_pool());
@@ -125,5 +127,22 @@ fn build_server_future(
}
async fn process_connection(stream: TcpStream, vault: Arc<Vault>, action_pool: Arc<ActionPool>) {
- let instance = ConnectionInstance::from(stream);
+ // Setup connection instance
+ let mut instance = ConnectionInstance::from(stream);
+
+ // Read action name and action arguments
+ let Ok(msg) = instance.read_msgpack::<RemoteActionInvoke>().await else {
+ return;
+ };
+
+ // Build context
+ let ctx = ActionContext::remote().insert_instance(instance);
+
+ // Process action
+ let Ok(_result_json) = action_pool
+ .process_json(&msg.action_name, ctx, msg.action_args_json)
+ .await
+ else {
+ return;
+ };
}
diff --git a/crates/vcs_actions/src/connection/protocol.rs b/crates/vcs_actions/src/connection/protocol.rs
new file mode 100644
index 0000000..2cebe79
--- /dev/null
+++ b/crates/vcs_actions/src/connection/protocol.rs
@@ -0,0 +1,7 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Default, Clone, Serialize, Deserialize)]
+pub struct RemoteActionInvoke {
+ pub action_name: String,
+ pub action_args_json: String,
+}
diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs
index d298099..56acdad 100644
--- a/crates/vcs_actions/src/registry/client_registry.rs
+++ b/crates/vcs_actions/src/registry/client_registry.rs
@@ -1,7 +1,10 @@
use action_system::{action::ActionContext, action_pool::ActionPool};
use tcp_connection::error::TcpTargetError;
-use crate::actions::local_actions::register_set_upstream_vault_action;
+use crate::{
+ actions::local_actions::register_set_upstream_vault_action,
+ connection::protocol::RemoteActionInvoke,
+};
fn register_actions(pool: &mut ActionPool) {
// Pool register here
@@ -22,18 +25,33 @@ pub fn client_action_pool() -> ActionPool {
pool
}
-async fn on_proc_begin(ctx: &ActionContext) -> Result<(), TcpTargetError> {
+async fn on_proc_begin(ctx: &mut ActionContext) -> Result<(), TcpTargetError> {
+ // Is ctx remote
+ let is_remote = ctx.is_remote();
+
+ // Action name and arguments
+ let action_name = ctx.action_name().to_string();
+ let action_args_json = ctx.action_args_json().clone();
+
// Get instance
- let Some(_instance) = ctx.instance() else {
+ let Some(instance) = ctx.instance_mut() else {
return Err(TcpTargetError::Unsupported(
"Missing ConnectionInstance in current context, this ActionPool does not support this call"
.to_string()));
};
// If it's remote, invoke action at server
- if ctx.is_remote() {
- // instance.write_text(text)
+ if is_remote {
+ // Build protocol message
+ let msg = RemoteActionInvoke {
+ action_name: action_name,
+ action_args_json: action_args_json,
+ };
+
+ // Send
+ instance.write_msgpack(msg).await?;
}
+ // Return OK, wait for client to execute Action locally
Ok(())
}