ingest/pipeline: Create functional producer for CaptiveStellarCore LedgerBackend

[sreuland]

What problem does your feature solve?

captive core ledger backend(CaptiveStellarCore) provides ledger close meta(LCM) individually per GetLedger,but there is no more efficient way to participate as a streaming producer of LCM's..

What would you like to see?

Follow design proposal on Functional Processors Lib
Provide LedgerBackend ‘Producer’ functions for CaptiveCore Backend.
The function will be used as ‘producer’ operator in a pipeline, emitting tx-meta LCM over a callback fn acts as a closure to encapsulate private instance of Captive Core LedgerBackend, avoid any unintended side effects.

// the returned channel will either return one error or it will be closed when the publishing is finished for bounded range
// if it's unbounded, then chan would never be closed.

PublishFromCaptiveCore(ledgerRange ledgerbackend.Range,
                                            captiveCoreConfig ledgerbackend.CaptiveCoreConfig, 
                                            ctx context.Context, 
                                            callback func (xdr.LedgerCloseMeta) error) chan error

The method will return immediate, creating an async worker routine in background to continue processing.

• it uses whatever prepared ledger range was configured for the backend.
• it goes into a loop of invoking internal Captive Core LedgerBacked.GetLedger to iterate each LedgerCloseMeta retrieved for the given prepared ledger range.
• It invokes the callback func per each LedgerCloseMeta.
• If the range was unbounded then it runs infinite until ctx is canceled.

Visualization of where the producer function fits in the larger CDP design for data transformation pipeline:

Relates to:

• Update 'Build Custom Ingestion Pipeline' sample code to use more CDP  stellar-docs#889
• ingest/pipeline: Create functional producer for BufferedStorageBackend #5412
• services/horizon/ingest: express parsing routines for Payment as public, reusable functions #5413
• services/horizon/ingest: express parsing routines for Offers as public, reusable functions #5414

What alternatives are there?

new streaming ingestion app use cases would have to implement the same locally.