Split historical and incoming message retrieval

[steven-sheehy]

Problem
We had to combine historical and incoming message retrieval temporarily in v0.5.4. These flows should be split again so that their implementations can vary. Specifically the historical flow we want to throttle and poll while the incoming flow we want to use the shared poller to scale up.

Solution

• Wrap the historical flow in a new class so its implementation can vary in the future:

public interface TopicMessageRetriever {
  Flux<TopicMessage> retrieve(TopicMessageFilter filter);
}

• Rename or copy PollingTopicListener to PollingTopicMessageRetriever and make it implement TopicMessageRetriever instead
• Enhance PollingTopicMessageRetriever to complete when next poll returns zero rows (or optionally < maxPageSize)
• Add RetrieverProperties with options enabled and pollingFrequency
• Use a more efficient/less racy polling mechanism than Flux.interval() like Flux.repeatWhen()
• Change TopicMessageRepository.findByConsensusTimestampGreaterThan() to:

  Flux<TopicMessage> findLatest(long realm, Collection<Integer> topicIds, long consensusTimestamp);

• Enhance SharedPollingTopicListener to keep track of subscribed topic ids in a concurrent set and pass to TopicMessageRepository.findLatest(...)
• Change SharedPollingTopicListener to default

Alternatives

Additional Context