mirrors/axum-connect
1
0
Fork 0
mirror of https://github.com/AThilenius/axum-connect.git synced 2025-01-05 01:39:14 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update to axum 0.7; remove TBody template types

Browse source
This commit is contained in:
Alec Thilenius 2023-12-28 11:50:13 -05:00
parent a7cec357ba
commit 48752280e4
10 changed files with 53 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
README.md
View file

@ -6,9 +6,8 @@ framework](https://connect.build/docs/introduction) to Rust via idiomatic
# Axum Version
> Axum `0.7` isn't yet supported, because I haven't figured out how to handle
> streaming responses with it. Use Axum `0.6` for now. PRs welcome, and I'll get
> it updated when I can.
- `axum-connect:0.3` works with `axum:0.7`
- `axum-connect:0.2` works with `axum:0.6`
# Features 🔍

2
axum-connect-build/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "axum-connect-build"
version = "0.2.0"
version = "0.3.0"
authors = ["Alec Thilenius <alec@thilenius.com>"]
edition = "2021"
categories = [

26
axum-connect-build/src/gen.rs
View file

@ -48,23 +48,20 @@ impl AxumConnectServiceGenerator {
if method.server_streaming {
quote! {
pub fn #method_name<T, H, S, B>(
pub fn #method_name<T, H, S>(
handler: H
) -> impl FnOnce(axum::Router<S, B>) -> axum_connect::router::RpcRouter<S, B>
) -> impl FnOnce(axum::Router<S>) -> axum_connect::router::RpcRouter<S>
where
H: axum_connect::handler::RpcHandlerStream<#input_type, #output_type, T, S, B>,
H: axum_connect::handler::RpcHandlerStream<#input_type, #output_type, T, S>,
T: 'static,
S: Clone + Send + Sync + 'static,
B: axum::body::HttpBody + Send + 'static,
B::Data: Send,
B::Error: Into<axum::BoxError>,
{
move |router: axum::Router<S, B>| {
move |router: axum::Router<S>| {
router.route(
#path,
axum::routing::post(|
axum::extract::State(state): axum::extract::State<S>,
request: axum::http::Request<B>
request: axum::http::Request<axum::body::Body>
| async move {
handler.call(request, state).await
}),
@ -74,23 +71,20 @@ impl AxumConnectServiceGenerator {
}
} else {
quote! {
pub fn #method_name<T, H, S, B>(
pub fn #method_name<T, H, S>(
handler: H
) -> impl FnOnce(axum::Router<S, B>) -> axum_connect::router::RpcRouter<S, B>
) -> impl FnOnce(axum::Router<S>) -> axum_connect::router::RpcRouter<S>
where
H: axum_connect::handler::RpcHandlerUnary<#input_type, #output_type, T, S, B>,
H: axum_connect::handler::RpcHandlerUnary<#input_type, #output_type, T, S>,
T: 'static,
S: Clone + Send + Sync + 'static,
B: axum::body::HttpBody + Send + 'static,
B::Data: Send,
B::Error: Into<axum::BoxError>,
{
move |router: axum::Router<S, B>| {
move |router: axum::Router<S>| {
router.route(
#path,
axum::routing::post(|
axum::extract::State(state): axum::extract::State<S>,
request: axum::http::Request<B>
request: axum::http::Request<axum::body::Body>
| async move {
handler.call(request, state).await
}),

2
axum-connect-examples/Cargo.toml
View file

@ -5,7 +5,7 @@ edition = "2021"
[dependencies]
async-stream = "0.3.5"
axum = "0.6.9"
axum = "0.7.2"
axum-connect = { path = "../axum-connect" }
prost = "0.12.1"
tokio = { version = "1.0", features = ["full"] }

10
axum-connect-examples/src/main.rs
View file

@ -1,5 +1,3 @@
use std::net::SocketAddr;
use async_stream::stream;
use axum::{extract::Host, Router};
use axum_connect::{futures::Stream, prelude::*};
@ -20,13 +18,11 @@ async fn main() {
.rpc(HelloWorldService::say_hello(say_hello_success))
.rpc(HelloWorldService::say_hello_stream(say_hello_stream));
// Axum boilerplate to start the server.
let addr = SocketAddr::from(([127, 0, 0, 1], 3030));
println!("listening on http://{}", addr);
axum::Server::bind(&addr)
.serve(app.into_make_service())
let listener = tokio::net::TcpListener::bind("127.0.0.1:3030")
.await
.unwrap();
println!("listening on http://{:?}", listener.local_addr());
axum::serve(listener, app).await.unwrap();
}
async fn say_hello_success(Host(host): Host, request: HelloRequest) -> HelloResponse {

4
axum-connect/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "axum-connect"
version = "0.2.0"
version = "0.3.0"
authors = ["Alec Thilenius <alec@thilenius.com>"]
edition = "2021"
categories = [
@ -17,7 +17,7 @@ repository = "https://github.com/AThilenius/axum-connect"
[dependencies]
async-stream = "0.3.5"
async-trait = "0.1.64"
axum = "0.6.9"
axum = { version = "0.7.2", features = ["multipart"] }
futures = "0.3.26"
pbjson = "0.6.0"
pbjson-types = "0.6.0"

24
axum-connect/src/handler/codec.rs
View file

@ -1,9 +1,8 @@
use axum::{
body::{Bytes, HttpBody},
body::{self, Body},
extract::FromRequest,
http::{header, request, Request, StatusCode},
response::{IntoResponse, Response},
BoxError,
};
use prost::Message;
use serde::de::DeserializeOwned;
@ -124,8 +123,8 @@ pub(crate) fn decode_check_headers(
Ok(ReqResInto { binary })
}
pub(crate) async fn decode_request_payload<M, S, B>(
req: Request<B>,
pub(crate) async fn decode_request_payload<M, S>(
req: Request<Body>,
state: &S,
as_binary: bool,
for_streaming: bool,
@ -133,26 +132,21 @@ pub(crate) async fn decode_request_payload<M, S, B>(
where
M: Message + DeserializeOwned + Default,
S: Send + Sync + 'static,
B: Send + Sync + 'static,
B: HttpBody + Send + 'static,
B::Data: Send,
B::Error: Into<BoxError>,
{
// Axum-connect only supports unary request types, so we can ignore for_streaming.
if as_binary {
let bytes = match Bytes::from_request(req, state).await {
Ok(bytes) => bytes,
Err(e) => {
return Err(encode_error_response(
let bytes = body::to_bytes(req.into_body(), usize::MAX)
.await
.map_err(|e| {
encode_error_response(
&RpcError::new(
RpcErrorCode::InvalidArgument,
format!("Failed to read request body. {}", e),
),
as_binary,
for_streaming,
))
}
};
)
})?;
let message: M = M::decode(bytes).map_err(|e| {
encode_error_response(

31
axum-connect/src/handler/handler_stream.rs
View file

@ -2,10 +2,9 @@ use std::{convert::Infallible, pin::Pin};
use async_stream::stream;
use axum::{
body::{HttpBody, StreamBody},
body::Body,
http::{header, Request, StatusCode},
response::{IntoResponse, Response},
BoxError,
};
use futures::{Future, Stream, StreamExt};
use prost::Message;
@ -22,12 +21,10 @@ use super::codec::{
decode_check_headers, decode_request_payload, encode_error, encode_error_response, ReqResInto,
};
pub trait RpcHandlerStream<TMReq, TMRes, TUid, TState, TBody>:
Clone + Send + Sized + 'static
{
pub trait RpcHandlerStream<TMReq, TMRes, TUid, TState>: Clone + Send + Sized + 'static {
type Future: Future<Output = Response> + Send + 'static;
fn call(self, req: Request<TBody>, state: TState) -> Self::Future;
fn call(self, req: Request<Body>, state: TState) -> Self::Future;
}
// TODO: Get "connect-timeout-ms" (number as string) and apply timeout.
@ -38,8 +35,8 @@ pub trait RpcHandlerStream<TMReq, TMRes, TUid, TState, TBody>:
// This is here because writing Rust macros sucks a**. So I uncomment this when I'm trying to modify
// the below macro.
// #[allow(unused_parens, non_snake_case, unused_mut)]
// impl<TMReq, TMRes, TInto, TFnItem, TFnFut, TFn, TState, TBody, T1>
// RpcHandlerStream<TMReq, TMRes, (T1, TMReq), TState, TBody> for TFn
// impl<TMReq, TMRes, TInto, TFnItem, TFnFut, TFn, TState, T1>
// RpcHandlerStream<TMReq, TMRes, (T1, TMReq), TState> for TFn
// where
// TMReq: Message + DeserializeOwned + Default + Send + 'static,
// TMRes: Message + Serialize + Send + 'static,
@ -47,15 +44,12 @@ pub trait RpcHandlerStream<TMReq, TMRes, TUid, TState, TBody>:
// TFnItem: Stream<Item = TInto> + Send + Sized + 'static,
// TFnFut: Future<Output = TFnItem> + Send + Sync,
// TFn: FnOnce(T1, TMReq) -> TFnFut + Clone + Send + Sync + 'static,
// TBody: HttpBody + Send + Sync + 'static,
// TBody::Data: Send,
// TBody::Error: Into<BoxError>,
// TState: Send + Sync + 'static,
// T1: RpcFromRequestParts<TMRes, TState> + Send,
// {
// type Future = Pin<Box<dyn Future<Output = Response> + Send>>;
// fn call(self, req: Request<TBody>, state: TState) -> Self::Future {
// fn call(self, req: Request<Body>, state: TState) -> Self::Future {
// Box::pin(async move {
// let (mut parts, body) = req.into_parts();
@ -136,7 +130,7 @@ pub trait RpcHandlerStream<TMReq, TMRes, TUid, TState, TBody>:
// "application/connect+json"
// },
// )],
// StreamBody::new(res),
// Body::from_stream(res),
// )
// .into_response()
// })
@ -148,8 +142,8 @@ macro_rules! impl_handler {
[$($ty:ident),*]
) => {
#[allow(unused_parens, non_snake_case, unused_mut)]
impl<TMReq, TMRes, TInto, TFnItem, TFnFut, TFn, TState, TBody, $($ty,)*>
RpcHandlerStream<TMReq, TMRes, ($($ty,)* TMReq), TState, TBody> for TFn
impl<TMReq, TMRes, TInto, TFnItem, TFnFut, TFn, TState, $($ty,)*>
RpcHandlerStream<TMReq, TMRes, ($($ty,)* TMReq), TState> for TFn
where
TMReq: Message + DeserializeOwned + Default + Send + 'static,
TMRes: Message + Serialize + Send + 'static,
@ -157,16 +151,13 @@ macro_rules! impl_handler {
TFnItem: Stream<Item = TInto> + Send + Sized + 'static,
TFnFut: Future<Output = TFnItem> + Send + Sync,
TFn: FnOnce($($ty,)* TMReq) -> TFnFut + Clone + Send + Sync + 'static,
TBody: HttpBody + Send + Sync + 'static,
TBody::Data: Send,
TBody::Error: Into<BoxError>,
TState: Send + Sync + 'static,
$( $ty: RpcFromRequestParts<TMRes, TState> + Send, )*
{
type Future = Pin<Box<dyn Future<Output = Response> + Send>>;
fn call(self, req: Request<TBody>, state: TState) -> Self::Future {
fn call(self, req: Request<Body>, state: TState) -> Self::Future {
Box::pin(async move {
let (mut parts, body) = req.into_parts();
@ -249,7 +240,7 @@ macro_rules! impl_handler {
"application/connect+json"
},
)],
StreamBody::new(res),
Body::from_stream(res),
)
.into_response()
})

27
axum-connect/src/handler/handler_unary.rs
View file

@ -1,10 +1,9 @@
use std::{convert::Infallible, pin::Pin};
use axum::{
body::HttpBody,
body::Body,
http::{header, Request, StatusCode},
response::{IntoResponse, Response},
BoxError,
};
use futures::Future;
use prost::Message;
@ -21,12 +20,10 @@ use super::codec::{
decode_check_headers, decode_request_payload, encode_error_response, ReqResInto,
};
pub trait RpcHandlerUnary<TMReq, TMRes, TUid, TState, TBody>:
Clone + Send + Sized + 'static
{
pub trait RpcHandlerUnary<TMReq, TMRes, TUid, TState>: Clone + Send + Sized + 'static {
type Future: Future<Output = Response> + Send + 'static;
fn call(self, req: Request<TBody>, state: TState) -> Self::Future;
fn call(self, req: Request<Body>, state: TState) -> Self::Future;
}
// This is for Unary.
@ -40,23 +37,20 @@ pub trait RpcHandlerUnary<TMReq, TMRes, TUid, TState, TBody>:
// This is here because writing Rust macros sucks a**. So I uncomment this when I'm trying to modify
// the below macro.
// #[allow(unused_parens, non_snake_case, unused_mut)]
// impl<TMReq, TMRes, TInto, TFnFut, TFn, TState, TBody, T1>
// RpcHandlerUnary<TMReq, TMRes, (T1, TMReq), TState, TBody> for TFn
// impl<TMReq, TMRes, TInto, TFnFut, TFn, TState, T1>
// RpcHandlerUnary<TMReq, TMRes, (T1, TMReq), TState> for TFn
// where
// TMReq: Message + DeserializeOwned + Default + Send + 'static,
// TMRes: Message + Serialize + Send + 'static,
// TInto: RpcIntoResponse<TMRes>,
// TFnFut: Future<Output = TInto> + Send,
// TFn: FnOnce(T1, TMReq) -> TFnFut + Clone + Send + 'static,
// TBody: HttpBody + Send + Sync + 'static,
// TBody::Data: Send,
// TBody::Error: Into<BoxError>,
// TState: Send + Sync + 'static,
// T1: RpcFromRequestParts<TMRes, TState> + Send,
// {
// type Future = Pin<Box<dyn Future<Output = Response> + Send>>;
// fn call(self, req: Request<TBody>, state: TState) -> Self::Future {
// fn call(self, req: Request<Body>, state: TState) -> Self::Future {
// Box::pin(async move {
// let (mut parts, body) = req.into_parts();
@ -127,23 +121,20 @@ macro_rules! impl_handler {
[$($ty:ident),*]
) => {
#[allow(unused_parens, non_snake_case, unused_mut)]
impl<TMReq, TMRes, TInto, TFnFut, TFn, TState, TBody, $($ty,)*>
RpcHandlerUnary<TMReq, TMRes, ($($ty,)* TMReq), TState, TBody> for TFn
impl<TMReq, TMRes, TInto, TFnFut, TFn, TState, $($ty,)*>
RpcHandlerUnary<TMReq, TMRes, ($($ty,)* TMReq), TState> for TFn
where
TMReq: Message + DeserializeOwned + Default + Send + 'static,
TMRes: Message + Serialize + Send + 'static,
TInto: RpcIntoResponse<TMRes>,
TFnFut: Future<Output = TInto> + Send,
TFn: FnOnce($($ty,)* TMReq) -> TFnFut + Clone + Send + 'static,
TBody: HttpBody + Send + Sync + 'static,
TBody::Data: Send,
TBody::Error: Into<BoxError>,
TState: Send + Sync + 'static,
$( $ty: RpcFromRequestParts<TMRes, TState> + Send, )*
{
type Future = Pin<Box<dyn Future<Output = Response> + Send>>;
fn call(self, req: Request<TBody>, state: TState) -> Self::Future {
fn call(self, req: Request<Body>, state: TState) -> Self::Future {
Box::pin(async move {
let (mut parts, body) = req.into_parts();

11
axum-connect/src/router.rs
View file

@ -1,19 +1,18 @@
use axum::Router;
pub trait RpcRouterExt<S, B>: Sized {
pub trait RpcRouterExt<S>: Sized {
fn rpc<F>(self, register: F) -> Self
where
F: FnOnce(Self) -> RpcRouter<S, B>;
F: FnOnce(Self) -> RpcRouter<S>;
}
impl<S, B> RpcRouterExt<S, B> for Router<S, B> {
impl<S> RpcRouterExt<S> for Router<S> {
fn rpc<F>(self, register: F) -> Self
where
F: FnOnce(Self) -> RpcRouter<S, B>,
F: FnOnce(Self) -> RpcRouter<S>,
{
register(self)
// unsafe { std::mem::transmute::<RpcRouter<S, B>, Router<S, B>>(register(self)) }
}
}
pub type RpcRouter<S, B> = Router<S, B>;
pub type RpcRouter<S> = Router<S>;