refactor the indexer grpc system.

Cargo.toml

@@ -82,6 +82,7 @@ members = [
     "ecosystem/indexer-grpc/indexer-grpc-cache-worker",
     "ecosystem/indexer-grpc/indexer-grpc-data-service",
     "ecosystem/indexer-grpc/indexer-grpc-file-store",
+    "ecosystem/indexer-grpc/indexer-grpc-file-store-data-verifier",
     "ecosystem/indexer-grpc/indexer-grpc-fullnode",
     "ecosystem/indexer-grpc/indexer-grpc-utils",
     "ecosystem/node-checker",
@@ -226,6 +227,7 @@ aptos-indexer-grpc-data-service = { path = "ecosystem/indexer-grpc/indexer-grpc-
 aptos-indexer-grpc-file-store = { path = "ecosystem/indexer-grpc/indexer-grpc-file-store" }
 aptos-indexer-grpc-fullnode = { path = "ecosystem/indexer-grpc/indexer-grpc-fullnode" }
 aptos-indexer-grpc-utils = { path = "ecosystem/indexer-grpc/indexer-grpc-utils" }
+aptos-indexer-grpc-file-store-data-verifier = { path = "ecosystem/indexer-grpc/indexer-grpc-file-store-data-verifier" }
 aptos-infallible = { path = "crates/aptos-infallible" }
 aptos-inspection-service = { path = "crates/inspection-service" }
 aptos-jellyfish-merkle = { path = "storage/jellyfish-merkle" }
@@ -328,7 +330,7 @@ clap = { version = "3.2.17", features = ["derive", "env", "suggestions"] }
 clap_complete = "3.2.3"
 cloud-storage = { version = "0.11.1", features = ["global-client"] }
 codespan-reporting = "0.11.1"
-console-subscriber = "0.1.8"
+console-subscriber = "0.1.6"
 const_format = "0.2.26"
 criterion = "0.3.5"
 criterion-cpu-time = "0.1.0"
@@ -426,8 +428,8 @@ quote = "1.0.18"
 rand = "0.7.3"
 rand_core = "0.5.1"
 rayon = "1.5.2"
-redis = { version = "0.22.3", features = ["tokio-comp"] }
-redis-test = "0.1.0"
+redis = { version = "0.22.3", features = ["tokio-comp", "script"] }
+redis-test = { version = "0.1.1", features = ["aio"] }
 regex = "1.5.5"
 reqwest = { version = "0.11.11", features = ["blocking", "cookies", "json", "stream"] }
 reqwest-middleware = "0.2.0"

crates/aptos-protos/proto/aptos/datastream/v1/datastream.proto

@@ -42,6 +42,10 @@ message StreamStatus {
 message RawDatastreamRequest {
   // Required; start version of current stream.
   uint64 starting_version = 1;
+
+  // Optional; number of transactions to return in current stream. 
+  // If not set, response streams infinitely.
+  optional uint64 transactions_count = 2;
 }
 
 message RawDatastreamResponse {