ingest/pipeline: create new PaymentProcessor

[sreuland]

What problem does your feature solve?

no common data processors exist based on ingest/pipeline sdk per #5400.

What would you like to see?

Create a new PaymentProcessor as an implementation of a Processor for deriving payments in a pipeline from origin network data - LedgerCloseMeta

Since payments are ubiquitous, implementing a new processor based on this model should be a good stable building block and example of deriving application data model from the network data model as the origin event in the ingest/pipeline SDK.

To receive origin network data, PaymentProcessor.subscribe(inboundChannel Channel) should be passed a Channel that will be carrying Message of type: LedgerCloseMeta. The PaymentProcessor will attach its own MessageHandler function using inboundChannel.Subscribe(MessageHandler). The MessageHandler function should validate that receivedMessages are of type: LedgerCloseMeta , if not, it should log error and drop the message.

The PaymentProcessor will perform an EIP Message Translator role taking the network origin data as input and using same parsing logic as Horizon ingestion of payments to derive payments from the tx-meta.

The PaymentProcessor will populate a new derived application model called PaymentEvent for each payment:

type PaymentEvent struct {
 Timestamp       uint
 BuyerAccountId  string
 SellerAccountId string
 AssetCode       string
 Amount          string
}

The PaymentProcessor will use Json for encoding of thePaymentEvent into the payload which must be byte array.

The PaymentProcessor will perform an EIP Splitter role, it will find many PaymentEvent per each LedgerCloseMeta it receives on inboundChannel, and will wrap each PaymentEvent into a separate Message and emit it individually onto the pipeline via outboundChannel.Send().

Depends on #5400
Depends on #5401

What alternatives are there?

[sreuland]

after processors lib design review, we decided to go with a different approach of pipelines created from observable streams using functional operators, this change is not needed anymore and further obsoleted by #5413 and #5414