Format task payload as JSON, when appropriate#15
Open
drauschenbach wants to merge 2 commits intoemo-crab:mainfrom
Open
Format task payload as JSON, when appropriate#15drauschenbach wants to merge 2 commits intoemo-crab:mainfrom
drauschenbach wants to merge 2 commits intoemo-crab:mainfrom
Conversation
Member
|
Thank you for your contribution. I understand what you mean. You would like to debug and print different formats based on different payload types. #[derive(Clone)]
pub enum Payload {
Bytes(Vec<u8>),
#[cfg(feature = "json")]
Json(Vec<u8>),
}
diff --git a/asynq/src/backend/pgdb/postgres_broker.rs b/asynq/src/backend/pgdb/postgres_broker.rs
index 4e6b73d..1b00dcf 100644
--- a/asynq/src/backend/pgdb/postgres_broker.rs
+++ b/asynq/src/backend/pgdb/postgres_broker.rs
@@ -208,7 +208,7 @@ impl PostgresBroker {
pub(crate) fn task_to_message(&self, task: &Task) -> TaskMessage {
TaskMessage {
r#type: task.task_type.clone(),
- payload: task.payload.clone(),
+ payload: task.payload.to_vec(),
headers: task.headers.clone(),
id: task
.options
diff --git a/asynq/src/backend/rdb/redis_broker.rs b/asynq/src/backend/rdb/redis_broker.rs
index 081bed0..b7af587 100644
--- a/asynq/src/backend/rdb/redis_broker.rs
+++ b/asynq/src/backend/rdb/redis_broker.rs
@@ -237,7 +237,7 @@ impl RedisBroker {
pub(crate) fn task_to_message(&self, task: &Task) -> TaskMessage {
TaskMessage {
r#type: task.task_type.clone(),
- payload: task.payload.clone(),
+ payload: task.payload.to_vec(),
headers: task.headers.clone(),
id: task
.options
diff --git a/asynq/src/backend/wsdb/ws_broker.rs b/asynq/src/backend/wsdb/ws_broker.rs
index dc9ca4d..0ae02d3 100644
--- a/asynq/src/backend/wsdb/ws_broker.rs
+++ b/asynq/src/backend/wsdb/ws_broker.rs
@@ -2,6 +2,7 @@
//!
//! A broker that connects to an asynq-server via WebSocket for cross-process communication.
+use std::ops::Deref;
use crate::backend::wsdb::message::{
ClientMessage, EnqueueRequest, ServerMessage, TaskDoneRequest, TaskMessageResponse,
};
@@ -219,7 +220,7 @@ impl WebSocketBroker {
pub(crate) fn task_to_enqueue_request(&self, task: &Task) -> EnqueueRequest {
EnqueueRequest {
task_type: task.task_type.clone(),
- payload: BASE64_STANDARD.encode(&task.payload),
+ payload: BASE64_STANDARD.encode(task.payload.deref()),
headers: task.headers.clone(),
queue: if task.options.queue.is_empty() {
None
diff --git a/asynq/src/components/aggregator.rs b/asynq/src/components/aggregator.rs
index 463e3a7..dab5df4 100644
--- a/asynq/src/components/aggregator.rs
+++ b/asynq/src/components/aggregator.rs
@@ -240,12 +240,11 @@ impl Aggregator {
// Convert TaskMessage to Task
let mut tasks = Vec::new();
for task_msg in task_messages {
- match Task::new_with_headers(
- &task_msg.r#type,
- &task_msg.payload,
- task_msg.headers,
- ) {
- Ok(task) => tasks.push(task),
+ match Task::new(&task_msg.r#type, &task_msg.payload) {
+ Ok(mut task) => {
+ task = task.with_headers(task_msg.headers);
+ tasks.push(task)
+ }
Err(e) => {
tracing::warn!("Aggregator: failed to create task from message: {}", e);
}
diff --git a/asynq/src/components/processor.rs b/asynq/src/components/processor.rs
index 347ce1f..aa65b78 100644
--- a/asynq/src/components/processor.rs
+++ b/asynq/src/components/processor.rs
@@ -378,12 +378,11 @@ impl Processor {
// 创建任务，包含头部信息
// Create task with headers to preserve workflow path and other metadata
- let mut task = match Task::new_with_headers(
- &task_msg.r#type,
- &task_msg.payload,
- task_msg.headers.clone(),
- ) {
- Ok(task) => task,
+ let mut task = match Task::new(&task_msg.r#type, &task_msg.payload) {
+ Ok(mut task) => {
+ task = task.with_headers(task_msg.headers.clone());
+ task
+ }
Err(e) => {
tracing::error!("Failed to create task: {}", e);
// 发送 worker 完成事件（即使失败也需要发送）
diff --git a/asynq/src/task.rs b/asynq/src/task.rs
index d493599..1b06174 100644
--- a/asynq/src/task.rs
+++ b/asynq/src/task.rs
@@ -13,6 +13,7 @@ use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fmt::{Debug, Formatter};
+use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;
use uuid::Uuid;
@@ -75,7 +76,46 @@ impl ResultWriter {
&self.task_id
}
}
-
+#[derive(Clone)]
+pub enum Payload {
+ Bytes(Vec<u8>),
+ #[cfg(feature = "json")]
+ Json(Vec<u8>),
+}
+impl Debug for Payload {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Payload::Bytes(bytes) => f.debug_tuple("Bytes").field(bytes).finish(),
+ #[cfg(feature = "json")]
+ Payload::Json(json) => match serde_json::from_slice::<serde_json::Value>(json) {
+ Ok(value) => f.debug_tuple("Json").field(&value).finish(),
+ Err(_e) => f.debug_tuple("Json").field(&json).finish(),
+ },
+ }
+ }
+}
+impl PartialEq for Payload {
+ fn eq(&self, other: &Self) -> bool {
+ match (self, other) {
+ (Payload::Bytes(a), Payload::Bytes(b)) => a == b,
+ #[cfg(feature = "json")]
+ (Payload::Json(a), Payload::Json(b)) => a == b,
+ #[cfg(feature = "json")]
+ _ => false,
+ }
+ }
+}
+impl Deref for Payload {
+ type Target = Vec<u8>;
+
+ fn deref(&self) -> &Self::Target {
+ match self {
+ Payload::Bytes(p) => p,
+ #[cfg(feature = "json")]
+ Payload::Json(p) => p,
+ }
+ }
+}
/// 表示要执行的工作单元的任务
/// Represents a task as a unit of work to be executed
///
@@ -97,7 +137,7 @@ pub struct Task {
pub task_type: String,
/// 任务负载数据
/// Task payload data
- pub payload: Vec<u8>,
+ pub payload: Payload,
/// 任务头信息
/// Task headers
pub headers: HashMap<String, String>,
@@ -137,30 +177,40 @@ impl Task {
Ok(Self {
task_type: task_type.to_string(),
- payload: payload.to_vec(),
+ payload: Payload::Bytes(payload.to_vec()),
headers: Default::default(),
options: TaskOptions::default(),
result_writer: None,
inspector: None,
})
}
- pub fn new_with_headers<T: AsRef<str>>(
- task_type: T,
- payload: &[u8],
- headers: HashMap<String, String>,
- ) -> Result<Self> {
- let mut task = Self::new(task_type, payload)?;
- task.headers = headers;
- Ok(task)
- }
#[cfg(feature = "json")]
/// 使用 JSON 负载创建新任务
/// Create a new task with JSON payload
pub fn new_with_json<T: AsRef<str>, P: Serialize>(task_type: T, payload: &P) -> Result<Self> {
let json_payload = serde_json::to_vec(payload)?;
- Self::new(task_type, &json_payload)
- }
+ let task_type = task_type.as_ref();
+ if task_type.trim().is_empty() {
+ return Err(Error::InvalidTaskType {
+ task_type: task_type.to_string(),
+ });
+ }
+ Ok(Self {
+ task_type: task_type.to_string(),
+ payload: Payload::Json(json_payload),
+ headers: Default::default(),
+ options: TaskOptions::default(),
+ result_writer: None,
+ inspector: None,
+ })
+ }
+ /// 设置任务头信息
+ /// Set task headers
+ pub fn with_headers(mut self, headers: HashMap<String, String>) -> Self {
+ self.headers = headers;
+ self
+ }
/// 设置任务选项
/// Set task options
pub fn with_options(mut self, options: TaskOptions) -> Self {
@@ -609,7 +659,7 @@ mod tests {
fn test_task_creation() {
let task = Task::new("test_task", b"test payload").unwrap();
assert_eq!(task.task_type, "test_task");
- assert_eq!(task.payload, b"test payload");
+ assert_eq!(task.payload, Payload::Bytes(b"test payload".to_vec()));
assert_eq!(task.options.queue, DEFAULT_QUEUE_NAME);
}
@@ -637,11 +687,7 @@ mod tests {
count: 42,
};
- let task = Task::new(
- "test_task",
- &serde_json::to_vec(&payload).unwrap_or_default(),
- )
- .unwrap();
+ let task = Task::new_with_json("test_task", &payload).unwrap();
let decoded: TestPayload = serde_json::from_slice(task.get_payload()).unwrap();
assert_eq!(decoded, payload);
}There is another issue, when the data returned after task dequeue is only of Vectype, the Payload type format is lost during the process from TaskMessage to Task. However, the original asymq version https://github.com/hibiken/asynq/blob/master/internal/proto/asynq.proto There is no place to define the type of payload, it is not enumerable, only Vec. |
Member
|
Are there any other updates? |
Author
|
I see a lot of changes in that diff that seem off-topic to this conversation. Aside from that signal noise, I think this PR will remain stalled on the fact that the contribution will only work for pre-submission tasks, and will not work for tasks re-hydrated from back-end state. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Enqueueing logic
Before
After