+4 -3

crates/jacquard-common/src/stream.rs

··· 44 44 45 45 use std::error::Error; 46 46 use std::fmt; 47 47 + use std::pin::Pin; 47 48 48 49 /// Boxed error type for streaming operations 49 50 pub type BoxError = Box<dyn Error + Send + Sync + 'static>; ··· 237 238 238 239 /// Platform-agnostic byte sink abstraction 239 240 pub struct ByteSink { 240 240 - inner: Box<dyn n0_future::Sink<Bytes, Error = StreamError>>, 241 241 + inner: Pin<Box<dyn n0_future::Sink<Bytes, Error = StreamError>>>, 241 242 } 242 243 243 244 impl ByteSink { ··· 247 248 S: n0_future::Sink<Bytes, Error = StreamError> + 'static, 248 249 { 249 250 Self { 250 250 - inner: Box::new(sink), 251 251 + inner: Box::pin(sink), 251 252 } 252 253 } 253 254 254 255 /// Convert into the inner boxed sink 255 255 - pub fn into_inner(self) -> Box<dyn n0_future::Sink<Bytes, Error = StreamError>> { 256 256 + pub fn into_inner(self) -> Pin<Box<dyn n0_future::Sink<Bytes, Error = StreamError>>> { 256 257 self.inner 257 258 } 258 259 }

+42 -6

crates/jacquard-common/src/websocket.rs

··· 4 4 use crate::stream::StreamError; 5 5 use bytes::Bytes; 6 6 use n0_future::Stream; 7 7 - use n0_future::stream::Boxed; 8 7 use std::borrow::Borrow; 9 8 use std::fmt::{self, Display}; 10 9 use std::future::Future; 11 10 use std::ops::Deref; 11 11 + use std::pin::Pin; 12 12 use url::Url; 13 13 14 14 /// UTF-8 validated bytes for WebSocket text messages ··· 282 282 } 283 283 284 284 /// WebSocket message stream 285 285 - pub struct WsStream(Boxed<Result<WsMessage, StreamError>>); 285 285 + #[cfg(not(target_arch = "wasm32"))] 286 286 + pub struct WsStream(Pin<Box<dyn Stream<Item = Result<WsMessage, StreamError>> + Send>>); 287 287 + 288 288 + /// WebSocket message stream 289 289 + #[cfg(target_arch = "wasm32")] 290 290 + pub struct WsStream(Pin<Box<dyn Stream<Item = Result<WsMessage, StreamError>>>>); 286 291 287 292 impl WsStream { 288 293 /// Create a new message stream ··· 304 309 } 305 310 306 311 /// Convert into the inner pinned boxed stream 307 307 - pub fn into_inner(self) -> Boxed<Result<WsMessage, StreamError>> { 312 312 + #[cfg(not(target_arch = "wasm32"))] 313 313 + pub fn into_inner(self) -> Pin<Box<dyn Stream<Item = Result<WsMessage, StreamError>> + Send>> { 314 314 + self.0 315 315 + } 316 316 + 317 317 + /// Convert into the inner pinned boxed stream 318 318 + #[cfg(target_arch = "wasm32")] 319 319 + pub fn into_inner(self) -> Pin<Box<dyn Stream<Item = Result<WsMessage, StreamError>>>> { 308 320 self.0 309 321 } 310 322 ··· 358 370 } 359 371 360 372 /// WebSocket message sink 361 361 - pub struct WsSink(Box<dyn n0_future::Sink<WsMessage, Error = StreamError>>); 373 373 + #[cfg(not(target_arch = "wasm32"))] 374 374 + pub struct WsSink(Pin<Box<dyn n0_future::Sink<WsMessage, Error = StreamError> + Send>>); 375 375 + 376 376 + /// WebSocket message sink 377 377 + #[cfg(target_arch = "wasm32")] 378 378 + pub struct WsSink(Pin<Box<dyn n0_future::Sink<WsMessage, Error = StreamError>>>); 362 379 363 380 impl WsSink { 364 381 /// Create a new message sink 382 382 + #[cfg(not(target_arch = "wasm32"))] 365 383 pub fn new<S>(sink: S) -> Self 366 384 where 367 385 S: n0_future::Sink<WsMessage, Error = StreamError> + Send + 'static, 368 386 { 369 369 - Self(Box::new(sink)) 387 387 + Self(Box::pin(sink)) 388 388 + } 389 389 + 390 390 + /// Create a new message sink 391 391 + #[cfg(target_arch = "wasm32")] 392 392 + pub fn new<S>(sink: S) -> Self 393 393 + where 394 394 + S: n0_future::Sink<WsMessage, Error = StreamError> + 'static, 395 395 + { 396 396 + Self(Box::pin(sink)) 397 397 + } 398 398 + 399 399 + /// Convert into the inner boxed sink 400 400 + #[cfg(not(target_arch = "wasm32"))] 401 401 + pub fn into_inner( 402 402 + self, 403 403 + ) -> Pin<Box<dyn n0_future::Sink<WsMessage, Error = StreamError> + Send>> { 404 404 + self.0 370 405 } 371 406 372 407 /// Convert into the inner boxed sink 373 373 - pub fn into_inner(self) -> Box<dyn n0_future::Sink<WsMessage, Error = StreamError>> { 408 408 + #[cfg(target_arch = "wasm32")] 409 409 + pub fn into_inner(self) -> Pin<Box<dyn n0_future::Sink<WsMessage, Error = StreamError>>> { 374 410 self.0 375 411 } 376 412 }

+146 -2

crates/jacquard-common/src/xrpc/subscription.rs

··· 3 3 //! This module defines traits and types for typed WebSocket subscriptions, 4 4 //! mirroring the request/response pattern used for HTTP XRPC endpoints. 5 5 6 6 + #[cfg(not(target_arch = "wasm32"))] 6 7 use n0_future::stream::Boxed; 8 8 + #[cfg(target_arch = "wasm32")] 9 9 + use n0_future::stream::BoxedLocal as Boxed; 7 10 use serde::{Deserialize, Serialize}; 8 11 use std::error::Error; 9 12 use std::future::Future; ··· 258 261 259 262 let (tx, rx) = self.connection.split(); 260 263 264 264 + #[cfg(not(target_arch = "wasm32"))] 261 265 let stream = match S::ENCODING { 262 266 MessageEncoding::Json => rx 263 267 .into_inner() ··· 269 273 .boxed(), 270 274 }; 271 275 276 276 + #[cfg(target_arch = "wasm32")] 277 277 + let stream = match S::ENCODING { 278 278 + MessageEncoding::Json => rx 279 279 + .into_inner() 280 280 + .filter_map(|msg| decode_json_msg::<S>(msg)) 281 281 + .boxed_local(), 282 282 + MessageEncoding::DagCbor => rx 283 283 + .into_inner() 284 284 + .filter_map(|msg| decode_cbor_msg::<S>(msg)) 285 285 + .boxed_local(), 286 286 + }; 287 287 + 272 288 (tx, stream) 273 289 } 274 290 ··· 288 304 serde_ipld_dagcbor::from_slice(bytes) 289 305 } 290 306 307 307 + #[cfg(not(target_arch = "wasm32"))] 291 308 let stream = match S::ENCODING { 292 309 MessageEncoding::Json => rx 293 310 .into_inner() ··· 343 360 .boxed(), 344 361 }; 345 362 363 363 + #[cfg(target_arch = "wasm32")] 364 364 + let stream = match S::ENCODING { 365 365 + MessageEncoding::Json => rx 366 366 + .into_inner() 367 367 + .filter_map(|msg_result| match msg_result { 368 368 + Ok(WsMessage::Text(text)) => Some( 369 369 + parse_msg(text.as_ref()) 370 370 + .map(|v| v.into_static()) 371 371 + .map_err(StreamError::decode), 372 372 + ), 373 373 + Ok(WsMessage::Binary(bytes)) => { 374 374 + #[cfg(feature = "zstd")] 375 375 + { 376 376 + match decompress_zstd(&bytes) { 377 377 + Ok(decompressed) => Some( 378 378 + parse_msg(&decompressed) 379 379 + .map(|v| v.into_static()) 380 380 + .map_err(StreamError::decode), 381 381 + ), 382 382 + Err(_) => Some( 383 383 + parse_msg(&bytes) 384 384 + .map(|v| v.into_static()) 385 385 + .map_err(StreamError::decode), 386 386 + ), 387 387 + } 388 388 + } 389 389 + #[cfg(not(feature = "zstd"))] 390 390 + { 391 391 + Some( 392 392 + parse_msg(&bytes) 393 393 + .map(|v| v.into_static()) 394 394 + .map_err(StreamError::decode), 395 395 + ) 396 396 + } 397 397 + } 398 398 + Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())), 399 399 + Err(e) => Some(Err(e)), 400 400 + }) 401 401 + .boxed_local(), 402 402 + MessageEncoding::DagCbor => rx 403 403 + .into_inner() 404 404 + .filter_map(|msg_result| match msg_result { 405 405 + Ok(WsMessage::Binary(bytes)) => Some( 406 406 + parse_cbor(&bytes) 407 407 + .map(|v| v.into_static()) 408 408 + .map_err(|e| StreamError::decode(crate::error::DecodeError::from(e))), 409 409 + ), 410 410 + Ok(WsMessage::Text(_)) => Some(Err(StreamError::wrong_message_format( 411 411 + "expected binary frame for CBOR, got text", 412 412 + ))), 413 413 + Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())), 414 414 + Err(e) => Some(Err(e)), 415 415 + }) 416 416 + .boxed_local(), 417 417 + }; 418 418 + 346 419 (tx, stream) 347 420 } 348 421 ··· 361 434 serde_ipld_dagcbor::from_slice(bytes) 362 435 } 363 436 437 437 + #[cfg(not(target_arch = "wasm32"))] 364 438 let stream = match S::ENCODING { 365 439 MessageEncoding::Json => rx 366 440 .into_inner() ··· 416 490 .boxed(), 417 491 }; 418 492 493 493 + #[cfg(target_arch = "wasm32")] 494 494 + let stream = match S::ENCODING { 495 495 + MessageEncoding::Json => rx 496 496 + .into_inner() 497 497 + .filter_map(|msg_result| match msg_result { 498 498 + Ok(WsMessage::Text(text)) => Some( 499 499 + parse_msg(text.as_ref()) 500 500 + .map(|v| v.into_static()) 501 501 + .map_err(StreamError::decode), 502 502 + ), 503 503 + Ok(WsMessage::Binary(bytes)) => { 504 504 + #[cfg(feature = "zstd")] 505 505 + { 506 506 + match decompress_zstd(&bytes) { 507 507 + Ok(decompressed) => Some( 508 508 + parse_msg(&decompressed) 509 509 + .map(|v| v.into_static()) 510 510 + .map_err(StreamError::decode), 511 511 + ), 512 512 + Err(_) => Some( 513 513 + parse_msg(&bytes) 514 514 + .map(|v| v.into_static()) 515 515 + .map_err(StreamError::decode), 516 516 + ), 517 517 + } 518 518 + } 519 519 + #[cfg(not(feature = "zstd"))] 520 520 + { 521 521 + Some( 522 522 + parse_msg(&bytes) 523 523 + .map(|v| v.into_static()) 524 524 + .map_err(StreamError::decode), 525 525 + ) 526 526 + } 527 527 + } 528 528 + Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())), 529 529 + Err(e) => Some(Err(e)), 530 530 + }) 531 531 + .boxed_local(), 532 532 + MessageEncoding::DagCbor => rx 533 533 + .into_inner() 534 534 + .filter_map(|msg_result| match msg_result { 535 535 + Ok(WsMessage::Binary(bytes)) => Some( 536 536 + parse_cbor(&bytes) 537 537 + .map(|v| v.into_static()) 538 538 + .map_err(|e| StreamError::decode(crate::error::DecodeError::from(e))), 539 539 + ), 540 540 + Ok(WsMessage::Text(_)) => Some(Err(StreamError::wrong_message_format( 541 541 + "expected binary frame for CBOR, got text", 542 542 + ))), 543 543 + Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())), 544 544 + Err(e) => Some(Err(e)), 545 545 + }) 546 546 + .boxed_local(), 547 547 + }; 548 548 + 419 549 (tx, stream) 420 550 } 421 551 ··· 442 572 // Put the raw stream back 443 573 *rx = raw_rx; 444 574 445 445 - match S::ENCODING { 575 575 + #[cfg(not(target_arch = "wasm32"))] 576 576 + let stream = match S::ENCODING { 446 577 MessageEncoding::Json => typed_rx_source 447 578 .into_inner() 448 579 .filter_map(|msg| decode_json_msg::<S>(msg)) ··· 451 582 .into_inner() 452 583 .filter_map(|msg| decode_cbor_msg::<S>(msg)) 453 584 .boxed(), 454 454 - } 585 585 + }; 586 586 + 587 587 + #[cfg(target_arch = "wasm32")] 588 588 + let stream = match S::ENCODING { 589 589 + MessageEncoding::Json => typed_rx_source 590 590 + .into_inner() 591 591 + .filter_map(|msg| decode_json_msg::<S>(msg)) 592 592 + .boxed_local(), 593 593 + MessageEncoding::DagCbor => typed_rx_source 594 594 + .into_inner() 595 595 + .filter_map(|msg| decode_cbor_msg::<S>(msg)) 596 596 + .boxed_local(), 597 597 + }; 598 598 + stream 455 599 } 456 600 } 457 601

+1

crates/jacquard/Cargo.toml

··· 47 47 "jacquard-api/streaming", 48 48 ] 49 49 websocket = ["jacquard-common/websocket"] 50 50 + zstd = ["jacquard-common/zstd"] 50 51 51 52 [[example]] 52 53 name = "oauth_timeline"

+4 -6

examples/subscribe_jetstream.rs

··· 85 85 let client = TungsteniteSubscriptionClient::from_base_uri(base_url); 86 86 87 87 // Subscribe with no filters (firehose mode) 88 88 - let mut params_builder = JetstreamParams::new(); 89 89 - 90 88 // Enable compression if zstd feature is available 91 89 #[cfg(feature = "zstd")] 92 92 - { 93 93 - params_builder = params_builder.compress(true); 94 94 - } 90 90 + let params = { JetstreamParams::new().compress(true).build() }; 95 91 96 96 - let params = params_builder.build(); 92 92 + #[cfg(not(feature = "zstd"))] 93 93 + let params = { JetstreamParams::new().build() }; 94 94 + 97 95 let stream = client.subscribe(&params).await.into_diagnostic()?; 98 96 99 97 println!("Connected! Streaming messages (Ctrl-C to stop)...\n");

+1 -1

justfile

··· 7 7 8 8 # Check that jacquard-common compiles for wasm32 9 9 check-wasm: 10 10 - cargo build --target wasm32-unknown-unknown -p jacquard-common --no-default-features 10 10 + cargo build --target wasm32-unknown-unknown -p jacquard-common --no-default-features --features websocket 11 11 12 12 # Run 'cargo run' on the project 13 13 run *ARGS: