xiaoxue_admin
/
dbx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(mongo): fallback to agent driver for pre-4.2 MongoDB 

When native Rust MongoDB driver fails with wire version error,
automatically fall back to the Java agent driver which supports
older MongoDB versions. All mongo_ops route through agent RPC
when the connection pool is Agent-backed.
This commit is contained in:
t8y2 2026-05-14 15:04:02 +08:00
parent 88ad579a27
commit a0bcb1d889
4 changed files with 84 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
crates/dbx-core/src/connection.rs
View File

@ -171,11 +171,20 @@ impl AppState {
let con = db::duckdb_driver::connect_path(&expand_tilde(&db_config.host))?;
PoolKind::DuckDb(con)
}
DatabaseType::MongoDb => {
let client = db::mongo_driver::connect(&url).await?;
db::mongo_driver::test_connection(&client).await?;
PoolKind::MongoDb(client)
}
DatabaseType::MongoDb => match db::mongo_driver::connect(&url).await {
Ok(client) => {
db::mongo_driver::test_connection(&client).await?;
PoolKind::MongoDb(client)
}
Err(e) if e.contains("wire version") => {
log::info!("Native MongoDB driver failed ({e}), falling back to agent driver");
let connect_params = serde_json::json!({ "connection": agent_connect_params(&db_config, &host, port, db_config.effective_database().unwrap_or("")) });
let mut client = self.agent_manager.spawn(&DatabaseType::MongoDb, None).await?;
client.call::<serde_json::Value>("connect", connect_params).await?;
PoolKind::Agent(Arc::new(tokio::sync::Mutex::new(client)))
}
Err(e) => return Err(e),
},
DatabaseType::ClickHouse => {
let username = if db_config.username.is_empty() { None } else { Some(db_config.username.clone()) };
let password = if db_config.password.is_empty() { None } else { Some(db_config.password.clone()) };

1
crates/dbx-core/src/database_capabilities.rs
View File

@ -22,6 +22,7 @@ pub fn agent_key(db_type: &DatabaseType, driver_profile: Option<&str>) -> Option
DatabaseType::Kylin => Some("kylin"),
DatabaseType::Sundb => Some("sundb"),
DatabaseType::Gaussdb => Some("gaussdb"),
DatabaseType::MongoDb => Some("mongodb"),
_ => None,
}
}

68
crates/dbx-core/src/mongo_ops.rs
View File

@ -7,6 +7,11 @@ pub async fn mongo_list_databases_core(state: &AppState, connection_id: &str) ->
match connections.get(connection_id).ok_or("Not found")? {
PoolKind::MongoDb(client) => mongo_driver::list_databases(client).await,
PoolKind::Elasticsearch(_) => Ok(vec!["default".to_string()]),
PoolKind::Agent(client) => {
let mut client = client.lock().await;
let result: Vec<serde_json::Value> = client.call("list_databases", serde_json::json!({})).await?;
Ok(result.iter().filter_map(|v| v.get("name")?.as_str().map(String::from)).collect())
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
}
}
@ -20,6 +25,10 @@ pub async fn mongo_list_collections_core(
match connections.get(connection_id).ok_or("Not found")? {
PoolKind::MongoDb(client) => mongo_driver::list_collections(client, database).await,
PoolKind::Elasticsearch(client) => elasticsearch_driver::list_indices(client).await,
PoolKind::Agent(client) => {
let mut client = client.lock().await;
client.call("list_collections", serde_json::json!({ "database": database })).await
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
}
}
@ -44,6 +53,22 @@ pub async fn mongo_find_documents_core(
drop(connections);
elasticsearch_driver::find_documents(&client, collection, skip, limit).await
}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
client
.call(
"find_documents",
serde_json::json!({
"database": database,
"collection": collection,
"skip": skip,
"limit": limit,
"filter": filter,
"sort": sort,
}),
)
.await
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
}
}
@ -63,6 +88,20 @@ pub async fn mongo_insert_document_core(
drop(connections);
elasticsearch_driver::insert_document(&client, collection, doc_json).await
}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
let result: serde_json::Value = client
.call(
"insert_document",
serde_json::json!({
"database": database,
"collection": collection,
"doc_json": doc_json,
}),
)
.await?;
Ok(result.get("inserted_id").and_then(|v| v.as_str()).unwrap_or("").to_string())
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
}
}
@ -83,6 +122,21 @@ pub async fn mongo_update_document_core(
drop(connections);
elasticsearch_driver::update_document(&client, collection, id, doc_json).await
}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
let result: serde_json::Value = client
.call(
"update_document",
serde_json::json!({
"database": database,
"collection": collection,
"id": id,
"doc_json": doc_json,
}),
)
.await?;
Ok(result.get("modified_count").and_then(|v| v.as_u64()).unwrap_or(0))
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
}
}
@ -102,6 +156,20 @@ pub async fn mongo_delete_document_core(
drop(connections);
elasticsearch_driver::delete_document(&client, collection, id).await
}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
let result: serde_json::Value = client
.call(
"delete_document",
serde_json::json!({
"database": database,
"collection": collection,
"id": id,
}),
)
.await?;
Ok(result.get("deleted_count").and_then(|v| v.as_u64()).unwrap_or(0))
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
}
}

1
src-tauri/src/commands/agents.rs
View File

@ -32,6 +32,7 @@ const AGENT_TYPES: &[(&str, &str)] = &[
("kylin", "Apache Kylin"),
("sundb", "SunDB"),
("gaussdb", "GaussDB"),
("mongodb", "MongoDB (Legacy)"),
];
fn build_agent_list(am: &AgentManager, registry: Option<&AgentRegistry>) -> Vec<AgentDriverInfo> {