Batch oriented consumer API - "read to end"

[jonaslb]

Currently the consumer API is very oriented toward stream processing, which makes sense, but I think there is also a use case where it is desired to "read to the end of stream" as a single batch and then no more. With Kafka, I believe it is possible to react to end-of-stream on a consumer, but I don't believe a similar thing is possible for rmq streams?

I have a crude local wrapper I use for this purpose, but it involves subscribing once using LAST offset spec to get the offset, and then subscribing once more using the stored/initial offset and stopping when the last offset is received. Which turns out to be a relatively large amount of code - especially taking into account that the stream might not exist / multiple streams can be queried - for what I feel is a conceptually simple problem. So, perhaps such a solution in polished form would be appropriate for inclusion here?

[Gsantomaggio]

Hi @jonaslb,
There is no concept of "end of stream" in RabbitMQ. AFAIK, it is the same in Kafka.
Stream is designed for continuous, real-time data streaming.

You need to implement your application logic to define the "end" of the stream. like:

application_properties = {
   "isTheLast":  False,
  }
  amqp_message = AMQPMessage(
       body=bytes("hello", "utf-8"),
                application_properties=application_properties,
 )
await producer.send(stream=STREAM, message=amqp_message)

application_properties = {
       "isTheLast":  True, ## like this
}
amqp_message = AMQPMessage(
                body=bytes("hello: {}".format(i), "utf-8"),
                application_properties=application_properties,
)
await producer.send(stream=STREAM, message=amqp_message)

[jonaslb]

It'll only be the end of the stream temporarily, so you couldn't put it in a message like that.

Kafka has (a) direct inspection of last offset of both topics and partitions via metadata query and (b) in at least the rdkafka client/driver, the ability to set enable.partition.eof which will raise ERR__PARTITION_EOF when the last message is read and there are no more messages.

I'm not saying either of these is best, but I guess what you're saying is that since it isn't native to RabbitMQ, it wouldn't be appropriate for rstream?

[Gsantomaggio]

@jonaslb In RabbitMQ, stream queues don't have this feature yet. So you need to implement something on the client side.

We are working on a feature to determine the last offset. No ETA btw.

[wrobell]

Currently the consumer API is very oriented toward stream processing, which makes sense, but I think there is also a use case where it is desired to "read to the end of stream"
[...]

@jonaslb For my own purpose I define "end of stream" as "there is no message for a period of time". In RbFly you can subscribe with timeout parameter (see https://wrobell.dcmod.org/rbfly/api.html#rbfly.streams.StreamsClient, sorry for the plug).

@Gsantomaggio Does rstream supports something similar?

[Gsantomaggio]

There is not. The user can easily implement it.

[jonaslb]

I thought about this originally, but I didn't like having to put a magic timeout in there. I would rather my batch process is busy for a while and then finishes immediately. That's why I did the workaround mentioned in the OP. I don't think it was entirely easy though, it did end up being a bit of code.