Hi Jon,

Thank you for creating and maintaining the crate, it comes handy for probably quite a lot of people. However I do have a few specific questions that I can't find the answers from the doc.

Q1: Is it possible to measure the time it takes while executing each job? I know that register can return something other than () wrapped within a Result but I don't think that is what c.run returns. Is this kind of measurement possible at all? I am doing logging right now but ideally the results can be stored and pull together at some point.

c.register("foobar", |job| -> io::Result<()> {
    println!("{:?}", job);
    Ok(())
});

Q2: I read the doc and see that this crate comes with a workers option where we can specify more than one thread. Is it possible to pin those threads to different cores to actually take advantage of all the cores I have? If not, I guess we will need to use thread + core affinity with more than one consumers, right?

Best,

To preface, I'm a fairly amateur rust developer and what I'm suggesting is no doubt a breaking change for the worker API, and I'm probably not skilled enough to implement a completely sound library change with this sort of scope.

I'm attempting to use workers written in rust, but I quickly ran into issues with managing state that exists outside of the worker itself. While reading over previously closed issues (particularly #13 and the PR that closed it), I'm wondering if there is a roadmap or want for something akin to rocket's request guards. My primary use case for this is managing DB connections with diesel. Something utilizing https://docs.rs/state/0.4.1/state/ might be possible?

I suppose with the current API I could write a closure that wraps a Fn and grabs a connection, but that seems a bit awkward and I could easily see the function stringing together the workers becoming harder to reason about.

This isn't an absolutely critical or show-stopping goal, since I'm writing my workers in python in the interim but it would be a "nice to have" so that my workers and main codebase can share the exact same model structs and logic without duplication.

Currently, the code will exit once the next job finishes (or immediately if there are no current jobs), but that's not sufficient. A sketch of the right approach is given as a TODO inside Consumer:

Hey there! I'm starting a new project using this library.

Now that failure is deprecated, and the ecosystem is moving towards a system where libraries report implementations of Error, would you be open to a PR replacing usage of failure with proper Error implementations?

Hello,

as per Faktory docs, we need to send a rety value of -1 for not retrying jobs but keep them in the dead queue for later inspection, as of now, the use of the usize type for the retry property prevent this.

Can you change it to isize so we are able to send negative numbers to the Faktory server?

Thanks,

Currently, run (and by extension, run_to_completion) only spawns a single worker thread. While users are free to manage threads themselves, this is still unfortunate as it duplicates heartbeats, and makes error handling more complex (e.g., run_to_completion returning on one thread would kill the others without allowing them to fail jobs).

This should be fixed by instead spinning up a worker pool of configurable size to handle jobs.

Adding support for TLS should be fairly straightforward:

• Add a dependency on native-tls
• Implement StreamConnector for TlsStream

Given that users may also want to be able to specify options, we might need some intermediate builder that exposes TlsConnectorBuilder. I'm not entirely sure what the most ergonomic way to do this is, but one suggestions is to change StreamConnector to:

pub trait StreamConnector {
    type Addr: FromUrl;
    type Stream: Read + Write + Sized + 'static;
    fn connect(self, addr: Self::Addr) -> io::Result<Self::Stream>;
}

And the following impl block for the connection methods on Client to:

impl<C: StreamConnector> Client<C::Stream> {

This will require adding a little more code to the impl for TcpStream, but shouldn't be too bad.

Is it possible to register async handlers? I want to do some work in a handler that ultimately invokes a function that calls await. Currently my code that looks like:

let mut c = ConsumerBuilder::default();
c.register("job_type", &async_handler);

...

async fn async_handler(job: Job) -> io::Result<()> {
	// here I would do stuff and .await
	Ok(())
}

produces:

error[E0271]: expected `&fn(Job) -> impl Future<Output = Result<(), Error>> {async_handler}` to be a reference that returns `Result<(), Error>`, but it returns `impl Future<Output = Result<(), Error>>`
   --> src/main.rs:48:22
    |
48  |     c.register("job_type", &async_handler);
    |       --------          ^^^^^^^^^^^^ expected `Result<(), Error>`, found future
    |       |
    |       required by a bound introduced by this call

I've got a small test running. I can send data through Faktory, and the consumer receives it fine.

If I enqueue a job with args:

vec! [1, 2, 3, 4]

I receive:

[Number(1), Number(2), Number(3), Number(4)]

That shows up with a type of &[Value]. How do I get it back to a Vec? I've been going through serde_json docs trying to figure this out.

Thanks.

Consumer::run_to_completion() has the following signature:

fn run_to_completion<Q, U>(self, queues: &[Q]) -> !

The U parameter appears to be used for nothing, and defining it to be anything when I call the function seems to change nothing; without an explicit definition for it the call does not compile.

Currently, the heartbeat thread sends a heartbeat every 5 seconds, as suggested by the Faktory documentation. However, the heartbeat handling code also performs additional operations, such as watching for the worker thread terminating. This means that there will be a ~2.5s delay between changes to the worker state and run returning, which is unfortunate.

The fix to this is to have the heartbeat thread sleep for shorter amounts of time, but only write once 5s has elapsed.

Per the failure guidance the failure::Error type is typically only recommended when prototyping or for "applications" that won't handle the error. It would be better to define a custom error type and 'derive Fail' for it.

I see there are some parts of the code using a custom error type but others are falling back on Error.

If this sounds reasonable I'm happy to make this change.

We should add a dependency on ctrlc so that the worker can be told to cleanly exit when run_to_completion is called. The implementation will likely be fairly straightforward, essentially just storing STATUS_TERMINATING and letting the heartbeat thread do the cleanup (including failing the currently execution job). This may require some minor changes to make the heartbeat thread recognize that this is indeed a forced exit.

We may also want to allow the user to opt-in to signal handling for run, although that's less clear (we might instead want to return some kind of handle that the caller can use to signal to exit or wait for termination).

Hi the documentation mentions a "labels" field on Job however as far as I can tell there is no such field. Is the documentation incorrect or is there another detail I'm missing?

Faktory now has a Pro edition, and that comes with (at the time of writing) two new API-facing features. Specifically, it adds unique jobs and expiring jobs. It'd be good to have a convenient interface to these without having to set fields in job.custom manually in the right format. These additional convenient methods should probably be under a non-default feature called "pro" so that non-pro users don't see methods they can't use.

https://docs.rs/faktory/0.9.2/faktory/struct.Consumer.html#connecting-to-faktory says:

Faktory suggests using the FAKTORY_PROVIDER and FAKTORY_URL environment variables (see their docs for more information) with localhost:7419 as the fallback default. If you want this behavior, use ConsumerBuilder::connect_env.

ConsumerBuilder::connect_env doesn't exist, and its functionality seems to have been subsumed by ConsumerBuilder::connect.

Sad but true. I want to have this:

    /// Same as `run()` but gets jobs from a faktory server.
impl Foo {
    fn run_faktory() -> Result<()> {
        let faktory_server = "tcp://localhost:7419";
        let mut c: faktory::ConsumerBuilder<&(Fn(faktory::Job) -> Result<()> + Send + Sync + 'static)> = faktory::ConsumerBuilder::default();

        c.register("pin", |job: faktory::Job| -> Result<()> {
            self.do_stuff()?;
            Ok(())
        });

        c.register("unpin", |job: faktory::Job| -> Result<()> {
            self.do_other_stuff()?;
            Ok(())
        });
        
        let mut consumer = c.connect(Some(faktory_server)).unwrap();
        consumer.run_to_completion::<_, ()>(&["default"]);
        if let Err(e) = consumer.run(&["default"]) {
            println!("Worker failed: {}", e);
        }
        Ok(())
    }
}

First problem: c gets closures of two different types. This is an error 'cause closure types are never equivalent. Okay, so we take references to them. This works, but only because of HECKIN BLACK MAGIC; if you do this:

        c.register("pin", &|job: faktory::Job| -> Result<()> {
            self.do_stuff()?;
            Ok(())
        });

it works, but if you do:

        let closure = |job: faktory::Job| -> Result<()> {
            self.do_stuff()?;
            Ok(())
        });
        c.register("pin", &closure);

it can't figure out that the lifetime for the closure is valid.

Okay, this is inconvenient but not impossible. But self.do_stuff() borrows self, which means the closure is not 'static and Can't Work. Okay, so we factor self out into another structure, which is initialized before c.register() and moved into the closure:

        let mock_self = String::from("foo");
        c.register("pin", &move |job: faktory::Job| -> Result<()> {
            println!("Foo is: {}", mock_self);
            Ok(())
        });

But for some reason the move annihilates rustc's magic ability to figure out the closure's lifetime is valid. And on stable, boxed closures do not implement Fn.

It gets even worse from there because the real mock_self I want to use is not Send but I might be able to engineer around that. But the lifetimes still Don't Work.

Summary: Rust closures suck ass. Sad but true. The only good way around it is to make the method take a trait object instead:

trait Runnable {
    fn run(job: faktory::Job) -> Result<()>;
}
impl Runnable for Box<T> where T: Runnable ...

impl ConnectionBuilder {
    pub fn register<K>(&mut self, kind: K, handler: F) -> &mut Self where
        K: Into<String>, F: Runnable { ... }
}

or something like that. Then your "mock-self" object can just implement Runnable, the object has a specific type and lifetime and all that jazz and you don't have to

Maybe I'm wrong and this works fine and I can't figure out how to make all the bits fit together. But every time I use non-trivial closures in Rust they're utterly awful. If you have any suggestions on how to make them better, I'm all ears.

Somewhat like #11 the Producer::connect_env does not seem to exist but is mentioned in the documentation; unlike #11 this functionality does not seem to exist elsewhere?

Currently, there exists a race between the thread that performs heartbeats and the worker thread that completes jobs. Specifically, when the worker waits for the OK following an ACK or a FAIL, it might receive the reply intended for the heartbeat, and vice-versa.

This is normally not an issue, since both heartbeats and workers generally receive OK responses. However, it can misbehave when a non-OK response is given (e.g., an error or a QUIET). There is also a write race here, where technically an outgoing message could be large enough that it ends up being flushed across multiple system calls, at which point a BEAT and an ACK/FAIL could be mixed.

Fixing this will require a bit of engineering. The easiest is probably to spin up a third RPC thread that is responsible for sending messages and waiting for replies in a single atomic step.

Currently setting properties on Job outside of kind and args requires manually building a Job or building a Job with new, then mutating its properties. I'd suggest using a builder to create jobs and make the fields of Job private. My suggested approach would probably be derive_builder.

If this sounds reasonable I'm happy to make this change.