Redis streams sbcdn

[Sbcdn]

Implementation of redis-sink using redis streams.
Each event type has its own stream.
Eventstreams can be activated by setting filters.

Example config:

[source]
type = "N2C"
address = ["Unix", "/home/user/remote_node.sock"]
magic = "testnet"

[source.mapper]
include_block_end_events = false
include_transaction_details = true
include_transaction_end_events = true
include_block_cbor = false
include_byron_ebb = false

[[filters]]
type = "Selection"

[filters.check]
predicate = "variant_in"
argument = ["Block","Transaction","Mint","RollBack","CIP25Asset","PlutusScript","NativeScript","TxOutput","Collateral"]

[sink]
type = "Redis"
redis_server = "redis://default:@127.0.0.1:6379/5"

Results in:
('Rollback' and 'PlutusScript' did not happend in the testframe, it creates the key with the first event)

 127.0.0.1:6379[5]> keys *
1) "nativ_script_tx"
2) "mint"
3) "cip25mint"
4) "txout"
5) "transaction"
6) "block"
7) "collateral"

127.0.0.1:6379[5]> XRANGE txout - +
 1) 1) "1650386493395-0"
    2) 1) "14909bd1394f3a75bfdd17a9434bbb05d7d25f6e031f43f8ba7ba7cfc0bd2b52#0"
       2) "{\"context\":{\"block_hash\":\"34bdc7cc4c0dd91e6c8320d11f7531154cdcdb3586137264a216ba07ed15d9aa\",\"block_number\":3487800,\"slot\":56017277,\"timestamp\":1650386493,\"tx_idx\":0,\"tx_hash\":\"14909bd1394f3a75bfdd17a9434bbb05d7d25f6e031f43f8ba7ba7cfc0bd2b52\",\"input_idx\":null,\"output_idx\":0,\"output_address\":null,\"certificate_idx\":null},\"tx_output\":{\"address\":\"addr_test1qpmtp5t0t5y6cqkaz7rfsyrx7mld77kpvksgkwm0p7en7qum7a589n30e80tclzrrnj8qr4qvzj6al0vpgtnmrkkksnqd8upj0\",\"amount\":34293525561,\"assets\":[]},\"fingerprint\":null}"
 2) 1) "1650386493395-1"
    2) 1) "edb6e3f33990fd522acedc5aa336b9f15d9ccb6a6f46a5d7e10e96bd04807fd8#0"
       2) "{\"context\":{\"block_hash\":\"34bdc7cc4c0dd91e6c8320d11f7531154cdcdb3586137264a216ba07ed15d9aa\",\"block_number\":3487800,\"slot\":56017277,\"timestamp\":1650386493,\"tx_idx\":1,\"tx_hash\":\"edb6e3f33990fd522acedc5aa336b9f15d9ccb6a6f46a5d7e10e96bd04807fd8\",\"input_idx\":null,\"output_idx\":0,\"output_address\":null,\"certificate_idx\":null},\"tx_output\":{\"address\":\"addr_test1qqsz4cfjrdr35t6myff6jnww3jl9k07gvhq2e8uwxqme3hy4dxlwlmy28mnpqu2nmkzdalwt4j95zg5pt85wuslhxg5qe9g6ky\",\"amount\":1100000,\"assets\":[]},\"fingerprint\":null}"

[scarmuega]

thanks for the PR @Sbcdn, awesome work!

do you see feasible to skip the custom serialization of each event type and maybe rely on serde?

we do something similar in the ElasticSearch sink:

oura/src/sinks/elastic/run.rs

Line 40 in aa3727c

let req_body = json!(ESRecord::from(event));

Also, the EventData enum has a to_string function that outputs the variant name as string, useful to define the name of the strings.

[Sbcdn]

Hey @scarmuega ,
I made changes as we discussed and added fingerprint support.
Let me know what you think.