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

connectrpc_workers/
lib.rs
