HandleMessage and msg.Requeue

[Jon Buffington]

I have a case where a message is received by a client but the message needs to processed at a later time. After reviewing the doc, I prototyped the following pseudo code:

func (l loggy) HandleMessage(m *nsq.Message) error {
data := Data{}
err := json.Unmarshal(m.Body, &data)
if err != nil {
return err
}
…
if !data.IsReady() {
m.Requeue(60000)
return nil
}
…
return nil
}

After looking at the source for Message.Requeue and Reader.syncHandler, I am not sure the above approach is correct. https://github.com/bitly/go-nsq/blob/master/reader.go#L972-L975 indicates that the changes by Message.Requeue would be overwritten/ignored.

Is there a recommended approach to requeue a message for later processing by a client?

[Matt Reiferson]

Hi Jon,

Thanks for your question.

You would want to use what go-nsq refers to as an “AsyncHandler”, which would *not* automatically reply on your behalf.

Admittedly, this API is a bit confusing.  I have a WIP pull request that I’m looking to land soon that cleans up the public go-nsq API and does away with the notion of two different types of handlers.

If you’re interested in checking it out, see https://github.com/bitly/go-nsq/pull/30

Regards,

Matt

[Jon Buffington]

Matt,

Thank you for your guidance and all your work on nsq.

I stumbled upon a similar conclusion to use AsyncHandler after reading the source further. The source is very clear once I understood the design better.

I will take a look at the pull request later today.

[Matt Reiferson]

You’re welcome Jon!

Yea, it makes more sense after seeing AsyncHandler used, it’s an artifact of how we structured consumers at bitly.

I actually realized you still might have one additional hurdle to overcome for your use case w/ go-nsq.

go-nsq doesn’t support the concept of “skipping backoff” - so when you explicitly requeue your message for the purpose of handling later what’s going to end up happening is that go-nsq will slow down the rate of ingestion exponentially.

You don’t want this :)

This is actually a missing feature in go-nsq (pynsq, for example, has this capability).

Since I’m in the process of doing some refactoring, I’m going to add this to the pull request I mentioned earlier.

[Jon Buffington]

Matt,

I will to test your public_api_30 branch this week. Finally have a chance.

Thanks for your heads up on the back-off timer. We are trying a work around where we change the reader's setting (e.g., r.SetMaxBackoffDuration(5 * time.Second)).

Let me know if I should take a crack at adding skipping back-off.

Regards,

Jon

[Matt Reiferson]

Yea, I guess you could work around it by setting max backoff duration to 0, which would turn off backoff (not really desirable though).

I should have time this week to finish up that PR and get it in, so any feedback on the (new) API would be helpful.

[Matt Reiferson]

Jon,

FYI I added the “requeue without backoff” to that pull request I mentioned.  It can be used by specifying an optional 2nd parameter to msg.Requeue(), see the gist below:

https://gist.github.com/mreiferson/11403980

Let me know how your testing goes!

[Jon Buffington]

Matt,

Thanks! Overall, the changes look great and I have started adapting one of our readers.

I see reader AddHandler() is now SetHandler() and SetMaxInFlight() is part of Config. I have made those changes to our code.

After reviewing reader handlerLoop(), to re-queue w/o back-off, the handler code would be similar to:

func (h handler) HandleMessage(message *nsq.Message) error {

  // ... decode message and attempt to handle.

  // if message needs to re-queue for later processing.

  message.EnableAsync()

  message.RequeueWithoutBackoff(30*time.Second)

  return nil

}

Is the above snippet what you intended?

[Matt Reiferson]

Great, thanks!

You don’t need to call `message.EnableAsync()` in that case, it’s only necessary if you don’t want go-nsq to respond automatically upon returning from HandleMessage.

[Jon Buffington]

Matt,

Duh. I should have waited for the gist to load before asking the below question. We need to add message.Finish() before return nil.

Allowing the message to become async on a per-message is really useful. Per your earlier response, is "Async" the clearest terminology? From my perspective, the difference is automatic vs. manual message handling. In our case, we have handled the message in HandleMessage() except that we took over message state handling.

[Jon Buffington]

So its as simple as:

func (h handler) HandleMessage(message *nsq.Message) error {

  // ... decode message and attempt to handle.

  // needs to re-queue for later processing.

  message.RequeueWithoutBackoff(30*time.Second)

  return nil

}

Thanks for the clarification!

[Matt Reiferson]

Yea, I’ve never been happy with the term async for that use case, but it does at least mirror pynsq which originated it.

Any suggestions?

[Jon Buffington]

Matt,

What do you think about EnableAsync() -> DisableAutoFinish()? IsAsync() could become !IsAutoFinishing(). The message handling patterns are similar to automatic vs. manual resource management.

[Matt Reiferson]

I like that actually, with one minor change, DisableAutoResponse() - “response" rather than “finish" because sometimes it requeues on your behalf...