mirror of
https://github.com/hcengineering/platform.git
synced 2026-09-10 19:57:43 +02:00
WS API: done ; HTTP API: add secret path '$'
This commit is contained in:
Leonid Kaganov
@@ -12,13 +12,34 @@ The service is exposed as REST and WebSocket API.
|
||||
- service posts a process status
|
||||
|
||||
## Key
|
||||
Key is a string that consists of one or multiple segments separated by some separator.
|
||||
Example: foo/bar/baz.
|
||||
|
||||
It is possible to use wildcard keys to list or subscribe to values with this prefix.
|
||||
Key is a string that consists of one or multiple segments separated by ‘/’. Example: foo/bar/baz.
|
||||
Key may not end with ‘/’
|
||||
Segment may not contain special characters (‘*’, ‘?’, ‘[’, ‘]’,‘\’,‘\x00..\xF1’,‘\x7F’,‘"’,‘'’)
|
||||
Segment may not be empty
|
||||
Key segment may be private (prefixed with ‘$’)
|
||||
|
||||
Query
|
||||
|
||||
May not contain special characters (‘*’, ‘?’, ‘[’, ‘]’,‘\’,‘\x00..\xF1’,‘\x7F’,‘"’,‘'’)
|
||||
It is possible to use prefix, for listings / subscriptions (prefix ends with segment separator ‘/’)
|
||||
|
||||
GET/SUBSCRIBE/.. a/b → single key
|
||||
GET/SUBSCRIBE/.. a/b/c/ → multiple
|
||||
|
||||
If multiple
|
||||
|
||||
select all keys starting with prefix
|
||||
skip keys, containing private segments to the right from the prefix
|
||||
|
||||
example
|
||||
1. /a/b/$c/$d, 2. /a/b/c, 3. /a/b/$c, 4. /a/b/$c/$d/e
|
||||
/ → [2]
|
||||
/a/b/ → [2]
|
||||
/a/b/$c/ → [3]
|
||||
/a/b/$c/$d/ → [4]
|
||||
/a/b/$c/$d → [1]
|
||||
|
||||
Key may contain a special section (guard) $that separates public and private data. “Private” data is available when querying or subscribing by exact key.
|
||||
Example foo/bar/$/private, this value can be queried by foo/bar/$/private or foo/bar/$/but not by foo/bar/
|
||||
|
||||
## Data
|
||||
“Data” is an arbitrary JSON document.
|
||||
|
||||
@@ -46,6 +46,7 @@
|
||||
border-radius: 6px;
|
||||
min-height: 100px;
|
||||
white-space: pre-wrap;
|
||||
overflow-wrap: anywhere;
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
|
||||
+123
-11
@@ -6,6 +6,129 @@ source ./pulse_lib.sh
|
||||
TOKEN=$(./token.sh claims.json)
|
||||
ZP="00000000-0000-0000-0000-000000000001/TESTS"
|
||||
|
||||
|
||||
echo "--------- if-match ----------"
|
||||
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2"
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS/1" "Value_1" "HULY-TTL: 2"
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS/2" "Value_1" "HULY-TTL: 2"
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS/3$" "Value_1" "HULY-TTL: 2"
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/4" "Value_1" "HULY-TTL: 2"
|
||||
get "00000000-0000-0000-0000-000000000001/TESTS"
|
||||
get "00000000-0000-0000-0000-000000000001/TESTS/"
|
||||
get "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/"
|
||||
|
||||
|
||||
exit
|
||||
Key
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Key is a string that consists of one or multiple segments separated by ‘/’. Example: foo/bar/baz.
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Segment may not contain special characters (‘$’, ‘*’, ‘?’)
|
||||
|
||||
|
||||
|
||||
Key may not end with ‘/’
|
||||
|
||||
|
||||
|
||||
Segment may not be empty
|
||||
|
||||
|
||||
|
||||
Key segment may be private (prefixed with ‘$’)
|
||||
|
||||
|
||||
|
||||
Query
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
May not contain special characters (‘*’, ‘?’)
|
||||
|
||||
|
||||
|
||||
It is possible to use prefix, for listings / subscriptions (prefix ends with segment separator ‘/’)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
GET/SUBSCRIBE/.. a/b → single key
|
||||
|
||||
|
||||
|
||||
GET/SUBSCRIBE/.. a/b/c/ → multiple
|
||||
|
||||
|
||||
|
||||
If multiple
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
select all keys starting with prefix
|
||||
|
||||
|
||||
|
||||
skip keys, containing private segments to the right from the prefix
|
||||
|
||||
|
||||
|
||||
example
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
1. /a/b/$c/$d, 2. /a/b/c, 3. /a/b/$c, 4. /a/b/$c/$d/e
|
||||
|
||||
|
||||
|
||||
/ → [2]
|
||||
|
||||
|
||||
|
||||
/a/b/ → [2]
|
||||
|
||||
|
||||
|
||||
/a/b/$c/ → [3]
|
||||
|
||||
|
||||
|
||||
/a/b/$c/$d/ → [4]
|
||||
|
||||
|
||||
|
||||
/a/b/$c/$d → (1)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
exit
|
||||
|
||||
echo "--------- Deprecated symbols ----------"
|
||||
|
||||
put "00000000-0000-0000-0000-000000000001/'TESTS" "Value_1" "HULY-TTL: 2"
|
||||
put "00000000-0000-0000-0000-000000000001/TES?TS" "Value_1" "HULY-TTL: 2"
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS*" "Value_1" "HULY-TTL: 2"
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS/" "Value_1" "HULY-TTL: 2"
|
||||
|
||||
echo "--------- if-match ----------"
|
||||
|
||||
delete ${ZP}
|
||||
@@ -42,17 +165,6 @@ echo "================> UPDATE PUT If-Match"
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 3"
|
||||
echo "sleep 1 sec"
|
||||
sleep 1
|
||||
|
||||
@@ -32,6 +32,9 @@ api() {
|
||||
*) echo -en "${GRAY}${status}${N}" ;;
|
||||
esac
|
||||
if [ -n "$etag" ]; then echo -n -e " ${F}${etag}${N}" ; fi
|
||||
|
||||
body=$(echo "$body" | sed 's/{/\\n{/g')
|
||||
|
||||
if [ -n "$body" ]; then echo -e "\n ${GRAY}[${body}]${N}" ; else echo -e " ${L}(no body)${N}" ; fi
|
||||
rm -f "$tmpfile"
|
||||
}
|
||||
|
||||
+13
-18
@@ -8,8 +8,6 @@ use tracing::{error, trace};
|
||||
use uuid::Uuid;
|
||||
use crate::ws_owner;
|
||||
|
||||
type ObjectPath = web::Path<(String, String)>;
|
||||
|
||||
use crate::redis::{
|
||||
Ttl, SaveMode,
|
||||
RedisArray,
|
||||
@@ -46,38 +44,35 @@ pub fn map_handler_error(err: impl std::fmt::Display) -> Error {
|
||||
|
||||
|
||||
/// list
|
||||
|
||||
// #[derive(Deserialize)]
|
||||
pub async fn list(
|
||||
req: HttpRequest,
|
||||
path: web::Path<String>,
|
||||
query: web::Query<HashMap<String, String>>,
|
||||
path: web::Path<(String, Option<String>)>,
|
||||
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
|
||||
) -> Result<HttpResponse, actix_web::Error> {
|
||||
|
||||
ws_owner::workspace_owner(&req)?; // Check workspace
|
||||
|
||||
let workspace = path.into_inner();
|
||||
let prefix = query.get("prefix").map(|s| s.as_str());
|
||||
let (workspace, key) = path.into_inner();
|
||||
|
||||
trace!(workspace, prefix, "list request");
|
||||
// trace!(workspace, prefix, "list request");
|
||||
|
||||
async move || -> anyhow::Result<HttpResponse> {
|
||||
|
||||
let mut conn = redis.lock().await;
|
||||
|
||||
let entries = redis_list(&mut *conn, &workspace, prefix).await?;
|
||||
let entries = redis_list(&mut *conn, &workspace, key.as_deref()).await?;
|
||||
|
||||
Ok(HttpResponse::Ok().json(entries))
|
||||
|
||||
}().await.map_err(map_handler_error)
|
||||
}
|
||||
|
||||
/// get / (test)
|
||||
|
||||
|
||||
/// get
|
||||
pub async fn get(
|
||||
req: HttpRequest,
|
||||
path: ObjectPath,
|
||||
path: web::Path<(String, String)>,
|
||||
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
|
||||
) -> Result<HttpResponse, actix_web::error::Error> {
|
||||
|
||||
@@ -85,7 +80,7 @@ pub async fn get(
|
||||
|
||||
let (workspace, key) = path.into_inner();
|
||||
|
||||
trace!(workspace, key, "get request");
|
||||
// trace!(workspace, key, "get request");
|
||||
|
||||
async move || -> anyhow::Result<HttpResponse> {
|
||||
|
||||
@@ -104,10 +99,9 @@ pub async fn get(
|
||||
|
||||
|
||||
/// put
|
||||
|
||||
pub async fn put(
|
||||
req: HttpRequest,
|
||||
path: ObjectPath,
|
||||
path: web::Path<(String, String)>,
|
||||
body: web::Bytes,
|
||||
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
|
||||
) -> Result<HttpResponse, actix_web::error::Error> {
|
||||
@@ -118,6 +112,8 @@ pub async fn put(
|
||||
|
||||
async move || -> anyhow::Result<HttpResponse> {
|
||||
|
||||
if !req.query_string().is_empty() { return Err(anyhow!("Query parameters are not allowed")); }
|
||||
|
||||
let mut conn = redis.lock().await;
|
||||
|
||||
// TTL logic
|
||||
@@ -151,11 +147,10 @@ pub async fn put(
|
||||
|
||||
|
||||
|
||||
// delete
|
||||
|
||||
/// delete
|
||||
pub async fn delete(
|
||||
req: HttpRequest,
|
||||
path: ObjectPath,
|
||||
path: web::Path<(String, String)>,
|
||||
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
|
||||
) -> Result<HttpResponse, actix_web::error::Error> {
|
||||
|
||||
|
||||
+271
-32
@@ -1,3 +1,27 @@
|
||||
use redis::aio::MultiplexedConnection;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
use actix::fut;
|
||||
use actix::ActorFutureExt;
|
||||
|
||||
use crate::redis::{
|
||||
Ttl, SaveMode,
|
||||
RedisArray,
|
||||
redis_save,
|
||||
redis_read,
|
||||
redis_delete,
|
||||
redis_list,
|
||||
error
|
||||
};
|
||||
|
||||
use std::future::Future;
|
||||
use serde_json::{Value, Map, json};
|
||||
|
||||
|
||||
type JsonMap = Map<String, Value>;
|
||||
|
||||
// ==================
|
||||
use actix::{Actor, StreamHandler, AsyncContext, ActorContext};
|
||||
use actix_web::{web, HttpRequest, HttpResponse, Error};
|
||||
use actix_web_actors::ws;
|
||||
@@ -10,22 +34,59 @@ use std::collections::HashSet;
|
||||
#[serde(rename_all = "lowercase", tag = "type")]
|
||||
pub enum WsCommand {
|
||||
Put {
|
||||
#[serde(default)]
|
||||
correlation: Option<String>,
|
||||
key: String,
|
||||
data: String,
|
||||
#[serde(default)]
|
||||
correlation: Option<String>,
|
||||
|
||||
#[serde(rename = "expiresAt")]
|
||||
#[serde(default)]
|
||||
expires_at: Option<u64>,
|
||||
|
||||
#[serde(rename = "TTL")]
|
||||
#[serde(default)]
|
||||
ttl: Option<u64>,
|
||||
|
||||
#[serde(rename = "ifMatch")]
|
||||
#[serde(default)]
|
||||
if_match: Option<String>,
|
||||
|
||||
#[serde(rename = "ifNoneMatch")]
|
||||
#[serde(default)]
|
||||
if_none_match: Option<String>,
|
||||
},
|
||||
Delete {
|
||||
key: String,
|
||||
|
||||
Get {
|
||||
#[serde(default)]
|
||||
correlation: Option<String>,
|
||||
},
|
||||
Sub {
|
||||
key: String,
|
||||
},
|
||||
|
||||
List {
|
||||
#[serde(default)]
|
||||
correlation: Option<String>,
|
||||
key: Option<String>,
|
||||
},
|
||||
|
||||
Delete {
|
||||
#[serde(default)]
|
||||
correlation: Option<String>,
|
||||
key: String,
|
||||
|
||||
#[serde(rename = "ifMatch")]
|
||||
#[serde(default)]
|
||||
if_match: Option<String>,
|
||||
},
|
||||
|
||||
Sub {
|
||||
#[serde(default)]
|
||||
correlation: Option<String>,
|
||||
key: String,
|
||||
},
|
||||
|
||||
Unsub {
|
||||
#[serde(default)]
|
||||
correlation: Option<String>,
|
||||
key: String,
|
||||
},
|
||||
}
|
||||
@@ -35,8 +96,10 @@ pub enum WsCommand {
|
||||
pub struct WsSession {
|
||||
pub workspace: String,
|
||||
pub subscriptions: HashSet<String>, // новые поля
|
||||
pub redis: Arc<Mutex<MultiplexedConnection>>, // вот он, тот же тип что и в HTTP API
|
||||
}
|
||||
|
||||
|
||||
/// Actor External trait: must be in separate impl block
|
||||
impl Actor for WsSession {
|
||||
type Context = ws::WebsocketContext<Self>;
|
||||
@@ -72,48 +135,224 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
|
||||
/// All logic in one impl
|
||||
impl WsSession {
|
||||
|
||||
fn wait_and_send<F>(
|
||||
&mut self,
|
||||
ctx: &mut ws::WebsocketContext<Self>,
|
||||
fut: F,
|
||||
mut base: JsonMap,
|
||||
)
|
||||
where
|
||||
F: std::future::Future<Output = Result<JsonMap, String>> + 'static,
|
||||
{
|
||||
ctx.wait(
|
||||
fut::wrap_future(fut).map(move |res, _actor: &mut Self, ctx| {
|
||||
match res {
|
||||
Ok(extra) => {
|
||||
base.extend(extra);
|
||||
}
|
||||
Err(err) => {
|
||||
base.insert("type".into(), json!("error"));
|
||||
base.insert("message".into(), json!(err));
|
||||
}
|
||||
}
|
||||
ctx.text(Value::Object(base).to_string());
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
/// When valid JSON recieved for WsSession
|
||||
fn handle_command(&mut self, cmd: WsCommand, ctx: &mut ws::WebsocketContext<Self>) {
|
||||
match cmd {
|
||||
WsCommand::Put { key, data, expires_at, correlation } => {
|
||||
println!("PUT {} = {} (expires_at: {:?})", key, data, expires_at);
|
||||
ctx.text(format!("OK PUT {}{}", key, Self::correlation_suffix(&correlation)));
|
||||
// Здесь — сохранить в Redis
|
||||
}
|
||||
WsCommand::Delete { key, correlation } => {
|
||||
|
||||
WsCommand::Put { key, data, expires_at, ttl, if_match, if_none_match, correlation } => {
|
||||
|
||||
println!("PUT {} = {} (expires_at: {:?}) (ttl: {:?}) ws={:?}", key, data, expires_at, ttl, self.workspace);
|
||||
|
||||
let redis = self.redis.clone();
|
||||
let workspace = self.workspace.clone();
|
||||
|
||||
let mut base = JsonMap::new();
|
||||
base.insert("action".into(), json!("put"));
|
||||
base.insert("workspace".into(), json!(&self.workspace));
|
||||
base.insert("key".into(), json!(&key));
|
||||
base.insert("data".into(), json!(&data));
|
||||
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
|
||||
if let Some(x) = &expires_at { base.insert("expiresAt".into(), json!(x)); }
|
||||
if let Some(x) = &ttl { base.insert("TTL".into(), json!(x)); }
|
||||
if let Some(x) = &if_match { base.insert("ifMatch".into(), json!(x)); }
|
||||
if let Some(x) = &if_none_match { base.insert("ifNoneMatch".into(),json!(x)); }
|
||||
|
||||
let fut = async move {
|
||||
|
||||
// TTL logic
|
||||
let real_ttl = if let Some(secs) = ttl {
|
||||
Some(Ttl::Sec(secs as usize))
|
||||
} else if let Some(timestamp) = expires_at {
|
||||
Some(Ttl::At(timestamp))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// MODE logic
|
||||
let mut mode = Some(SaveMode::Upsert);
|
||||
if let Some(s) = if_match { // `If-Match: *` - update only if the key exists
|
||||
if s == "*" { // `If-Match: *` — update only if exist
|
||||
mode = Some(SaveMode::Update);
|
||||
} else { // `If-Match: <md5>` — update only if current
|
||||
mode = Some(SaveMode::Equal(s.to_string()));
|
||||
}
|
||||
} else if let Some(s) = if_none_match { // `If-None-Match: *` — insert only if does not exist
|
||||
if s == "*" {
|
||||
mode = Some(SaveMode::Insert);
|
||||
} else {
|
||||
return Err::<JsonMap, String>("ifNoneMatch must contain only '*'".into());
|
||||
}
|
||||
}
|
||||
|
||||
let mut conn = redis.lock().await;
|
||||
|
||||
redis_save(&mut *conn, &workspace, &key, &data, real_ttl, mode)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut extra = JsonMap::new();
|
||||
extra.insert("response".into(), json!("OK"));
|
||||
Ok::<JsonMap, String>(extra)
|
||||
|
||||
};
|
||||
|
||||
self.wait_and_send(ctx, fut, base);
|
||||
}
|
||||
|
||||
|
||||
WsCommand::Delete { key, correlation, if_match } => {
|
||||
println!("DELETE {}", key);
|
||||
ctx.text(format!("OK DELETE {}{}", key, Self::correlation_suffix(&correlation)));
|
||||
// Здесь — удалить из Redis
|
||||
|
||||
let redis = self.redis.clone();
|
||||
let workspace = self.workspace.clone();
|
||||
|
||||
let mut base = JsonMap::new();
|
||||
base.insert("action".into(), json!("delete"));
|
||||
base.insert("workspace".into(), json!(&self.workspace));
|
||||
base.insert("key".into(), json!(&key));
|
||||
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
|
||||
if let Some(x) = &if_match { base.insert("ifMatch".into(), json!(x)); }
|
||||
|
||||
let fut = async move {
|
||||
|
||||
let mut conn = redis.lock().await;
|
||||
|
||||
let deleted = redis_delete(&mut *conn, &workspace, &key)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
if deleted {
|
||||
let mut extra = JsonMap::new();
|
||||
extra.insert("response".into(), json!("OK"));
|
||||
Ok::<JsonMap, String>(extra)
|
||||
} else {
|
||||
Err::<JsonMap, String>("not found".into())
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
self.wait_and_send(ctx, fut, base);
|
||||
}
|
||||
WsCommand::Sub { key } => {
|
||||
println!("SUB {}", key);
|
||||
|
||||
WsCommand::Get { key, correlation } => {
|
||||
println!("GET {}{:?}", key, correlation);
|
||||
|
||||
let redis = self.redis.clone();
|
||||
let workspace = self.workspace.clone();
|
||||
|
||||
let mut base = JsonMap::new();
|
||||
base.insert("action".into(), json!("get"));
|
||||
base.insert("workspace".into(), json!(&self.workspace));
|
||||
base.insert("key".into(), json!(&key));
|
||||
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
|
||||
|
||||
let fut = async move {
|
||||
|
||||
let mut conn = redis.lock().await;
|
||||
|
||||
let data_opt = redis_read(&mut *conn, &workspace, &key)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
match data_opt {
|
||||
Some(data) => {
|
||||
let mut extra = JsonMap::new();
|
||||
let data_value = serde_json::to_value(&data).map_err(|e| e.to_string())?;
|
||||
extra.insert("response".into(), data_value);
|
||||
Ok::<JsonMap, String>(extra)
|
||||
}
|
||||
None => Err::<JsonMap, String>("not found".into())
|
||||
}
|
||||
};
|
||||
|
||||
self.wait_and_send(ctx, fut, base);
|
||||
}
|
||||
|
||||
WsCommand::List { key, correlation } => {
|
||||
println!("LIST {:?}{:?}", key, correlation);
|
||||
|
||||
let redis = self.redis.clone();
|
||||
let workspace = self.workspace.clone();
|
||||
|
||||
let mut base = JsonMap::new();
|
||||
base.insert("action".into(), json!("get"));
|
||||
base.insert("workspace".into(), json!(&self.workspace));
|
||||
if let Some(x) = &key { base.insert("key".into(), json!(x)); }
|
||||
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
|
||||
|
||||
let fut = async move {
|
||||
|
||||
let mut conn = redis.lock().await;
|
||||
|
||||
let data = redis_list(&mut *conn, &workspace, key.as_deref())
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut extra = JsonMap::new();
|
||||
let data_value = serde_json::to_value(&data).map_err(|e| e.to_string())?;
|
||||
extra.insert("response".into(), data_value);
|
||||
Ok::<JsonMap, String>(extra)
|
||||
};
|
||||
|
||||
self.wait_and_send(ctx, fut, base);
|
||||
}
|
||||
|
||||
WsCommand::Sub { key, correlation } => {
|
||||
println!("SUB {}{:?}", key, correlation);
|
||||
ctx.text(format!("OK SUB {}", key));
|
||||
// Здесь — подписка (в будущем pub/sub)
|
||||
}
|
||||
WsCommand::Unsub { key } => {
|
||||
println!("UNSUB {}", key);
|
||||
|
||||
|
||||
WsCommand::Unsub { key, correlation } => {
|
||||
println!("UNSUB {}{:?}", key, correlation);
|
||||
ctx.text(format!("OK UNSUB {}", key));
|
||||
// Здесь — отписка
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
fn correlation_suffix(corr: &Option<String>) -> String {
|
||||
match corr {
|
||||
Some(c) => format!(" [correlation: {}]", c),
|
||||
None => "".to_string(),
|
||||
}
|
||||
//
|
||||
// corr.as_ref()
|
||||
// .map(|c| format!(" [correlation: {}]", c))
|
||||
// .unwrap_or_default()
|
||||
//
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
pub async fn handler(req: HttpRequest, stream: web::Payload, path: web::Path<String>) -> Result<HttpResponse, Error> {
|
||||
|
||||
pub async fn handler(
|
||||
req: HttpRequest,
|
||||
stream: web::Payload,
|
||||
path: web::Path<String>,
|
||||
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
|
||||
) -> Result<HttpResponse, Error> {
|
||||
let workspace = path.into_inner();
|
||||
let session = WsSession { workspace, subscriptions: HashSet::new() };
|
||||
let session = WsSession {
|
||||
workspace,
|
||||
subscriptions: HashSet::new(),
|
||||
redis: redis.get_ref().clone(),
|
||||
};
|
||||
ws::start(session, &req, stream)
|
||||
}
|
||||
|
||||
+5
-4
@@ -105,10 +105,11 @@ async fn main() -> anyhow::Result<()> {
|
||||
.service(
|
||||
web::scope("/api")
|
||||
.wrap(middleware::from_fn(interceptor))
|
||||
.route("/{workspace}", web::get().to(handlers_http::list))
|
||||
.route("/{workspace}/{key:.*}",web::get().to(handlers_http::get))
|
||||
.route("/{workspace}/{key:.*}",web::put().to(handlers_http::put))
|
||||
.route("/{workspace}/{key:.*}",web::delete().to(handlers_http::delete))
|
||||
.route("/{workspace}/", web::get().to(handlers_http::list))
|
||||
.route("/{workspace}/{key:.+/}", web::get().to(handlers_http::list))
|
||||
.route("/{workspace}/{key:.+}", web::get().to(handlers_http::get))
|
||||
.route("/{workspace}/{key:.+}", web::put().to(handlers_http::put))
|
||||
.route("/{workspace}/{key:.+}", web::delete().to(handlers_http::delete))
|
||||
)
|
||||
.route("/status", web::get().to(async || "ok"))
|
||||
.route("/ws/{workspace}", web::get().to(handlers_ws::handler)) // WebSocket
|
||||
|
||||
+58
-18
@@ -2,6 +2,7 @@ use crate::config::{CONFIG, RedisMode};
|
||||
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
#[derive(serde::Serialize)]
|
||||
pub enum Ttl {
|
||||
Sec(usize), // EX
|
||||
At(u64), // EXAT (timestamp in seconds)
|
||||
@@ -32,51 +33,76 @@ pub struct RedisArray {
|
||||
pub etag: String, // md5 hash (data)
|
||||
}
|
||||
|
||||
fn error<T>(code: u16, msg: impl Into<String>) -> redis::RedisResult<T> {
|
||||
/// return Error
|
||||
pub fn error<T>(code: u16, msg: impl Into<String>) -> redis::RedisResult<T> {
|
||||
let msg = msg.into();
|
||||
let full = format!("{}: {}", code, msg);
|
||||
Err(redis::RedisError::from(( redis::ErrorKind::ExtensionError, "", full )))
|
||||
}
|
||||
|
||||
|
||||
/// Check for redis-deprecated symbols
|
||||
pub fn redis_deprecate_symbols(s: &str) -> redis::RedisResult<()> {
|
||||
if s.chars().any(|c| matches!( c,
|
||||
'*' | '?' | '[' | ']' | '\\' |
|
||||
'\0'..='\x1F' | '\x7F' |
|
||||
'"' | '\'' // | ' '
|
||||
)) {
|
||||
error(412, "Deprecated symbols in workspace or key")
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/// redis_list(&connection,workspace,prefix)
|
||||
pub async fn redis_list(
|
||||
conn: &mut MultiplexedConnection,
|
||||
workspace: &str,
|
||||
prefix: Option<&str>,
|
||||
key: Option<&str>,
|
||||
) -> redis::RedisResult<Vec<RedisArray>> {
|
||||
|
||||
let pattern = if let Some(k) = key {
|
||||
if !k.ends_with('/') { return error(412, "Key must end with slash"); }
|
||||
Some(format!("{k}*"))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
redis_deprecate_symbols(&workspace)?;
|
||||
if let Some(k) = key { redis_deprecate_symbols(k)?; }
|
||||
|
||||
let mut cursor = 0;
|
||||
let mut results = Vec::new();
|
||||
let pattern = prefix.map(|p| format!("{}*", p));
|
||||
|
||||
loop {
|
||||
let mut cmd = redis::cmd("HSCAN");
|
||||
cmd.arg(workspace).arg(cursor);
|
||||
if let Some(ref p) = pattern {
|
||||
cmd.arg("MATCH").arg(p);
|
||||
}
|
||||
cmd.arg("MATCH").arg(p);
|
||||
}
|
||||
|
||||
// cmd.arg("COUNT").arg(100);
|
||||
|
||||
let (next_cursor, items): (u64, Vec<(String, String)>) = cmd.query_async(conn).await?;
|
||||
|
||||
for (key, value) in items {
|
||||
for (k, v) in items {
|
||||
|
||||
// Check for $-security path
|
||||
if let Some(prefix) = key {
|
||||
if k[prefix.len()..].contains('$') { continue; }
|
||||
}
|
||||
|
||||
// TTL
|
||||
let ttl_vec: Vec<i64> = redis::cmd("HTTL")
|
||||
.arg(workspace)
|
||||
.arg("FIELDS")
|
||||
.arg(1)
|
||||
.arg(&key)
|
||||
.query_async(conn)
|
||||
.await?;
|
||||
|
||||
let ttl_vec: Vec<i64> = redis::cmd("HTTL").arg(workspace).arg("FIELDS").arg(1).arg(&k).query_async(conn).await?;
|
||||
let ttl = ttl_vec.get(0).copied().unwrap_or(-3);
|
||||
|
||||
if ttl >= 0 {
|
||||
results.push(RedisArray {
|
||||
workspace: workspace.to_string(),
|
||||
key,
|
||||
data: value.clone(),
|
||||
key: k,
|
||||
data: v.clone(),
|
||||
expires_at: ttl as u64,
|
||||
etag: hex::encode(md5::compute(&value).0),
|
||||
etag: hex::encode(md5::compute(&v).0),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -97,6 +123,10 @@ pub async fn redis_read(
|
||||
key: &str,
|
||||
) -> redis::RedisResult<Option<RedisArray>> {
|
||||
|
||||
redis_deprecate_symbols(&workspace)?;
|
||||
redis_deprecate_symbols(&key)?;
|
||||
if key.ends_with('/') { return error(412, "Key must not end with a slash"); }
|
||||
|
||||
let data: Option<String> = redis::cmd("HGET").arg(workspace).arg(key).query_async(conn).await?;
|
||||
let Some(data) = data else { return Ok(None); };
|
||||
|
||||
@@ -137,6 +167,10 @@ pub async fn redis_save<T: ToRedisArgs>(
|
||||
mode: Option<SaveMode>,
|
||||
) -> RedisResult<()> {
|
||||
|
||||
redis_deprecate_symbols(&workspace)?;
|
||||
redis_deprecate_symbols(&key)?;
|
||||
if key.ends_with('/') { return error(412, "Key must not end with a slash"); }
|
||||
|
||||
// TTL logic
|
||||
let sec = match ttl {
|
||||
Some(Ttl::Sec(secs)) => secs,
|
||||
@@ -201,6 +235,10 @@ pub async fn redis_delete(
|
||||
key: &str,
|
||||
) -> redis::RedisResult<bool> {
|
||||
|
||||
redis_deprecate_symbols(&workspace)?;
|
||||
redis_deprecate_symbols(&key)?;
|
||||
if key.ends_with('/') { return error(412, "Key must not end with a slash"); }
|
||||
|
||||
let deleted: i32 = redis::cmd("HDEL")
|
||||
.arg(workspace)
|
||||
.arg(key)
|
||||
@@ -268,3 +306,5 @@ pub async fn redis_connect() -> anyhow::Result<MultiplexedConnection> {
|
||||
|
||||
Ok(conn)
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user