-
Notifications
You must be signed in to change notification settings - Fork 28
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[refactoring proposal] consumer api #65
Comments
Hi @withinboredom i feel that current api is a bit outdated and we need to reinvent it. |
Yes, I think that is a good thing though. Where in the code the message is being processed might be quite far from a client (such as deep in a job worker) and keeping (n)ack'ing near where an error or success is actually determined should result in more maintainable code, vs. passing a message (or id) back up the call stack to be handled may be far more complex. |
Is it better to rethink whole api like channel publishing and subscription? $client->getChannel('tester')->publish('hello world') |
currenly im quite happy with the client, but i had some issues discovering features like ACK/NAK as well. Also confusing is we settled preferring the client usage directly our flow is
So perhaps simply add |
@ro0NL @withinboredom please, check queue implementation in latest release. |
no BC breaks 👍 the new API doesnt work for our case, since we fetch/ack/nack in an abstraction layer. So we need to subscribe once, while keep reusing that subscription for subsequent fetches Also only payload+replyTo leaves the abstraction layer, so we cant use |
@ro0NL yep, the idea was to save api as much as possible) ...
// consumer can be used via queue interface
$queue = $consumer->getQueue();
while ($message = $queue->next()) {
if (rand(1, 10) % 2 == 0) {
mail($message->payload, "See you later");
$message->ack();
} else {
// not ack with 1 second timeout
$message->nack(1);
}
// stop processing
if (rand(1, 10) % 2 == 10) {
// don't forget to unsubscribe
$client->unsubscribe($queue);
break;
}
}
// use fetchAll method to batch process messages
// let's set batch size to 50
$queue = $goodbyer->setBatching(50)->create()->getQueue();
// fetching 100 messages provides 2 stream requests
// limit message fetching to 1 second
// it means no more that 100 messages would be fetched
$messages = $queue->setTimeout(1)->fetchAll(100);
$recipients = [];
foreach ($messages as $message) {
$recipients[] = (string) $message->payload;
}
mail_to_all($recipients, "See you later");
// ack all messages
foreach ($messages as $message) {
$message->ack();
} |
it's fair API for standard usage, i agree 👍 edit: actually i can reuse Queue:
|
@nekufa etrias-nl/php-toolkit@eeb68a9 👌 on a side note, perhaps tag 1.0.0 soonish 👼 |
When working on #62, I discovered there are some issues with implementing the consumer model as-is. Further, some "newer" features of NATs is not easily available (NAKs, ordered consumers, delayed messages, etc) so it would be nice to have them.
I took a deep look at the Go implementation to get an idea of what it might look like in PHP. Go tends to be written in such a way that things feel synchronous even though they aren't. So it has the best chance of being a compatible model for PHP (for both traditional PHP and async PHP via fibers).
Taking inspiration from that, there are a few low-level types that this depends on:
Here's the interface inspired by the Go consumer:
What are your thoughts?
The text was updated successfully, but these errors were encountered: