connectrpc_workers/
lib.rs

1//! ConnectRPC [`ClientTransport`] implementations backed by the Cloudflare
2//! Workers fetch APIs.
3//!
4//! Two transports are provided:
5//!
6//! * [`FetcherTransport`] wraps a [`worker::Fetcher`] (a `[[services]]`
7//!   binding). Use it for inter-service calls within the same Cloudflare
8//!   zone. The runtime short-circuits these requests, so there's no DNS
9//!   lookup, no TLS handshake, and no trip out to the public internet.
10//!   It also supports Durable Object stubs via [`FetcherTransport::from_stub`],
11//!   because `DurableObjectStub` extends `Fetcher` in the Workers runtime.
12//! * [`FetchTransport`] wraps the global [`worker::Fetch`] for arbitrary
13//!   `http://` / `https://` URLs.
14//!
15//! # Sendness
16//!
17//! `ClientTransport` requires `Send + Sync + 'static` on the type and a
18//! `Send + 'static` future. Workers' fetch is `!Send` (everything in
19//! JS-land is `!Send`). We use [`worker::send::SendFuture`] /
20//! [`worker::send::SendWrapper`] to satisfy the bound. workers-rs ships
21//! these specifically because the Workers isolate is single-threaded, so
22//! nothing is ever actually moved across threads.
23//!
24//! # Protocol
25//!
26//! Use [`connectrpc::Protocol::Connect`] (or `GrpcWeb`). Workers fetch
27//! subrequests don't expose raw HTTP/2, so gRPC's trailer requirement
28//! won't survive. Connect over HTTP/1.1 and GrpcWeb (trailers in body)
29//! both work.
30//!
31//! # Example
32//!
33//! ```ignore
34//! use connectrpc::client::ClientConfig;
35//! use connectrpc::Protocol;
36//! use connectrpc_workers::FetcherTransport;
37//!
38//! // Inside a #[event(fetch)] handler:
39//! let echo = env.service("ECHO")?;
40//! let transport = FetcherTransport::new(echo);
41//! let config = ClientConfig::new("http://echo/".parse()?).protocol(Protocol::Connect);
42//! let client = EchoServiceClient::new(transport, config);
43//! let resp = client.echo(EchoRequest { message: "hi".into() }).await?;
44//! ```
45
46use connectrpc::client::{BoxFuture, ClientBody, ClientTransport};
47use http::uri::{Authority, PathAndQuery, Scheme};
48use http::{Request, Response, Uri};
49use worker::send::{SendFuture, SendWrapper};
50use worker::{Body, Fetch, Fetcher, Stub};
51
52/// `ClientTransport` backed by a Workers service binding.
53///
54/// Construct with [`Self::new`]. Cloning is cheap because the underlying
55/// `Fetcher` is just a `JsValue` handle.
56#[derive(Clone)]
57pub struct FetcherTransport {
58    fetcher: SendWrapper<Fetcher>,
59}
60
61impl FetcherTransport {
62    pub fn new(fetcher: Fetcher) -> Self {
63        Self {
64            fetcher: SendWrapper::new(fetcher),
65        }
66    }
67
68    /// Create a transport from a Durable Object [`Stub`].
69    ///
70    /// In the Workers runtime a `DurableObjectStub` extends `Fetcher`, so
71    /// the underlying JS object supports the same `fetch()` API. This
72    /// method casts the stub to a [`Fetcher`] via [`Stub::into_rpc`] and
73    /// wraps it in a [`FetcherTransport`].
74    ///
75    /// ```ignore
76    /// let ns = env.durable_object("MY_DO")?;
77    /// let stub = ns.get_by_name("instance-1")?;
78    /// let transport = FetcherTransport::from_stub(stub);
79    /// let config = ClientConfig::new("http://my-do/".parse()?);
80    /// let client = MyServiceClient::new(transport, config);
81    /// ```
82    pub fn from_stub(stub: Stub) -> Self {
83        Self::new(stub.into_rpc())
84    }
85}
86
87impl std::fmt::Debug for FetcherTransport {
88    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
89        f.debug_struct("FetcherTransport").finish_non_exhaustive()
90    }
91}
92
93impl ClientTransport for FetcherTransport {
94    type ResponseBody = Body;
95    type Error = worker::Error;
96
97    fn send(
98        &self,
99        request: Request<ClientBody>,
100    ) -> BoxFuture<'static, Result<Response<Self::ResponseBody>, Self::Error>> {
101        // Clone the JsValue handle so the future owns it.
102        let fetcher = (*self.fetcher).clone();
103        Box::pin(SendFuture::new(async move {
104            fetcher.fetch_request(request).await
105        }))
106    }
107}
108
109/// `ClientTransport` backed by the global Workers `fetch` (arbitrary URL).
110///
111/// Useful for hitting an external ConnectRPC server. For same-zone calls
112/// prefer [`FetcherTransport`], since service bindings skip DNS and TLS
113/// and don't count against egress.
114///
115/// The transport rewrites the request URI's scheme and authority to point
116/// at `base`, so generated clients can keep using arbitrary `Uri`s in
117/// their `ClientConfig` without caring where the service actually lives.
118#[derive(Clone, Debug)]
119pub struct FetchTransport {
120    scheme: Scheme,
121    authority: Authority,
122}
123
124impl FetchTransport {
125    pub fn new(base: Uri) -> Result<Self, worker::Error> {
126        let parts = base.into_parts();
127        let scheme = parts.scheme.ok_or_else(|| {
128            worker::Error::RustError("FetchTransport base URI is missing a scheme".into())
129        })?;
130        let authority = parts.authority.ok_or_else(|| {
131            worker::Error::RustError("FetchTransport base URI is missing an authority".into())
132        })?;
133        Ok(Self { scheme, authority })
134    }
135}
136
137impl ClientTransport for FetchTransport {
138    type ResponseBody = Body;
139    type Error = worker::Error;
140
141    fn send(
142        &self,
143        request: Request<ClientBody>,
144    ) -> BoxFuture<'static, Result<Response<Self::ResponseBody>, Self::Error>> {
145        let scheme = self.scheme.clone();
146        let authority = self.authority.clone();
147        Box::pin(SendFuture::new(async move {
148            let request = rewrite_uri(request, scheme, authority)?;
149            let req = worker::Request::try_from(request)?;
150            Fetch::Request(req).send().await.and_then(|resp| {
151                let resp: http::Response<Body> = resp.try_into()?;
152                Ok(resp)
153            })
154        }))
155    }
156}
157
158fn rewrite_uri<B>(
159    mut req: http::Request<B>,
160    scheme: Scheme,
161    authority: Authority,
162) -> Result<http::Request<B>, worker::Error> {
163    let path_and_query = req
164        .uri()
165        .path_and_query()
166        .cloned()
167        .unwrap_or_else(|| PathAndQuery::from_static("/"));
168    let new_uri = Uri::builder()
169        .scheme(scheme)
170        .authority(authority)
171        .path_and_query(path_and_query)
172        .build()
173        .map_err(|e| worker::Error::RustError(format!("rewrite uri: {e}")))?;
174    *req.uri_mut() = new_uri;
175    Ok(req)
176}
177
178#[cfg(test)]
179mod tests {
180    use super::*;
181
182    #[test]
183    fn fetch_transport_rejects_uri_without_scheme() {
184        // Path-only URIs parse fine but don't have enough info to send a
185        // real request; we surface that at construction time, not at the
186        // first call.
187        let uri: Uri = "/some/path".parse().unwrap();
188        let err = FetchTransport::new(uri).unwrap_err();
189        assert!(format!("{err}").contains("scheme"));
190    }
191
192    #[test]
193    fn fetch_transport_rejects_uri_without_authority() {
194        // `mailto:` parses with a scheme but no authority.
195        let uri: Uri = "mailto:user@example.com".parse().unwrap();
196        let err = FetchTransport::new(uri).unwrap_err();
197        let msg = format!("{err}");
198        // Either authority or scheme could trigger first depending on
199        // parsing; both are "this URI can't address an HTTP server."
200        assert!(
201            msg.contains("authority") || msg.contains("scheme"),
202            "unexpected error: {msg}"
203        );
204    }
205
206    #[test]
207    fn fetch_transport_accepts_https_uri() {
208        let uri: Uri = "https://example.com:8443/base".parse().unwrap();
209        let t = FetchTransport::new(uri).expect("valid base URI");
210        assert_eq!(t.scheme, Scheme::HTTPS);
211        assert_eq!(t.authority.as_str(), "example.com:8443");
212    }
213
214    #[test]
215    fn rewrite_uri_replaces_scheme_and_authority_keeps_path() {
216        let req: http::Request<()> = http::Request::builder()
217            .uri("http://placeholder.invalid/foo/bar?x=1")
218            .body(())
219            .unwrap();
220        let scheme = Scheme::HTTPS;
221        let authority: Authority = "real.example:8443".parse().unwrap();
222        let rewritten = rewrite_uri(req, scheme, authority).unwrap();
223        let uri = rewritten.uri();
224        assert_eq!(uri.scheme_str(), Some("https"));
225        assert_eq!(
226            uri.authority().map(|a| a.as_str()),
227            Some("real.example:8443")
228        );
229        assert_eq!(uri.path(), "/foo/bar");
230        assert_eq!(uri.query(), Some("x=1"));
231    }
232
233    #[test]
234    fn rewrite_uri_defaults_path_to_root_when_missing() {
235        // `http::Uri` parsed from just `http://host` has no path-and-query.
236        let req: http::Request<()> = http::Request::builder()
237            .uri("http://placeholder")
238            .body(())
239            .unwrap();
240        let scheme = Scheme::HTTP;
241        let authority: Authority = "real.example".parse().unwrap();
242        let rewritten = rewrite_uri(req, scheme, authority).unwrap();
243        assert_eq!(rewritten.uri().path(), "/");
244    }
245
246    /// Compile-time check that both transports satisfy `ClientTransport`,
247    /// including its `Send + Sync + 'static` bound. If this stops compiling,
248    /// one of the `SendWrapper` / `SendFuture` shims is no longer pulling
249    /// its weight.
250    #[test]
251    fn transports_implement_client_transport() {
252        fn assert_transport<T: ClientTransport>() {}
253        assert_transport::<FetcherTransport>();
254        assert_transport::<FetchTransport>();
255    }
256}
connectrpc_workers/lib.rs

connectrpc_workers/
lib.rs
