Infinite Streaming from Cassandra

Akka Akka Streams & Alpakka
1

How to infinitely stream from Cassandra, assuming some other potentially unknown component writes new entries to the table of interest rather continuously, with alpakka-cassandra?

I could use unfold basically to create a stream of CassandraSources that use a prepared statement and that each query for a result set newer than and disjoint from the previous sources…

…plus a mechanism that infinitely retries the query if the result set was empty in order to wait for new entries, and use flatMapConcat to get one stream.

Or is there already something built-in, or perhaps a trick for the query so that it delivers an infinite stream’ish result set?

1 Like
2

That seems like a good approach. For restarting the source you can use RestartSource and wrap CassandraSources in it.

3

I am now looking for a non-time-based way to fail the stream if upstream completes and not a single element has yet passed through the checkpoint. This I want to apply to fail the inner Source if the query does not result in any elements, which in turn triggers the restart via RestartSource.onFailuresWithBackoff.

4

Just found orElse which should do exactly that.

5

How do you stop polling Cassandra when there’s no demand from upstream? I struggle with buffering inside stream stages: even if there’s no demand in my Sink, if the internal buffers are not filled up, they will keep trying to consume elements. Is there a way around this?