Redefine ExcService and add SendExcService (#72)

examples/examples/box_exc_service.rs

@@ -2,7 +2,10 @@ use std::{ops::RangeBounds, str::FromStr, time::Duration};
 
 use clap::Parser;
 use exc::{
-    core::types::{Candle, SubscribeTickers, SubscribeTrades, Ticker, Trade},
+    core::{
+        types::{Candle, SubscribeTickers, SubscribeTrades, Ticker, Trade},
+        IntoService,
+    },
     prelude::*,
     types::instrument::InstrumentMeta,
     util::instrument::PollInstrumentsLayer,
@@ -48,7 +51,7 @@ impl Exchange {
         let trade_svc =
             exc::Okx::endpoint()
                 .connect_exc()
-                .into_layered(&layer_fn(|svc: exc::Okx| {
+                .into_layered(&layer_fn(|svc: IntoService<_, _>| {
                     svc
                         // We use the `adapt` method to convert the request type to `SubscribeTickers`.
                         .adapt::<SubscribeTickers>()

exc-core/src/lib.rs

@@ -2,9 +2,6 @@
 
 #![deny(missing_docs)]
 
-/// Core services.
-pub use exc_service as service;
-
 /// The definition of an exchange.
 pub mod exchange;
 
@@ -24,6 +21,10 @@ pub mod util;
 /// Exc Symbol.
 pub use exc_symbol as symbol;
 
-pub use self::service::{Adaptor, Exc, ExcLayer, ExcService, ExcServiceExt, IntoExc, Request};
-pub use exc_service::{error::InstrumentError, ExchangeError};
+pub use self::service::{
+    traits::{AsService, IntoService},
+    Adaptor, Exc, ExcLayer, ExcService, ExcServiceExt, IntoExc, Request,
+};
+pub use exc_service::{self as service, error::InstrumentError, ExchangeError, SendExcService};
+
 pub use positions::prelude::{Asset, Instrument, ParseAssetError, ParseSymbolError, Str, Symbol};

exc-core/src/util/fetch_candles.rs

@@ -1,4 +1,5 @@
 use async_stream::try_stream;
+use exc_service::ExcServiceExt;
 use futures::{future::BoxFuture, FutureExt, StreamExt};
 use std::ops::Bound;
 use tower::{Layer, Service, ServiceExt};
@@ -8,7 +9,7 @@ use crate::{
         candle::{CandleStream, QueryCandles, QueryLastCandles},
         QueryFirstCandles,
     },
-    ExcService, ExchangeError,
+    ExcService, ExchangeError, IntoService,
 };
 
 use std::num::NonZeroUsize;
@@ -50,7 +51,7 @@ where
 
     fn layer(&self, inner: S) -> Self::Service {
         FetchCandlesBackward {
-            svc: Buffer::new(inner, self.bound),
+            svc: Buffer::new(inner.into_service(), self.bound),
             limit: self.limit,
         }
     }
@@ -61,7 +62,7 @@ pub struct FetchCandlesBackward<S>
 where
     S: ExcService<QueryLastCandles> + 'static,
 {
-    svc: Buffer<S, QueryLastCandles>,
+    svc: Buffer<IntoService<S, QueryLastCandles>, QueryLastCandles>,
     limit: NonZeroUsize,
 }
 
@@ -145,7 +146,7 @@ where
 
     fn layer(&self, inner: S) -> Self::Service {
         FetchCandlesForward {
-            svc: Buffer::new(inner, self.bound),
+            svc: Buffer::new(inner.into_service(), self.bound),
             limit: self.limit,
         }
     }
@@ -156,7 +157,7 @@ pub struct FetchCandlesForward<S>
 where
     S: ExcService<QueryFirstCandles> + 'static,
 {
-    svc: Buffer<S, QueryFirstCandles>,
+    svc: Buffer<IntoService<S, QueryFirstCandles>, QueryFirstCandles>,
     limit: NonZeroUsize,
 }
 

exc-core/src/util/poll_instruments.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use async_stream::stream;
-use exc_service::{ExcService, ExchangeError};
+use exc_service::{ExcService, ExcServiceExt, ExchangeError};
 use exc_types::{FetchInstruments, SubscribeInstruments};
 use futures::{
     future::{ready, Ready},
@@ -46,7 +46,12 @@ where
                 interval.tick().await;
             }
         };
-        let stream = self.inner.clone().call_all(req).try_flatten();
+        let stream = self
+            .inner
+            .clone()
+            .into_service()
+            .call_all(req)
+            .try_flatten();
         ready(Ok(stream.boxed()))
     }
 }

exc-core/src/util/trade_bid_ask.rs

@@ -52,20 +52,20 @@ where
     S: Clone + Send + 'static,
     S: ExcService<SubscribeTrades>,
     S: ExcService<SubscribeBidAsk>,
-    <S as Service<SubscribeTrades>>::Future: Send,
-    <S as Service<SubscribeBidAsk>>::Future: Send,
+    <S as ExcService<SubscribeTrades>>::Future: Send,
+    <S as ExcService<SubscribeBidAsk>>::Future: Send,
 {
     type Response = TickerStream;
     type Error = ExchangeError;
     type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Service::<SubscribeTrades>::poll_ready(&mut self.svc, cx)
+        Service::<SubscribeTrades>::poll_ready(&mut self.svc.as_service(), cx)
     }
 
     fn call(&mut self, req: SubscribeTickers) -> Self::Future {
         let trade = Service::<SubscribeTrades>::call(
-            &mut self.svc,
+            &mut self.svc.as_service(),
             SubscribeTrades {
                 instrument: req.instrument.clone(),
             },

exc-make/Cargo.toml

@@ -11,7 +11,7 @@ description.workspace = true
 rust-version.workspace = true
 
 [dependencies]
-exc-service = { workspace = true }
+exc-service = { workspace = true, features = ["send"] }
 exc-types = { workspace = true }
 futures = { workspace = true }
 tower-make = { workspace = true }

exc-make/src/cancel.rs

@@ -1,4 +1,4 @@
-use exc_service::{ExcService, ExchangeError, Request};
+use exc_service::{ExchangeError, Request, SendExcService};
 use exc_types::CancelOrder;
 use futures::{future::MapErr, TryFutureExt};
 use std::{
@@ -14,7 +14,7 @@ pub struct MakeCancelOrderOptions {}
 /// Make a service to cancel orders.
 pub trait MakeCancelOrder {
     /// Service to cancel orders.
-    type Service: ExcService<CancelOrder>;
+    type Service: SendExcService<CancelOrder>;
 
     /// The future of the service.
     type Future: Future<Output = Result<Self::Service, ExchangeError>>;
@@ -42,7 +42,7 @@ where
         Response = <CancelOrder as Request>::Response,
         Error = ExchangeError,
     >,
-    M::Service: ExcService<CancelOrder>,
+    M::Service: SendExcService<CancelOrder>,
     M::MakeError: Into<ExchangeError>,
 {
     type Service = M::Service;

exc-make/src/candles.rs

@@ -1,4 +1,4 @@
-use exc_service::{ExcService, ExchangeError, Request};
+use exc_service::{ExchangeError, Request, SendExcService};
 use exc_types::QueryCandles;
 use futures::{future::MapErr, TryFutureExt};
 use std::{
@@ -19,7 +19,7 @@ pub struct MakeFetchCandlesOptions {
 /// Make a service to subscribe instruments.
 pub trait MakeFetchCandles {
     /// Service to fetch candles.
-    type Service: ExcService<QueryCandles>;
+    type Service: SendExcService<QueryCandles>;
 
     /// The future of the service.
     type Future: Future<Output = Result<Self::Service, ExchangeError>>;
@@ -47,7 +47,7 @@ where
         Response = <QueryCandles as Request>::Response,
         Error = ExchangeError,
     >,
-    M::Service: ExcService<QueryCandles>,
+    M::Service: SendExcService<QueryCandles>,
     M::MakeError: Into<ExchangeError>,
 {
     type Service = M::Service;

exc-make/src/check.rs

@@ -1,4 +1,4 @@
-use exc_service::{ExcService, ExchangeError, Request};
+use exc_service::{ExchangeError, Request, SendExcService};
 use exc_types::GetOrder;
 use futures::{future::MapErr, TryFutureExt};
 use std::{
@@ -14,7 +14,7 @@ pub struct MakeCheckOrderOptions {}
 /// Make a service to check orders.
 pub trait MakeCheckOrder {
     /// Service to check orders.
-    type Service: ExcService<GetOrder>;
+    type Service: SendExcService<GetOrder>;
 
     /// The future of the service.
     type Future: Future<Output = Result<Self::Service, ExchangeError>>;
@@ -42,7 +42,7 @@ where
         Response = <GetOrder as Request>::Response,
         Error = ExchangeError,
     >,
-    M::Service: ExcService<GetOrder>,
+    M::Service: SendExcService<GetOrder>,
     M::MakeError: Into<ExchangeError>,
 {
     type Service = M::Service;

exc-make/src/instruments.rs

@@ -1,4 +1,4 @@
-use exc_service::{ExcService, ExchangeError, Request};
+use exc_service::{ExchangeError, Request, SendExcService};
 use exc_types::SubscribeInstruments;
 use futures::{future::MapErr, TryFutureExt};
 use std::{
@@ -14,7 +14,7 @@ pub struct MakeInstrumentsOptions {}
 /// Make a service to subscribe instruments.
 pub trait MakeInstruments {
     /// Service to subscribe instruments.
-    type Service: ExcService<SubscribeInstruments>;
+    type Service: SendExcService<SubscribeInstruments>;
 
     /// The future of the service.
     type Future: Future<Output = Result<Self::Service, ExchangeError>>;
@@ -42,7 +42,7 @@ where
         Response = <SubscribeInstruments as Request>::Response,
         Error = ExchangeError,
     >,
-    M::Service: ExcService<SubscribeInstruments>,
+    M::Service: SendExcService<SubscribeInstruments>,
     M::MakeError: Into<ExchangeError>,
 {
     type Service = M::Service;

exc-make/src/orders.rs

@@ -1,4 +1,4 @@
-use exc_service::{ExcService, ExchangeError, Request};
+use exc_service::{ExchangeError, Request, SendExcService};
 use exc_types::SubscribeOrders;
 use futures::{future::MapErr, TryFutureExt};
 use std::{
@@ -14,7 +14,7 @@ pub struct MakeSubscribeOrdersOptions {}
 /// Make a service to subscribe orders.
 pub trait MakeSubscribeOrders {
     /// Service to subscribe orders.
-    type Service: ExcService<SubscribeOrders>;
+    type Service: SendExcService<SubscribeOrders>;
 
     /// The future of the service.
     type Future: Future<Output = Result<Self::Service, ExchangeError>>;
@@ -42,7 +42,7 @@ where
         Response = <SubscribeOrders as Request>::Response,
         Error = ExchangeError,
     >,
-    M::Service: ExcService<SubscribeOrders>,
+    M::Service: SendExcService<SubscribeOrders>,
     M::MakeError: Into<ExchangeError>,
 {
     type Service = M::Service;

exc-make/src/place.rs

@@ -1,4 +1,4 @@
-use exc_service::{ExcService, ExchangeError, Request};
+use exc_service::{ExchangeError, Request, SendExcService};
 use exc_types::PlaceOrder;
 use futures::{future::MapErr, TryFutureExt};
 use std::{
@@ -14,7 +14,7 @@ pub struct MakePlaceOrderOptions {}
 /// Make a service to place orders.
 pub trait MakePlaceOrder {
     /// Service to place orders.
-    type Service: ExcService<PlaceOrder>;
+    type Service: SendExcService<PlaceOrder>;
 
     /// The future of the service.
     type Future: Future<Output = Result<Self::Service, ExchangeError>>;
@@ -42,7 +42,7 @@ where
         Response = <PlaceOrder as Request>::Response,
         Error = ExchangeError,
     >,
-    M::Service: ExcService<PlaceOrder>,
+    M::Service: SendExcService<PlaceOrder>,
     M::MakeError: Into<ExchangeError>,
 {
     type Service = M::Service;

exc-make/src/tickers.rs

@@ -1,4 +1,4 @@
-use exc_service::{ExcService, ExchangeError, Request};
+use exc_service::{ExchangeError, Request, SendExcService};
 use exc_types::SubscribeTickers;
 use futures::{future::MapErr, TryFutureExt};
 use std::{
@@ -14,7 +14,7 @@ pub struct MakeTickersOptions {}
 /// Make a service to subscribe tickers.
 pub trait MakeTickers {
     /// Service to subscribe tickers.
-    type Service: ExcService<SubscribeTickers>;
+    type Service: SendExcService<SubscribeTickers>;
 
     /// The future of the service.
     type Future: Future<Output = Result<Self::Service, ExchangeError>>;
@@ -42,7 +42,7 @@ where
         Response = <SubscribeTickers as Request>::Response,
         Error = ExchangeError,
     >,
-    M::Service: ExcService<SubscribeTickers>,
+    M::Service: SendExcService<SubscribeTickers>,
     M::MakeError: Into<ExchangeError>,
 {
     type Service = M::Service;

exc-service/Cargo.toml

@@ -15,6 +15,10 @@ retry = ["tower/retry", "humantime", "tokio/time", "tracing"]
 limit = ["tower/limit"]
 http = ["hyper"]
 
+# Add [`SendExcSerivce`] which is a [`ExcService`] that is `Send`.
+# as a workaround for https://github.com/rust-lang/rust/issues/20671
+send = []
+
 [dependencies]
 tower = { workspace = true, default-features = false, features = ["util"] }
 futures = { workspace = true }