restructure files and some docs (#134)

* restructe files and some docs

* rename

README.md

@@ -4,6 +4,8 @@ Substrate JSON RPC Gateway.
 
 This is a generalized JSON RPC proxy server with features specifically designed for Substrate RPC and Ethereum RPC.
 
+![alt text](subway-diagram.png)
+
 ## Getting Started
 
 Pull vendors: `git submodule update --init --recursive`

benches/bench/main.rs

@@ -4,7 +4,6 @@ use futures_util::FutureExt;
 use jsonrpsee::core::params::BatchRequestBuilder;
 use pprof::criterion::{Output, PProfProfiler};
 use std::{sync::Arc, time::Duration};
-use subway::extensions::api::SubstrateApiConfig;
 use tokio::runtime::Runtime as TokioRuntime;
 
 use helpers::{
@@ -14,8 +13,7 @@ use helpers::{
 
 use subway::{
     config::{Config, MergeStrategy, MethodParam, MiddlewaresConfig, RpcDefinitions, RpcMethod, RpcSubscription},
-    extensions::{client::ClientConfig, server::ServerConfig, ExtensionsConfig},
-    server::{start_server, SubwayServerHandle},
+    extensions::{api::SubstrateApiConfig, client::ClientConfig, server::ServerConfig, ExtensionsConfig},
 };
 
 mod helpers;
@@ -308,9 +306,9 @@ fn config() -> Config {
     }
 }
 
-async fn server() -> SubwayServerHandle {
+async fn server() -> subway::server::SubwayServerHandle {
     let config = config();
-    start_server(config).await.unwrap()
+    subway::server::build(config).await.unwrap()
 }
 
 fn ws_concurrent_conn_calls(

src/config/mod.rs

@@ -146,6 +146,7 @@ impl From<ParseConfig> for Config {
     }
 }
 
+// read config file specified in command line
 pub fn read_config() -> Result<Config, String> {
     let cmd = Command::parse();
 

src/extensions/api/eth.rs

@@ -5,13 +5,10 @@ use jsonrpsee::core::JsonValue;
 use serde::Deserialize;
 use tokio::{sync::watch, task::JoinHandle};
 
-use crate::{
-    extension::Extension,
-    extensions::{
-        api::{BaseApi, ValueHandle},
-        client::Client,
-    },
-    middleware::ExtensionRegistry,
+use crate::extensions::{
+    api::{BaseApi, ValueHandle},
+    client::Client,
+    Extension, ExtensionRegistry,
 };
 
 pub struct EthApi {

src/extensions/api/substrate.rs

@@ -5,13 +5,10 @@ use jsonrpsee::core::JsonValue;
 use serde::Deserialize;
 use tokio::{sync::watch, task::JoinHandle};
 
-use crate::{
-    extension::Extension,
-    extensions::{
-        api::{BaseApi, ValueHandle},
-        client::Client,
-    },
-    middleware::ExtensionRegistry,
+use crate::extensions::{
+    api::{BaseApi, ValueHandle},
+    client::Client,
+    Extension, ExtensionRegistry,
 };
 
 pub struct SubstrateApi {

src/extensions/cache.rs → src/extensions/cache/mod.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use serde::Deserialize;
 
-use crate::{extension::Extension, middleware::ExtensionRegistry};
+use super::{Extension, ExtensionRegistry};
 
 pub struct Cache {
     pub config: CacheConfig,

src/extensions/client/mod.rs

@@ -22,9 +22,9 @@ use rand::{seq::SliceRandom, thread_rng};
 use serde::Deserialize;
 use tokio::sync::Notify;
 
+use super::ExtensionRegistry;
 use crate::{
-    extension::Extension,
-    middleware::ExtensionRegistry,
+    extensions::Extension,
     utils::{self, errors},
 };
 

src/extensions/event_bus.rs → src/extensions/event_bus/mod.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use serde::Deserialize;
 
-use crate::{extension::Extension, middleware::ExtensionRegistry};
+use super::{Extension, ExtensionRegistry};
 
 pub struct EventBus;
 

src/extensions/merge_subscription.rs → src/extensions/merge_subscription/mod.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use serde::Deserialize;
 
-use crate::{extension::Extension, middleware::ExtensionRegistry};
+use super::{Extension, ExtensionRegistry};
 
 pub struct MergeSubscription {
     pub config: MergeSubscriptionConfig,

src/extensions/mod.rs

@@ -1,16 +1,12 @@
+use async_trait::async_trait;
+use serde::Deserialize;
 use std::{
     any::{Any, TypeId},
     sync::Arc,
 };
-
-use async_trait::async_trait;
-use serde::Deserialize;
 use tokio::sync::RwLock;
 
-use crate::{
-    extension::{Extension, ExtensionBuilder, ExtensionRegistry},
-    utils::{TypeRegistry, TypeRegistryRef},
-};
+use crate::utils::{TypeRegistry, TypeRegistryRef};
 
 pub mod api;
 pub mod cache;
@@ -20,6 +16,57 @@ pub mod merge_subscription;
 pub mod server;
 pub mod telemetry;
 
+#[async_trait]
+pub trait Extension: Sized {
+    type Config: serde::Deserialize<'static>;
+
+    async fn from_config(config: &Self::Config, registry: &ExtensionRegistry) -> Result<Self, anyhow::Error>;
+}
+
+#[async_trait]
+pub trait ExtensionBuilder {
+    fn has(&self, type_id: TypeId) -> bool;
+    async fn build(&self, type_id: TypeId, registry: &ExtensionRegistry) -> anyhow::Result<Arc<dyn Any + Send + Sync>>;
+}
+
+/// ExtensionRegistry is a struct that holds a registry of types and an extension builder.
+/// It allows to get an instance of a type from the registry or build it using the extension builder.
+pub struct ExtensionRegistry {
+    pub registry: TypeRegistryRef,
+    builder: Arc<dyn ExtensionBuilder + Send + Sync>,
+}
+
+impl ExtensionRegistry {
+    /// Creates a new ExtensionRegistry instance with the given registry and builder.
+    pub fn new(registry: TypeRegistryRef, builder: Arc<dyn ExtensionBuilder + Send + Sync>) -> Self {
+        Self { registry, builder }
+    }
+
+    /// Gets an instance of the given type from the registry or builds it using the extension builder.
+    pub async fn get<T: Send + Sync + 'static>(&self) -> Option<Arc<T>> {
+        let reg = self.registry.read().await;
+
+        let ext = reg.get::<T>();
+
+        if ext.is_none() && self.builder.has(TypeId::of::<T>()) {
+            drop(reg);
+            let ext = self
+                .builder
+                .build(TypeId::of::<T>(), self)
+                .await
+                .expect("Failed to build extension");
+            self.registry.write().await.insert_raw(ext);
+            let reg = self.registry.read().await;
+            let ext = reg.get::<T>();
+            assert!(ext.is_some());
+            ext
+        } else {
+            ext
+        }
+    }
+}
+
+// This macro generates the ExtensionsConfig and implements ExtensionBuilder so extensions can be built from config.
 macro_rules! define_all_extensions {
     (
         $(
@@ -87,6 +134,6 @@ define_all_extensions! {
     merge_subscription: merge_subscription::MergeSubscription,
     substrate_api: api::SubstrateApi,
     eth_api: api::EthApi,
-    server: server::Server,
+    server: server::SubwayServerBuilder,
     event_bus: event_bus::EventBus,
 }

src/extensions/server/mod.rs

@@ -4,14 +4,14 @@ use async_trait::async_trait;
 use jsonrpsee::server::{RandomStringIdProvider, RpcModule, ServerBuilder, ServerHandle};
 use serde::Deserialize;
 
-use crate::{extension::Extension, middleware::ExtensionRegistry};
+use super::{Extension, ExtensionRegistry};
 use proxy_get_request::ProxyGetRequestLayer;
 
 use self::proxy_get_request::ProxyGetRequestMethod;
 
 mod proxy_get_request;
 
-pub struct Server {
+pub struct SubwayServerBuilder {
     pub config: ServerConfig,
 }
 
@@ -37,20 +37,20 @@ fn default_request_timeout_seconds() -> u64 {
 }
 
 #[async_trait]
-impl Extension for Server {
+impl Extension for SubwayServerBuilder {
     type Config = ServerConfig;
 
     async fn from_config(config: &Self::Config, _registry: &ExtensionRegistry) -> Result<Self, anyhow::Error> {
         Ok(Self::new(config.clone()))
     }
 }
 
-impl Server {
+impl SubwayServerBuilder {
     pub fn new(config: ServerConfig) -> Self {
         Self { config }
     }
 
-    pub async fn create_server<Fut: Future<Output = anyhow::Result<RpcModule<()>>>>(
+    pub async fn build<Fut: Future<Output = anyhow::Result<RpcModule<()>>>>(
         &self,
         builder: impl FnOnce() -> Fut,
     ) -> anyhow::Result<(SocketAddr, ServerHandle)> {

src/extensions/telemetry.rs → src/extensions/telemetry/mod.rs

@@ -4,7 +4,7 @@ use async_trait::async_trait;
 use opentelemetry::{global, sdk::trace::Tracer, trace::TraceError};
 use serde::Deserialize;
 
-use crate::{extension::Extension, middleware::ExtensionRegistry};
+use super::{Extension, ExtensionRegistry};
 
 #[derive(Deserialize, Debug)]
 #[serde(rename_all = "snake_case")]

src/lib.rs

@@ -1,11 +1,9 @@
 pub mod config;
-pub mod extension;
 pub mod extensions;
 pub mod logger;
-pub mod middleware;
 pub mod middlewares;
 pub mod server;
 pub mod utils;
 
 #[cfg(test)]
-mod integration_tests;
+mod tests;

src/main.rs

@@ -1,27 +1,22 @@
-use opentelemetry::global::shutdown_tracer_provider;
-
-use subway::{config, logger::enable_logger, server};
-
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let config = match config::read_config() {
+    // read config from file
+    let config = match subway::config::read_config() {
         Ok(config) => config,
         Err(e) => {
             return Err(anyhow::anyhow!(e));
         }
     };
 
-    enable_logger();
-
+    subway::logger::enable_logger();
     tracing::trace!("{:#?}", config);
 
-    let subway_server = server::start_server(config).await?;
-    let addr = subway_server.addr;
-    tracing::info!("Server running at {addr}");
+    let subway_server = subway::server::build(config).await?;
+    tracing::info!("Server running at {}", subway_server.addr);
 
     subway_server.handle.stopped().await;
 
-    shutdown_tracer_provider();
+    opentelemetry::global::shutdown_tracer_provider();
 
     Ok(())
 }