feat: implement utxorpc source

Cargo.toml

@@ -10,6 +10,21 @@ license = "Apache-2.0"
 readme = "README.md"
 authors = ["Santiago Carmuega <[email protected]>"]
 
+[features]
+default = ["deno"]
+deno = ["deno_core", "deno_runtime"]
+sink-file-rotate = ["file-rotate"]
+sink-webhook = ["reqwest"]
+sink-rabbitmq = ["lapin"]
+sink-kafka = ["kafka"]
+sink-aws-sqs = ["aws-config", "aws-types", "aws-sdk-sqs"]
+sink-aws-lambda = ["aws-config", "aws-types", "aws-sdk-lambda"]
+sink-aws-s3 = ["aws-config", "aws-types", "aws-sdk-s3"]
+sink-gcp-pubsub = ["google-cloud-pubsub", "google-cloud-googleapis", "google-cloud-default"]
+sink-gcp-cloudfunction = ["reqwest", "jsonwebtoken"]
+sink-redis = ["r2d2_redis"]
+sink-elasticsearch = ["elasticsearch"]
+source-utxorpc = ["tonic","futures"]
 
 [dependencies]
 pallas = "0.19.0-alpha.1"
@@ -28,11 +43,7 @@ clap = { version = "4.2.7", features = ["derive"] }
 env_logger = "0.10.0"
 crossterm = "0.26"
 merge = "0.1.0"
-config = { version = "0.13.2", default-features = false, features = [
-    "toml",
-    "yaml",
-    "json",
-] }
+config = { version = "0.13.2", default-features = false, features = ["toml","yaml","json"] }
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = { version = "1.0.89", features = ["arbitrary_precision"] }
 strum = "0.24"
@@ -65,18 +76,6 @@ deno_core = { version = "0.188.0", optional = true }
 deno_runtime = { version = "0.114.0", optional = true }
 jsonwebtoken = { version = "8.3.0", optional = true }
 file-rotate = { version = "0.7.5", optional = true }
+tonic = { version = "0.9.2", features = ["tls", "tls-roots"], optional = true}
+futures = { version = "0.3.28", optional = true }
 
-[features]
-default = ["deno"]
-deno = ["deno_core", "deno_runtime"]
-sink-file-rotate = ["file-rotate"]
-sink-webhook = ["reqwest"]
-sink-rabbitmq = ["lapin"]
-sink-kafka = ["kafka"]
-sink-aws-sqs = ["aws-config", "aws-types", "aws-sdk-sqs"]
-sink-aws-lambda = ["aws-config", "aws-types", "aws-sdk-lambda"]
-sink-aws-s3 = ["aws-config", "aws-types", "aws-sdk-s3"]
-sink-gcp-pubsub = ["google-cloud-pubsub", "google-cloud-googleapis", "google-cloud-default"]
-sink-gcp-cloudfunction = ["reqwest", "jsonwebtoken"]
-sink-redis = ["r2d2_redis"]
-sink-elasticsearch = ["elasticsearch"]

docs/pages/v2/sources/utxorpc.mdx

@@ -0,0 +1,31 @@
+# UtxoRPC with Dolos
+
+The UtxoRPC with Dolos (UtxoRPC) source uses gRPC to fetch blocks and receive blocks from a no Dolos.
+
+## Configuration
+
+The following snippet shows an example of how to set up a typical UtxoRPC source:
+
+```toml
+[source]
+type = "UtxoRPC"
+url = ["https://<hostname>"]
+max_items_per_page = 20
+```
+
+### Section `source`:
+
+- `type`: this field must be set to the literal value `UtxoRPC`
+- `url`: A string contains Dolos gRPC url
+- `max_items_per_page`: Number of blocks that will be requested from Dolos when using a chain distance point. Default value is `20`
+
+## Examples
+
+Connecting to a remote Dolos node in preprod through gRPC:
+
+```toml
+[source]
+type = "UtxoRPC"
+url = ["https://50051-romantic-calmness-b55bqg.us1.demeter.run"]
+max_items_per_page = 20
+```

examples/dolos_source/daemon.toml

@@ -0,0 +1,9 @@
+[source]
+type = "UtxoRPC"
+url = "https://50051-romantic-calmness-b55bqg.us1.demeter.run"
+
+[intersect]
+type = "Tip"
+
+[sink]
+type = "Stdout"

examples/n2c_source/.gitignore

@@ -0,0 +1 @@
+node

examples/n2c_source/daemon.toml

@@ -0,0 +1,9 @@
+[source]
+type = "N2C"
+socket_path = "examples/n2c_source/node/node.socket"
+
+[intersect]
+type = "Tip"
+
+[sink]
+type = "Stdout"

examples/n2c_source/docker-compose.yml

@@ -0,0 +1,20 @@
+version: "3"
+services:
+  node:
+    image: inputoutput/cardano-node
+    container_name: node
+    ports:
+      - 3001:3001
+    volumes:
+      - ./node:/data:rw
+      - ./node:/ipc:rw
+    environment:
+      - NETWORK=preprod
+      - CARDANO_BIND_ADDR=0.0.0.0
+      - CARDANO_PORT=3001
+    networks:
+      - dolos-network
+
+networks:
+  dolos-network:
+    driver: bridge

src/sources/mod.rs

@@ -12,12 +12,18 @@ pub mod n2n;
 #[cfg(feature = "aws")]
 pub mod s3;
 
+#[cfg(feature = "source-utxorpc")]
+pub mod utxorpc;
+
 pub enum Bootstrapper {
     N2N(n2n::Stage),
     N2C(n2c::Stage),
 
     #[cfg(feature = "aws")]
     S3(s3::Stage),
+
+    #[cfg(feature = "source-utxorpc")]
+    UtxoRPC(utxorpc::Stage),
 }
 
 impl StageBootstrapper for Bootstrapper {
@@ -28,6 +34,9 @@ impl StageBootstrapper for Bootstrapper {
 
             #[cfg(feature = "aws")]
             Bootstrapper::S3(p) => p.output.connect(adapter),
+
+            #[cfg(feature = "source-utxorpc")]
+            Bootstrapper::UtxoRPC(p) => p.output.connect(adapter),
         }
     }
 
@@ -42,6 +51,9 @@ impl StageBootstrapper for Bootstrapper {
 
             #[cfg(feature = "aws")]
             Bootstrapper::S3(x) => gasket::runtime::spawn_stage(x, policy),
+
+            #[cfg(feature = "source-utxorpc")]
+            Bootstrapper::UtxoRPC(x) => gasket::runtime::spawn_stage(x, policy),
         }
     }
 }
@@ -56,6 +68,9 @@ pub enum Config {
 
     #[cfg(feature = "aws")]
     S3(s3::Config),
+
+    #[cfg(feature = "source-utxorpc")]
+    UtxoRPC(utxorpc::Config),
 }
 
 impl Config {
@@ -66,6 +81,9 @@ impl Config {
 
             #[cfg(feature = "aws")]
             Config::S3(c) => Ok(Bootstrapper::S3(c.bootstrapper(ctx)?)),
+
+            #[cfg(feature = "source-utxorpc")]
+            Config::UtxoRPC(c) => Ok(Bootstrapper::UtxoRPC(c.bootstrapper(ctx)?)),
         }
     }
 }