feat: Create new stream subscriber plugin

[jsync-swirlds]

Reviewer Notes

Adding a new plugin

• Instead of beating on the old implementation, implement a new subscriber plugin
  that uses a single client thread and feeds that alternately from messaging or history
  This avoids the countdown latch and state transition logic.
  Also avoids a state machine, just take the next block from one source or the other,
  always getting just the next block to send the client.
  Does require a secondary queue from the messaging, but it's small and simple.
• Adjust the subscriber "method" class to be reused, so we track sessions properly
• Added a latch so the pipeline thread holds until the session is registered with
  messaging before returning. This has minimal impact in production, and makes
  it possible to test reliably.
• Added live block push to queue, live blocks stream correctly.
• Added handling for request ahead of live condition
  • Session just skips over live blocks until live reaches the start block.
• Added proper shutdown of all sessions if the plugin is stopped.
• History and Live both work, and the test disables history to prove it
  • This is necessary because otherwise the test "history" plugin is too fast
    and we always serve history.

TODO

• Add more unit tests
• Add testing that forces a client to fall behind and catch up
• Add some end-to-end test suites to exercise this plugin better
• Add more and better javadoc
• Clean up the (excessive) trace logging
• Add better metrics, the "transition" metrics aren't super helpful.
• Suggested metrics
  • Live batches sent to client (would be good if we can tag this with a client ID)
  • History blocks sent to client (would be good if we can tag this with a client ID)
  • Count of times the client thread must wait for live blocks, and time spent waiting
  • Count of times the client cannot read from history, but subsequently reads from live
  • Client connection and time from connect to session established

Related Issue(s)

Resolves: #1051

[georgi-l95]

Also we'd need to replace in block-node/app/build.gradle.kts:
runtimeOnly("org.hiero.block.node.subscriber")
with:
runtimeOnly("org.hiero.block.node.stream.subscriber")

Looking good, done some manual testing, which will be transformed into E2E tests. Those included:

• Live streaming - Send request with -1 start block to -1 end block. ✅
• Historic streaming - Send request with 0 start block to 10 end block. ✅
• Historic -> Live -> Historic -> Live streaming - Send request with 0 start block to -1 end block. We started receiving blocks, then (with some modifications on simulator side) waited couple of seconds to get behind, then the plugin switched us to historic and when we caught up, we switched to live. ✅

Things left to try:

• Live streaming on two or more consumers
• Historic streaming on two or more consumers
• Historic -> Live -> Historic -> Live streaming on two or more consumers

[ata-nas]

@jsync-swirlds looks good! Looking forward to see the improvements done based on the todo. 🔥

[Nana-EC]

Nice work.
Made it a 1/3 of the way through and will circle back

[AlfredoG87]

LGTM

[AlfredoG87]

Just a nit.

[jasperpotts]

I want to go through in detail but don't want to hold up as I will not get a chance till after devcon. So can merge and I will review later.