Struct WorkerOpts

pub struct WorkerOpts<Args, W: Worker<Args> + ?Sized> { /* private fields */ }

Implementations

impl<Args, W> WorkerOpts<Args, W>

where W: Worker<Args>,

pub fn new() -> Self

Examples found in repository

examples/consumer-demo.rs (line 41)

    fn opts() -> sidekiq::WorkerOpts<PaymentReportArgs, Self> {
        sidekiq::WorkerOpts::new().queue("yolo")
    }

examples/demo.rs (line 52)

    fn opts() -> sidekiq::WorkerOpts<PaymentReportArgs, Self> {
        sidekiq::WorkerOpts::new().queue("yolo")
    }

examples/producer-demo.rs (line 41)

    fn opts() -> sidekiq::WorkerOpts<PaymentReportArgs, Self> {
        sidekiq::WorkerOpts::new().queue("yolo")
    }

examples/unique.rs (line 13)

    fn opts() -> sidekiq::WorkerOpts<CustomerNotification, Self> {
        // Use default options to set the unique_for option by default.
        sidekiq::WorkerOpts::new()
            .queue("customers")
            .unique_for(std::time::Duration::from_secs(30))
    }

pub fn retry<RO>(self, retry: RO) -> Self

where RO: Into<RetryOpts>,

pub fn retry_queue<S: Into<String>>(self, retry_queue: S) -> Self

pub fn queue<S: Into<String>>(self, queue: S) -> Self

Examples found in repository

examples/consumer-demo.rs (line 41)

    fn opts() -> sidekiq::WorkerOpts<PaymentReportArgs, Self> {
        sidekiq::WorkerOpts::new().queue("yolo")
    }

examples/demo.rs (line 52)

    fn opts() -> sidekiq::WorkerOpts<PaymentReportArgs, Self> {
        sidekiq::WorkerOpts::new().queue("yolo")
    }

    async fn perform(&self, args: PaymentReportArgs) -> Result<()> {
        use redis::AsyncCommands;

        let times_called: usize = self
            .redis
            .get()
            .await?
            .unnamespaced_borrow_mut()
            .incr("example_of_accessing_the_raw_redis_connection", 1)
            .await?;

        debug!({ "times_called" = times_called }, "Called this worker");

        self.send_report(args.user_guid).await
    }
}

struct FilterExpiredUsersMiddleware;

#[derive(Deserialize)]
struct FiltereExpiredUsersArgs {
    user_guid: String,
}

impl FiltereExpiredUsersArgs {
    fn is_expired(&self) -> bool {
        self.user_guid == "USR-123-EXPIRED"
    }
}

#[async_trait]
impl ServerMiddleware for FilterExpiredUsersMiddleware {
    async fn call(
        &self,
        chain: ChainIter,
        job: &Job,
        worker: Arc<WorkerRef>,
        redis: RedisPool,
    ) -> Result<()> {
        let args: std::result::Result<(FiltereExpiredUsersArgs,), serde_json::Error> =
            serde_json::from_value(job.args.clone());

        // If we can safely deserialize then attempt to filter based on user guid.
        if let Ok((filter,)) = args {
            if filter.is_expired() {
                error!({
                    "class" = &job.class,
                    "jid" = &job.jid,
                    "user_guid" = filter.user_guid
                }, "Detected an expired user, skipping this job");
                return Ok(());
            }
        }

        chain.next(job, worker, redis).await
    }
}

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

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

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

        async move {
            loop {
                PaymentReportWorker::perform_async(
                    &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(
        &redis,
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

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

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

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

    // Enqueue a job
    sidekiq::perform_async(
        &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(
            &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 (line 41)

    fn opts() -> sidekiq::WorkerOpts<PaymentReportArgs, Self> {
        sidekiq::WorkerOpts::new().queue("yolo")
    }

examples/unique.rs (line 14)

    fn opts() -> sidekiq::WorkerOpts<CustomerNotification, Self> {
        // Use default options to set the unique_for option by default.
        sidekiq::WorkerOpts::new()
            .queue("customers")
            .unique_for(std::time::Duration::from_secs(30))
    }

pub fn unique_for(self, unique_for: Duration) -> Self

Examples found in repository

examples/unique.rs (line 15)

    fn opts() -> sidekiq::WorkerOpts<CustomerNotification, Self> {
        // Use default options to set the unique_for option by default.
        sidekiq::WorkerOpts::new()
            .queue("customers")
            .unique_for(std::time::Duration::from_secs(30))
    }

    async fn perform(&self, _args: CustomerNotification) -> Result<()> {
        Ok(())
    }
}

#[derive(Deserialize, Debug, Serialize)]
struct CustomerNotification {
    customer_guid: String,
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let 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(
            &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(())
}

pub fn into_opts(&self) -> EnqueueOpts

pub async fn perform_async( &self, redis: &RedisPool, args: impl Serialize + Send + 'static, ) -> Result<()>

Examples found in repository

examples/unique.rs (lines 59-64)

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

    // Redis
    let manager = RedisConnectionManager::new("redis://127.0.0.1/")?;
    let 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(
            &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 160-165)

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

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

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

        async move {
            loop {
                PaymentReportWorker::perform_async(
                    &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(
        &redis,
        PaymentReportArgs {
            user_guid: "USR-123".into(),
        },
    )
    .await?;

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

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

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

    // Enqueue a job
    sidekiq::perform_async(
        &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(
            &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(())
}

pub async fn perform_in( &self, redis: &RedisPool, duration: Duration, args: impl Serialize + Send + 'static, ) -> Result<()>

Trait Implementations

impl<Args, W: Worker<Args>> Default for WorkerOpts<Args, W>

fn default() -> Self

Returns the “default value” for a type.

impl<Args, W: Worker<Args>> From<&WorkerOpts<Args, W>> for EnqueueOpts

fn from(opts: &WorkerOpts<Args, W>) -> Self

Converts to this type from the input type.