use serde_json::Value as JsonValue;
pub use crate::broker::RequestContext;
pub use crate::proto::RequestContext as ProtoRequestContext;
pub use prost_types::{Struct, Value as ProstValue};
use crate::runtime::executor_utils as eu;
pub fn struct_to_json(value: &Struct) -> JsonValue {
eu::struct_to_json(value)
}
pub fn prost_value_to_json(value: &ProstValue) -> JsonValue {
eu::prost_value_to_json(value)
}
pub fn json_to_struct(value: &JsonValue) -> Option<Struct> {
eu::json_to_struct(value)
}
pub fn json_to_prost_value(value: &JsonValue) -> Option<ProstValue> {
eu::json_to_prost_value(value)
}
pub fn json_into_struct(value: JsonValue) -> Option<Struct> {
eu::json_into_struct(value)
}
pub fn merge_context(
proto_context: Option<&ProtoRequestContext>,
metadata_context: RequestContext,
) -> RequestContext {
eu::merge_context(proto_context, metadata_context)
}
pub fn parse_sql_dispatch(request_json: &str) -> Result<(String, Vec<JsonValue>), tonic::Status> {
eu::parse_sql_dispatch(request_json)
}
#[cfg(any(feature = "gcs", feature = "azureblob"))]
pub fn parse_object_dispatch(
req: &str,
bucket_keys: &[&str],
object_keys: &[&str],
bucket_label: &str,
) -> Result<(String, String, String, Option<String>), tonic::Status> {
eu::parse_object_dispatch(req, bucket_keys, object_keys, bucket_label)
}
#[cfg(any(feature = "pinecone", feature = "weaviate", feature = "elasticsearch"))]
pub fn parse_rest_dispatch(
req: &str,
) -> Result<(reqwest::Method, String, JsonValue), tonic::Status> {
eu::parse_rest_dispatch(req)
}
pub fn object_bytes_from_json(value: &JsonValue) -> Result<Vec<u8>, tonic::Status> {
eu::object_bytes_from_json(value)
}
#[cfg(any(feature = "mysql", feature = "sqlite", feature = "mssql"))]
pub fn base64_cell(bytes: &[u8]) -> JsonValue {
eu::base64_cell(bytes)
}
#[cfg(feature = "simd-codecs")]
pub mod codec_bench {
pub fn scalar_encode(data: &[u8]) -> String {
crate::runtime::accel::scalar::base64_encode(data)
}
pub fn simd_encode(data: &[u8]) -> String {
base64_simd::STANDARD.encode_to_string(data)
}
pub fn scalar_decode(s: &str) -> Vec<u8> {
crate::runtime::accel::scalar::base64_decode(s).expect("valid base64")
}
pub fn simd_decode(s: &str) -> Vec<u8> {
base64_simd::STANDARD
.decode_to_vec(s.as_bytes())
.expect("valid base64")
}
}
#[cfg(feature = "simd-checksum")]
pub mod checksum_bench {
pub fn scalar_crc32(data: &[u8]) -> u32 {
crate::runtime::accel::scalar::crc32(data)
}
pub fn simd_crc32(data: &[u8]) -> u32 {
crc32fast::hash(data)
}
}
pub enum SqlBench {
#[cfg(feature = "postgres")]
Postgres(crate::runtime::executors::postgres::PostgresExecutor),
#[cfg(feature = "mysql")]
MySql(crate::runtime::executors::mysql::MysqlExecutor),
#[cfg(feature = "mssql")]
Mssql(crate::runtime::executors::mssql::MssqlClient),
}
impl SqlBench {
#[cfg(feature = "postgres")]
pub async fn connect_postgres(dsn: &str) -> Result<Self, String> {
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(8)
.connect(dsn)
.await
.map_err(|e| e.to_string())?;
Ok(Self::Postgres(
crate::runtime::executors::postgres::PostgresExecutor::with_pool(pool),
))
}
#[cfg(feature = "mysql")]
pub async fn connect_mysql(dsn: &str) -> Result<Self, String> {
let pool = sqlx::mysql::MySqlPoolOptions::new()
.max_connections(8)
.connect(dsn)
.await
.map_err(|e| e.to_string())?;
Ok(Self::MySql(
crate::runtime::executors::mysql::MysqlExecutor::with_pool(pool),
))
}
#[cfg(feature = "mssql")]
pub fn connect_mssql(ado_connection_string: &str) -> Self {
Self::Mssql(crate::runtime::executors::mssql::MssqlClient::new(
ado_connection_string,
))
}
pub async fn query(&self, request_json: &str) -> Result<String, String> {
use crate::runtime::executors::QueryExecutor as _;
let r = match self {
#[cfg(feature = "postgres")]
Self::Postgres(e) => e.query(request_json).await,
#[cfg(feature = "mysql")]
Self::MySql(e) => e.query(request_json).await,
#[cfg(feature = "mssql")]
Self::Mssql(c) => {
crate::runtime::executors::mssql::MssqlExecutor::new(c.clone())
.query(request_json)
.await
}
};
r.map_err(|s| s.to_string())
}
pub async fn mutate(&self, request_json: &str) -> Result<String, String> {
use crate::runtime::executors::MutationExecutor as _;
let r = match self {
#[cfg(feature = "postgres")]
Self::Postgres(e) => e.mutate(request_json).await,
#[cfg(feature = "mysql")]
Self::MySql(e) => e.mutate(request_json).await,
#[cfg(feature = "mssql")]
Self::Mssql(c) => {
crate::runtime::executors::mssql::MssqlExecutor::new(c.clone())
.mutate(request_json)
.await
}
};
r.map_err(|s| s.to_string())
}
pub async fn raw_exec(&self, sql: &str) -> Result<(), String> {
match self {
#[cfg(feature = "postgres")]
Self::Postgres(e) => sqlx::query(sql)
.execute(&e.pool)
.await
.map(|_| ())
.map_err(|x| x.to_string()),
#[cfg(feature = "mysql")]
Self::MySql(e) => sqlx::query(sql)
.execute(&e.pool)
.await
.map(|_| ())
.map_err(|x| x.to_string()),
#[cfg(feature = "mssql")]
Self::Mssql(c) => c.simple_batch(sql).await,
}
}
}
#[cfg(feature = "s3")]
pub struct S3Bench(crate::runtime::executors::s3::S3Executor);
#[cfg(feature = "s3")]
impl S3Bench {
pub fn connect(endpoint: &str, access: &str, secret: &str, region: &str) -> Self {
use aws_sdk_s3::config::{Credentials, Region};
let creds = Credentials::new(access, secret, None, None, "udb-bench");
let conf = aws_sdk_s3::Config::builder()
.behavior_version(aws_config::BehaviorVersion::latest())
.credentials_provider(creds)
.region(Region::new(region.to_string()))
.endpoint_url(endpoint)
.force_path_style(true)
.build();
Self(crate::runtime::executors::s3::S3Executor(
aws_sdk_s3::Client::from_conf(conf),
))
}
pub async fn ensure_bucket(&self, bucket: &str) -> Result<(), String> {
use crate::runtime::executors::ResourceAdminExecutor as _;
self.0
.ensure_resource(bucket, "{}")
.await
.map_err(|s| s.to_string())
}
pub async fn put(&self, request_json: &str, bytes: Vec<u8>) -> Result<String, String> {
use crate::runtime::executors::ObjectExecutor as _;
self.0
.put_object(request_json, bytes)
.await
.map_err(|s| s.to_string())
}
pub async fn get(&self, request_json: &str) -> Result<Vec<u8>, String> {
use crate::runtime::executors::ObjectExecutor as _;
self.0
.get_object(request_json)
.await
.map_err(|s| s.to_string())
}
pub async fn delete(&self, request_json: &str) -> Result<(), String> {
use crate::runtime::executors::ObjectExecutor as _;
self.0
.delete_object(request_json)
.await
.map_err(|s| s.to_string())
}
}