asynq 0.1.8

Simple, reliable & efficient distributed task queue in Rust, inspired by hibiken/asynq
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

//! Tests for macro functionality

#[cfg(feature = "macros")]
mod tests {
  use asynq::{
    error::Result, register_async_handlers, register_handlers, serve_mux::ServeMux,
    server::Handler, task::Task, task_handler, task_handler_async,
  };

  // Define test handlers using macros
  #[task_handler("test:sync")]
  fn test_sync_handler(task: Task) -> Result<()> {
    assert_eq!(task.get_type(), "test:sync");
    Ok(())
  }

  #[task_handler_async("test:async")]
  async fn test_async_handler(task: Task) -> Result<()> {
    assert_eq!(task.get_type(), "test:async");
    Ok(())
  }

  #[task_handler("test:sync2")]
  fn test_sync_handler2(_task: Task) -> Result<()> {
    Ok(())
  }

  #[task_handler_async("test:async2")]
  async fn test_async_handler2(_task: Task) -> Result<()> {
    Ok(())
  }

  #[tokio::test]
  async fn test_task_handler_macro() {
    // Create a ServeMux
    let mut mux = ServeMux::new();

    // Register handlers using macros
    register_handlers!(mux, test_sync_handler);
    register_async_handlers!(mux, test_async_handler);

    // Test sync handler
    let task = Task::new("test:sync", b"test payload").unwrap();
    let result = mux.process_task(task).await;
    assert!(result.is_ok());

    // Test async handler
    let task = Task::new("test:async", b"test payload").unwrap();
    let result = mux.process_task(task).await;
    assert!(result.is_ok());

    // Test non-existent handler
    let task = Task::new("test:nonexistent", b"test payload").unwrap();
    let result = mux.process_task(task).await;
    assert!(result.is_err());
  }

  #[tokio::test]
  async fn test_multiple_handlers_registration() {
    let mut mux = ServeMux::new();

    // Register multiple handlers at once
    register_handlers!(mux, test_sync_handler, test_sync_handler2);
    register_async_handlers!(mux, test_async_handler, test_async_handler2);

    // Test all handlers
    let task1 = Task::new("test:sync", b"payload").unwrap();
    assert!(mux.process_task(task1).await.is_ok());

    let task2 = Task::new("test:sync2", b"payload").unwrap();
    assert!(mux.process_task(task2).await.is_ok());

    let task3 = Task::new("test:async", b"payload").unwrap();
    assert!(mux.process_task(task3).await.is_ok());

    let task4 = Task::new("test:async2", b"payload").unwrap();
    assert!(mux.process_task(task4).await.is_ok());
  }

  #[test]
  fn test_pattern_constants_exist() {
    // Verify that the pattern constants are created by the macros
    assert_eq!(__TEST_SYNC_HANDLER_PATTERN, "test:sync");
    assert_eq!(__TEST_ASYNC_HANDLER_PATTERN, "test:async");
    assert_eq!(__TEST_SYNC_HANDLER2_PATTERN, "test:sync2");
    assert_eq!(__TEST_ASYNC_HANDLER2_PATTERN, "test:async2");
  }

  // Test handlers with wildcard patterns
  #[task_handler("email:*")]
  fn handle_all_emails(task: Task) -> Result<()> {
    assert!(task.get_type().starts_with("email:"));
    Ok(())
  }

  #[task_handler("*:urgent")]
  fn handle_urgent(task: Task) -> Result<()> {
    assert!(task.get_type().ends_with(":urgent"));
    Ok(())
  }

  #[task_handler("*")]
  fn handle_catchall(_task: Task) -> Result<()> {
    Ok(())
  }

  #[tokio::test]
  async fn test_wildcard_patterns_with_macros() {
    let mut mux = ServeMux::new();

    // Register handlers with wildcard patterns
    register_handlers!(mux, handle_all_emails, handle_urgent, handle_catchall);

    // Test prefix wildcard - email:*
    let task1 = Task::new("email:send", b"test").unwrap();
    assert!(mux.process_task(task1).await.is_ok());

    let task2 = Task::new("email:deliver", b"test").unwrap();
    assert!(mux.process_task(task2).await.is_ok());

    // Test suffix wildcard - *:urgent
    let task3 = Task::new("payment:urgent", b"test").unwrap();
    assert!(mux.process_task(task3).await.is_ok());

    let task4 = Task::new("notification:urgent", b"test").unwrap();
    assert!(mux.process_task(task4).await.is_ok());

    // Test catch-all - *
    let task5 = Task::new("any:random:task", b"test").unwrap();
    assert!(mux.process_task(task5).await.is_ok());
  }
}