Trait sidekiq::Worker

source · 
pub trait Worker<Args>: Send + Sync {
    // Required method
    fn perform<'life0, 'async_trait>(
        &'life0 self,
        args: Args
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided methods
    fn disable_argument_coercion(&self) -> bool { ... }
    fn opts() -> WorkerOpts<Args, Self>
       where Self: Sized { ... }
    fn max_retries(&self) -> usize { ... }
    fn class_name() -> String
       where Self: Sized { ... }
    fn perform_async<'life0, 'async_trait>(
        redis: &'life0 RedisPool,
        args: Args
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: Sized + 'async_trait,
             Args: Send + Sync + Serialize + 'static,
             'life0: 'async_trait { ... }
    fn perform_in<'life0, 'async_trait>(
        redis: &'life0 RedisPool,
        duration: Duration,
        args: Args
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: Sized + 'async_trait,
             Args: Send + Sync + Serialize + 'static,
             'life0: 'async_trait { ... }
}

Required Methods§

source

fn perform<'life0, 'async_trait>( &'life0 self, args: Args ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Provided Methods§

source

fn disable_argument_coercion(&self) -> bool

Signal to WorkerRef to not attempt to modify the JsonValue args before calling the perform function. This is useful if the args are expected to be a Vec<T> that might be len() == 1 or a single sized tuple (T,).

source

fn opts() -> WorkerOpts<Args, Self>
where Self: Sized,

Examples found in repository?
examples/unique.rs (line 57)
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

async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let mut redis = Pool::builder().build(manager).await?;

    // Sidekiq server
    let mut p = Processor::new(redis.clone(), vec!["customers".to_string()]);

    // Add known workers
    p.register(CustomerNotificationWorker);

    // Create a bunch of jobs with the default uniqueness options. Only
    // one of these should be created within a 30 second period.
    for _ in 1..10 {
        CustomerNotificationWorker::perform_async(
            &mut redis,
            CustomerNotification {
                customer_guid: "CST-123".to_string(),
            },
        )
        .await?;
    }

    // Override the unique_for option. Note: Because the code above
    // uses the default unique_for value of 30, this code is essentially
    // a no-op.
    CustomerNotificationWorker::opts()
        .unique_for(std::time::Duration::from_secs(90))
        .perform_async(
            &redis,
            CustomerNotification {
                customer_guid: "CST-123".to_string(),
            },
        )
        .await?;

    p.run().await;
    Ok(())
}
More examples
Hide additional examples
examples/demo.rs (line 158)
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233

async fn main() -> Result<()> {
    tracing_subscriber::fmt().with_max_level(Level::INFO).init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let mut redis = Pool::builder().build(manager).await?;

    tokio::spawn({
        let mut redis = redis.clone();

        async move {
            loop {
                PaymentReportWorker::perform_async(
                    &mut redis,
                    PaymentReportArgs {
                        user_guid: "USR-123".into(),
                    },
                )
                .await
                .unwrap();

                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            }
        }
    });

    // Enqueue a job with the worker! There are many ways to do this.
    PaymentReportWorker::perform_async(
        &mut redis,
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

    PaymentReportWorker::perform_in(
        &mut redis,
        std::time::Duration::from_secs(10),
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

    PaymentReportWorker::opts()
        .queue("brolo")
        .perform_async(
            &mut redis,
            PaymentReportArgs {
                user_guid: "USR-123-EXPIRED".into(),
            },
        )
        .await?;

    sidekiq::perform_async(
        &mut redis,
        "PaymentReportWorker".into(),
        "yolo".into(),
        PaymentReportArgs {
            user_guid: "USR-123".to_string(),
        },
    )
    .await?;

    // Enqueue a job
    sidekiq::perform_async(
        &mut redis,
        "PaymentReportWorker".into(),
        "yolo".into(),
        PaymentReportArgs {
            user_guid: "USR-123".to_string(),
        },
    )
    .await?;

    // Enqueue a job with options
    sidekiq::opts()
        .queue("yolo".to_string())
        .perform_async(
            &mut redis,
            "PaymentReportWorker".into(),
            PaymentReportArgs {
                user_guid: "USR-123".to_string(),
            },
        )
        .await?;

    // Sidekiq server
    let mut p = Processor::new(redis.clone(), vec!["yolo".to_string(), "brolo".to_string()]);

    // Add known workers
    p.register(HelloWorker);
    p.register(PaymentReportWorker::new(redis.clone()));

    // Custom Middlewares
    p.using(FilterExpiredUsersMiddleware).await;

    // Reset cron jobs
    periodic::destroy_all(redis.clone()).await?;

    // Cron jobs
    periodic::builder("0 * * * * *")?
        .name("Payment report processing for a user using json args")
        .queue("yolo")
        .args(json!({ "user_guid": "USR-123-PERIODIC-FROM-JSON-ARGS" }))?
        .register(&mut p, PaymentReportWorker::new(redis.clone()))
        .await?;

    periodic::builder("0 * * * * *")?
        .name("Payment report processing for a user using typed args")
        .queue("yolo")
        .args(PaymentReportArgs {
            user_guid: "USR-123-PERIODIC-FROM-TYPED-ARGS".to_string(),
        })?
        .register(&mut p, PaymentReportWorker::new(redis.clone()))
        .await?;

    p.run().await;
    Ok(())
}
source

fn max_retries(&self) -> usize

source

fn class_name() -> String
where Self: Sized,

Derive a class_name from the Worker type to be used with sidekiq. By default this method will

Examples found in repository?
examples/consumer-demo.rs (line 27)
25
26
27
28
29
30

    async fn send_report(&self, user_guid: String) -> Result<()> {
        // TODO: Some actual work goes here...
        info!({"user_guid" = user_guid, "class_name" = Self::class_name()}, "Sending payment report to user");

        Ok(())
    }
More examples
Hide additional examples
examples/producer-demo.rs (line 27)
25
26
27
28
29
30

    async fn send_report(&self, user_guid: String) -> Result<()> {
        // TODO: Some actual work goes here...
        info!({"user_guid" = user_guid, "class_name" = Self::class_name()}, "Sending payment report to user");

        Ok(())
    }
examples/demo.rs (line 37)
33
34
35
36
37
38
39
40
41

    async fn send_report(&self, user_guid: String) -> Result<()> {
        // TODO: Some actual work goes here...
        info!({
            "user_guid" = user_guid,
            "class_name" = Self::class_name()
        }, "Sending payment report to user");

        Ok(())
    }
source

fn perform_async<'life0, 'async_trait>( redis: &'life0 RedisPool, args: Args ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: Sized + 'async_trait, Args: Send + Sync + Serialize + 'static, 'life0: 'async_trait,

Examples found in repository?
examples/namespaced_demo.rs (line 34)
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

async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let redis = Pool::builder()
        .max_size(100)
        .connection_customizer(sidekiq::with_custom_namespace("yolo_app".to_string()))
        .build(manager)
        .await?;

    tokio::spawn({
        let mut redis = redis.clone();

        async move {
            loop {
                HelloWorker::perform_async(&mut redis, ()).await.unwrap();

                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            }
        }
    });

    // Sidekiq server
    let mut p = Processor::new(redis.clone(), vec!["default".to_string()]);

    // Add known workers
    p.register(HelloWorker);

    // Start!
    p.run().await;
    Ok(())
}
More examples
Hide additional examples
examples/unique.rs (lines 45-50)
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

async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let mut redis = Pool::builder().build(manager).await?;

    // Sidekiq server
    let mut p = Processor::new(redis.clone(), vec!["customers".to_string()]);

    // Add known workers
    p.register(CustomerNotificationWorker);

    // Create a bunch of jobs with the default uniqueness options. Only
    // one of these should be created within a 30 second period.
    for _ in 1..10 {
        CustomerNotificationWorker::perform_async(
            &mut redis,
            CustomerNotification {
                customer_guid: "CST-123".to_string(),
            },
        )
        .await?;
    }

    // Override the unique_for option. Note: Because the code above
    // uses the default unique_for value of 30, this code is essentially
    // a no-op.
    CustomerNotificationWorker::opts()
        .unique_for(std::time::Duration::from_secs(90))
        .perform_async(
            &redis,
            CustomerNotification {
                customer_guid: "CST-123".to_string(),
            },
        )
        .await?;

    p.run().await;
    Ok(())
}
examples/demo.rs (lines 126-131)
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233

async fn main() -> Result<()> {
    tracing_subscriber::fmt().with_max_level(Level::INFO).init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let mut redis = Pool::builder().build(manager).await?;

    tokio::spawn({
        let mut redis = redis.clone();

        async move {
            loop {
                PaymentReportWorker::perform_async(
                    &mut redis,
                    PaymentReportArgs {
                        user_guid: "USR-123".into(),
                    },
                )
                .await
                .unwrap();

                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            }
        }
    });

    // Enqueue a job with the worker! There are many ways to do this.
    PaymentReportWorker::perform_async(
        &mut redis,
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

    PaymentReportWorker::perform_in(
        &mut redis,
        std::time::Duration::from_secs(10),
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

    PaymentReportWorker::opts()
        .queue("brolo")
        .perform_async(
            &mut redis,
            PaymentReportArgs {
                user_guid: "USR-123-EXPIRED".into(),
            },
        )
        .await?;

    sidekiq::perform_async(
        &mut redis,
        "PaymentReportWorker".into(),
        "yolo".into(),
        PaymentReportArgs {
            user_guid: "USR-123".to_string(),
        },
    )
    .await?;

    // Enqueue a job
    sidekiq::perform_async(
        &mut redis,
        "PaymentReportWorker".into(),
        "yolo".into(),
        PaymentReportArgs {
            user_guid: "USR-123".to_string(),
        },
    )
    .await?;

    // Enqueue a job with options
    sidekiq::opts()
        .queue("yolo".to_string())
        .perform_async(
            &mut redis,
            "PaymentReportWorker".into(),
            PaymentReportArgs {
                user_guid: "USR-123".to_string(),
            },
        )
        .await?;

    // Sidekiq server
    let mut p = Processor::new(redis.clone(), vec!["yolo".to_string(), "brolo".to_string()]);

    // Add known workers
    p.register(HelloWorker);
    p.register(PaymentReportWorker::new(redis.clone()));

    // Custom Middlewares
    p.using(FilterExpiredUsersMiddleware).await;

    // Reset cron jobs
    periodic::destroy_all(redis.clone()).await?;

    // Cron jobs
    periodic::builder("0 * * * * *")?
        .name("Payment report processing for a user using json args")
        .queue("yolo")
        .args(json!({ "user_guid": "USR-123-PERIODIC-FROM-JSON-ARGS" }))?
        .register(&mut p, PaymentReportWorker::new(redis.clone()))
        .await?;

    periodic::builder("0 * * * * *")?
        .name("Payment report processing for a user using typed args")
        .queue("yolo")
        .args(PaymentReportArgs {
            user_guid: "USR-123-PERIODIC-FROM-TYPED-ARGS".to_string(),
        })?
        .register(&mut p, PaymentReportWorker::new(redis.clone()))
        .await?;

    p.run().await;
    Ok(())
}
examples/producer-demo.rs (lines 105-110)
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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224

async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let mut redis = Pool::builder().build(manager).await?;

    let mut n = 0;
    let mut last = 0;
    let mut then = std::time::Instant::now();

    loop {
        PaymentReportWorker::perform_async(
            &mut redis,
            PaymentReportArgs {
                user_guid: "USR-123".into(),
            },
        )
        .await
        .unwrap();

        //tokio::time::sleep(std::time::Duration::from_millis(1)).await;

        n += 1;

        if n % 100000 == 0 {
            let now = std::time::Instant::now();
            let delta = n - last;
            last = n;
            let delta_time = now - then;
            if delta_time.as_secs() == 0 {
                continue;
            }
            then = now;
            let rate = delta / delta_time.as_secs();
            println!("Iterations since last: {delta} at a rate of: {rate} iter/sec");
        }
    }

    //    // Enqueue a job with the worker! There are many ways to do this.
    //    PaymentReportWorker::perform_async(
    //        &mut redis,
    //        PaymentReportArgs {
    //            user_guid: "USR-123".into(),
    //        },
    //    )
    //    .await?;
    //
    //    PaymentReportWorker::perform_in(
    //        &mut redis,
    //        std::time::Duration::from_secs(10),
    //        PaymentReportArgs {
    //            user_guid: "USR-123".into(),
    //        },
    //    )
    //    .await?;
    //
    //    PaymentReportWorker::opts()
    //        .queue("brolo")
    //        .perform_async(
    //            &mut redis,
    //            PaymentReportArgs {
    //                user_guid: "USR-123-EXPIRED".into(),
    //            },
    //        )
    //        .await?;
    //
    //    sidekiq::perform_async(
    //        &mut redis,
    //        "PaymentReportWorker".into(),
    //        "yolo".into(),
    //        PaymentReportArgs {
    //            user_guid: "USR-123".to_string(),
    //        },
    //    )
    //    .await?;
    //
    //    // Enqueue a job
    //    sidekiq::perform_async(
    //        &mut redis,
    //        "PaymentReportWorker".into(),
    //        "yolo".into(),
    //        PaymentReportArgs {
    //            user_guid: "USR-123".to_string(),
    //        },
    //    )
    //    .await?;
    //
    //    // Enqueue a job with options
    //    sidekiq::opts()
    //        .queue("yolo".to_string())
    //        .perform_async(
    //            &mut redis,
    //            "PaymentReportWorker".into(),
    //            PaymentReportArgs {
    //                user_guid: "USR-123".to_string(),
    //            },
    //        )
    //        .await?;

    //    // Sidekiq server
    //    let mut p = Processor::new(
    //        redis.clone(),
    //        logger.clone(),
    //        //vec!["yolo".to_string(), "brolo".to_string()],
    //        vec![],
    //    );
    //
    //    //    // Add known workers
    //    //    p.register(HelloWorker);
    //    //    p.register(PaymentReportWorker::new(logger.clone()));
    //    //
    //    // Custom Middlewares
    //    p.using(FilterExpiredUsersMiddleware::new(logger.clone()))
    //        .await;
    //
    //    // Reset cron jobs
    //    periodic::destroy_all(redis.clone()).await?;
    //
    //    // Cron jobs
    //    periodic::builder("0 * * * * *")?
    //        .name("Payment report processing for a random user")
    //        .queue("yolo")
    //        //.args(PaymentReportArgs {
    //        //    user_guid: "USR-123-PERIODIC".to_string(),
    //        //})?
    //        .args(json!({ "user_guid": "USR-123-PERIODIC" }))?
    //        .register(&mut p, PaymentReportWorker::new(logger.clone()))
    //        .await?;
    //
    //    p.run().await;
}
source

fn perform_in<'life0, 'async_trait>( redis: &'life0 RedisPool, duration: Duration, args: Args ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: Sized + 'async_trait, Args: Send + Sync + Serialize + 'static, 'life0: 'async_trait,

Examples found in repository?
examples/demo.rs (lines 149-155)
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233

async fn main() -> Result<()> {
    tracing_subscriber::fmt().with_max_level(Level::INFO).init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let mut redis = Pool::builder().build(manager).await?;

    tokio::spawn({
        let mut redis = redis.clone();

        async move {
            loop {
                PaymentReportWorker::perform_async(
                    &mut redis,
                    PaymentReportArgs {
                        user_guid: "USR-123".into(),
                    },
                )
                .await
                .unwrap();

                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            }
        }
    });

    // Enqueue a job with the worker! There are many ways to do this.
    PaymentReportWorker::perform_async(
        &mut redis,
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

    PaymentReportWorker::perform_in(
        &mut redis,
        std::time::Duration::from_secs(10),
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

    PaymentReportWorker::opts()
        .queue("brolo")
        .perform_async(
            &mut redis,
            PaymentReportArgs {
                user_guid: "USR-123-EXPIRED".into(),
            },
        )
        .await?;

    sidekiq::perform_async(
        &mut redis,
        "PaymentReportWorker".into(),
        "yolo".into(),
        PaymentReportArgs {
            user_guid: "USR-123".to_string(),
        },
    )
    .await?;

    // Enqueue a job
    sidekiq::perform_async(
        &mut redis,
        "PaymentReportWorker".into(),
        "yolo".into(),
        PaymentReportArgs {
            user_guid: "USR-123".to_string(),
        },
    )
    .await?;

    // Enqueue a job with options
    sidekiq::opts()
        .queue("yolo".to_string())
        .perform_async(
            &mut redis,
            "PaymentReportWorker".into(),
            PaymentReportArgs {
                user_guid: "USR-123".to_string(),
            },
        )
        .await?;

    // Sidekiq server
    let mut p = Processor::new(redis.clone(), vec!["yolo".to_string(), "brolo".to_string()]);

    // Add known workers
    p.register(HelloWorker);
    p.register(PaymentReportWorker::new(redis.clone()));

    // Custom Middlewares
    p.using(FilterExpiredUsersMiddleware).await;

    // Reset cron jobs
    periodic::destroy_all(redis.clone()).await?;

    // Cron jobs
    periodic::builder("0 * * * * *")?
        .name("Payment report processing for a user using json args")
        .queue("yolo")
        .args(json!({ "user_guid": "USR-123-PERIODIC-FROM-JSON-ARGS" }))?
        .register(&mut p, PaymentReportWorker::new(redis.clone()))
        .await?;

    periodic::builder("0 * * * * *")?
        .name("Payment report processing for a user using typed args")
        .queue("yolo")
        .args(PaymentReportArgs {
            user_guid: "USR-123-PERIODIC-FROM-TYPED-ARGS".to_string(),
        })?
        .register(&mut p, PaymentReportWorker::new(redis.clone()))
        .await?;

    p.run().await;
    Ok(())
}

Implementors§