Allow specify the ShardIterator where consumer should start reading from Horizon or Latest

[ghost]

PR regarding #52 . WIP

We open this PR so we can see how this evolves. Notice it's a Draft Pull Request, do not merge.

We copy-paste the conversation of the issue here:

Adding the position to the EnableEventSource would also be slightly confusing, I would prefer to:

• Add a @EnableMessageLogReceiverEndpoint similar to the @EnableEventSource and @EnableMessageQueueReceiverEndpoint. The annotation should then have an optional property to select the start position (horizon, latest)
• Add a @MessageLogConsumer annotation similar to the @MessageQueueConsumer or @EventSourceConsumer annotation.
• Add StartFrom.LATEST as another value
• Extend ShardPosition by ShardPosition.fromLatest()
• Extend ChannelPosition to support ShardPosition.fromLatest() in addition to the current ShardPosition.fromHorizon()

[ghost]

@guidosteinacker1969

Adding the position to the EnableEventSource would also be slightly confusing.

Yes, I assume an EventSource shouldn't have the option of start reading from a point. It must consume all the messages to become a real EventSourced application.

Creating the Annotation EnableMessageLogReceiverEndpoint & MessageLogConsumer solves this problem.

Even though we create a new EnableMessageLogReceiverEndpointRegisterBean for those new annotations, we found that we end up modifying the AbstractMessageEndpoint, implying a lot of changes because the need of storing the iteratorAt with this new information.

We decided to have a default value to Horizon, and just override the value if it comes from this new annotation. We will proceed from this idea.

Note: We are working on removing the iteratorAt from EventSource interface.

[gsteinacker]

Why do you need to modify the AbstractMessageEndpoint? This is used both for senders and receivers. Event AbstractMessageReceiverEndpoint should not need to be modified. It should be enough to add another case to KinesisShardIterator.buildIteratorShardRequest() (plus changes in InMemoryChannel.consumeUntil()). The start position (horizon, latest) must only be made available in ChannelPosition, so the shard iterator knows about where to start.

[ghost]

Event AbstractMessageReceiverEndpoint should not need to be modified. It should be enough to add another case to KinesisShardIterator.buildIteratorShardRequest() (plus changes in InMemoryChannel.consumeUntil())

OK. It might was because the kind of test we had that let into that modification. We will keep sharing the things we found just to have a more verbose track history.