disabling streaming support by default for now

crates/jacquard-api/Cargo.toml

···
 lexicon_community = ["community_lexicon", "my_skylights", "com_whtwnd", "social_psky", "blue_linkat", "minimal"]
 ufos = [ "moe_karashiiro", "dev_regnault", "buzz_bookhive", "uk_ewancroft",  "win_tomo_x", "dev_fudgeu", "dev_ocbwoy3", "blog_pckt",  "net_bnewbold", "club_stellz", "dev_baileytownsend", "net_aftertheinter", "org_devcon","org_robocracy", "garden_lexicon", "social_clippr", "social_grain", "social_pmsky", "com_crabdance", "app_blebbit", "app_ocho", "uk_skyblur", "us_polhem", "blue_atplane", "net_mmatt", "minimal", "app_rocksky"]
 
+streaming = []
+
 # --- generated ---
 # Generated namespace features
 app_blebbit = []

crates/jacquard-api/src/com_atproto/label.rs

···
 // Any manual changes will be overwritten on the next regeneration.
 
 pub mod query_labels;
+#[cfg(feature = "streaming")]
 pub mod subscribe_labels;
 
 /// Metadata tag on an atproto resource (eg, repo or record).
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Label<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct LabelValueDefinition<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct LabelValueDefinitionStrings<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    Default
+    Default,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct SelfLabel<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct SelfLabels<'a> {
     #[serde(borrow)]
     pub values: Vec<crate::com_atproto::label::SelfLabel<'a>>,
-}
+}

crates/jacquard-api/src/com_atproto/sync.rs

···
 pub mod list_repos_by_collection;
 pub mod notify_of_update;
 pub mod request_crawl;
+#[cfg(feature = "streaming")]
 pub mod subscribe_repos;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
···
             HostStatus::Other(v) => HostStatus::Other(v.into_static()),
         }
     }
-}
+}

crates/jacquard-common/Cargo.toml

···
 exclude.workspace = true
 license.workspace = true
 
+[features]
+default = ["service-auth", "reqwest-client", "crypto"]
+crypto = []
+crypto-ed25519 = ["crypto", "dep:ed25519-dalek"]
+crypto-k256 = ["crypto", "dep:k256", "k256/ecdsa"]
+crypto-p256 = ["crypto", "dep:p256", "p256/ecdsa"]
+service-auth = ["crypto-k256", "crypto-p256", "dep:signature"]
+reqwest-client = ["dep:reqwest"]
+tracing = ["dep:tracing"]
+streaming = ["n0-future", "futures"]
+websocket = ["streaming", "tokio-tungstenite-wasm"]
 
 [dependencies]
 trait-variant.workspace = true
···
 reqwest = { workspace = true, optional = true, features = [ "http2", "system-proxy", "rustls-tls"] }
 tokio-util = { version = "0.7.16", features = ["io"] }
 
-[features]
-default = ["service-auth", "reqwest-client", "crypto", "websocket"]
-crypto = []
-crypto-ed25519 = ["crypto", "dep:ed25519-dalek"]
-crypto-k256 = ["crypto", "dep:k256", "k256/ecdsa"]
-crypto-p256 = ["crypto", "dep:p256", "p256/ecdsa"]
-service-auth = ["crypto-k256", "crypto-p256", "dep:signature"]
-reqwest-client = ["dep:reqwest"]
-tracing = ["dep:tracing"]
-streaming = ["n0-future", "futures"]
-websocket = ["streaming", "tokio-tungstenite-wasm"]
+
 
 [dependencies.ed25519-dalek]
 version = "2"

crates/jacquard-common/src/stream.rs

···
 
 impl ByteStream {
     /// Create a new byte stream from any compatible stream
-    #[cfg(not(target_arch = "wasm32"))]
     pub fn new<S>(stream: S) -> Self
     where
         S: n0_future::Stream<Item = Result<Bytes, StreamError>> + Unpin + Send + 'static,
-    {
-        Self {
-            inner: Box::pin(stream),
-        }
-    }
-
-    /// Create a new byte stream from any compatible stream
-    #[cfg(target_arch = "wasm32")]
-    pub fn new<S>(stream: S) -> Self
-    where
-        S: n0_future::Stream<Item = Result<Bytes, StreamError>> + Unpin + 'static,
     {
         Self {
             inner: Box::pin(stream),

crates/jacquard-common/src/types/blob.rs

···
-use crate::{
-    CowStr, IntoStatic,
-    types::cid::{Cid, CidLink},
-};
-#[allow(unused)]
+use crate::{CowStr, IntoStatic, types::cid::CidLink};
 use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error};
 use smol_str::ToSmolStr;
 use std::convert::Infallible;
-#[allow(unused)]
-use std::{
-    borrow::Cow,
-    fmt,
-    hash::{Hash, Hasher},
-    ops::Deref,
-    str::FromStr,
-};
+use std::{fmt, hash::Hash, ops::Deref, str::FromStr};
 
 /// Blob reference for binary data in AT Protocol
 ///

crates/jacquard-common/src/xrpc.rs

···
 #[cfg(feature = "websocket")]
 pub mod subscription;
 
-#[cfg(feature = "websocket")]
-pub use subscription::{
-    BasicSubscriptionClient, MessageEncoding, SubscriptionCall, SubscriptionClient,
-    SubscriptionEndpoint, SubscriptionExt, SubscriptionOptions, SubscriptionResp,
-    SubscriptionStream, TungsteniteSubscriptionClient, XrpcSubscription,
-};
-
+#[cfg(feature = "streaming")]
+use crate::StreamError;
+use crate::http_client::HttpClient;
+#[cfg(feature = "streaming")]
+use crate::http_client::HttpClientExt;
+use crate::types::value::Data;
+use crate::{AuthorizationToken, error::AuthError};
+use crate::{CowStr, error::XrpcResult};
+use crate::{IntoStatic, error::DecodeError};
+use crate::{error::TransportError, types::value::RawData};
 use bytes::Bytes;
 use http::{
     HeaderName, HeaderValue, Request, StatusCode,
···
 use smol_str::SmolStr;
 use std::fmt::{self, Debug};
 use std::{error::Error, marker::PhantomData};
+#[cfg(feature = "websocket")]
+pub use subscription::{
+    BasicSubscriptionClient, MessageEncoding, SubscriptionCall, SubscriptionClient,
+    SubscriptionEndpoint, SubscriptionExt, SubscriptionOptions, SubscriptionResp,
+    SubscriptionStream, TungsteniteSubscriptionClient, XrpcSubscription,
+};
 use url::Url;
-
-use crate::http_client::{HttpClient, HttpClientExt};
-use crate::types::value::Data;
-use crate::{AuthorizationToken, error::AuthError};
-use crate::{CowStr, error::XrpcResult};
-use crate::{IntoStatic, error::DecodeError};
-#[cfg(feature = "streaming")]
-use crate::StreamError;
-use crate::{error::TransportError, types::value::RawData};
 
 /// Error type for encoding XRPC requests
 #[derive(Debug, thiserror::Error, miette::Diagnostic)]
···
     where
         R: XrpcRequest + Send + Sync,
         <R as XrpcRequest>::Response: Send + Sync;
-
 }
 
 /// Stateful XRPC streaming client trait
···
     fn stream<S>(
         &self,
         stream: XrpcProcedureSend<S::Frame<'static>>,
-    ) -> impl Future<Output = Result<XrpcResponseStream<<<S as XrpcProcedureStream>::Response as XrpcStreamResp>::Frame<'static>>, StreamError>>
+    ) -> impl Future<
+        Output = Result<
+            XrpcResponseStream<
+                <<S as XrpcProcedureStream>::Response as XrpcStreamResp>::Frame<'static>,
+            >,
+            StreamError,
+        >,
+    >
     where
         S: XrpcProcedureStream + 'static,
         <<S as XrpcProcedureStream>::Response as XrpcStreamResp>::Frame<'static>: XrpcStreamResp,
···
     fn stream<S>(
         &self,
         stream: XrpcProcedureSend<S::Frame<'static>>,
-    ) -> impl Future<Output = Result<XrpcResponseStream<<<S as XrpcProcedureStream>::Response as XrpcStreamResp>::Frame<'static>>, StreamError>>
+    ) -> impl Future<
+        Output = Result<
+            XrpcResponseStream<
+                <<S as XrpcProcedureStream>::Response as XrpcStreamResp>::Frame<'static>,
+            >,
+            StreamError,
+        >,
+    >
     where
         S: XrpcProcedureStream + 'static,
         <<S as XrpcProcedureStream>::Response as XrpcStreamResp>::Frame<'static>: XrpcStreamResp;

crates/jacquard-oauth/Cargo.toml

···
 exclude.workspace = true
 license.workspace = true
 
+
+[features]
+default = []
+loopback = ["dep:rouille"]
+browser-open = ["dep:webbrowser"]
+tracing = ["dep:tracing"]
+websocket = ["jacquard-common/websocket"]
+streaming = ["jacquard-common/streaming", "dep:n0-future"]
+
 [dependencies]
 jacquard-common = { version = "0.5", path = "../jacquard-common", features = ["reqwest-client"] }
 jacquard-identity = { version = "0.5", path = "../jacquard-identity" }
···
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 tokio = { workspace = true, features = ["rt", "net", "time"] }
 rouille = { version = "3.6.2", optional = true }
-
-[features]
-default = []
-loopback = ["dep:rouille"]
-browser-open = ["dep:webbrowser"]
-tracing = ["dep:tracing"]
-websocket = ["jacquard-common/websocket"]
-streaming = ["jacquard-common/streaming", "dep:n0-future"]

crates/jacquard/Cargo.toml

···
 license.workspace = true
 
 [features]
-default = ["api_full", "dns", "loopback", "derive", "streaming"]
+default = ["api_full", "dns", "loopback", "derive"]
 derive = ["dep:jacquard-derive"]
 # Minimal API bindings
 api = ["jacquard-api/minimal"]
···
 ]
 dns = ["jacquard-identity/dns"]
 streaming = [
-    "jacquard-common/streaming",
+    "jacquard-common/websocket",
     "jacquard-oauth/streaming",
     "jacquard-identity/streaming",
     "dep:n0-future",
-    "dep:futures"
+    "dep:futures",
+    "jacquard-api/streaming",
 ]
 websocket = ["jacquard-common/websocket"]
 

crates/jacquard/src/client/credential_session.rs

···
         let mut opts = self.options.read().await.clone();
         opts.auth = self.access_token().await;
 
-        let mut url = base_uri;
+        let mut url = base_uri.clone();
         let mut path = url.path().trim_end_matches('/').to_owned();
         path.push_str("/xrpc/");
         path.push_str(<Str::Request as jacquard_common::xrpc::XrpcRequest>::NSID);
···
         let body_stream =
             jacquard_common::stream::ByteStream::new(stream.0.map_ok(|f| f.buffer).boxed());
 
+        // Clone the stream for potential retry
+        let (body1, body2) = body_stream.tee();
+
         let response = self
             .client
-            .send_http_bidirectional(parts.clone(), body_stream.into_inner())
+            .send_http_bidirectional(parts.clone(), body1.into_inner())
             .await
             .map_err(StreamError::transport)?;
 
···
                 .map_err(|e| StreamError::protocol(e.to_string()))?
                 .into_parts();
 
-            // Can't retry with the same stream - it's been consumed
-            // This is a limitation of streaming upload with auth refresh
-            return Err(StreamError::protocol("Authentication failed on streaming upload and stream cannot be retried".to_string()));
+            // Retry with the second half of the cloned stream
+            let response = self
+                .client
+                .send_http_bidirectional(parts, body2.into_inner())
+                .await
+                .map_err(StreamError::transport)?;
+            let (resp_parts, resp_body) = response.into_parts();
+            Ok(jacquard_common::xrpc::streaming::XrpcResponseStream::from_typed_parts(
+                resp_parts, resp_body,
+            ))
+        } else {
+            Ok(jacquard_common::xrpc::streaming::XrpcResponseStream::from_typed_parts(
+                resp_parts, resp_body,
+            ))
         }
-
-        Ok(jacquard_common::xrpc::streaming::XrpcResponseStream::from_typed_parts(
-            resp_parts, resp_body,
-        ))
     }
 }
 

crates/jacquard/src/lib.rs

···
 pub mod client;
 
 #[cfg(feature = "streaming")]
+/// Experimental streaming endpoints
 pub mod streaming;
 
 pub use common::*;