feature flags for anything that uses tokio since it might now be available

Author: nossrannug

tonic/src/codec/decode.rs

@@ -210,25 +210,34 @@ impl StreamingInner {
             }
 
             let decode_buf = if let Some(encoding) = compression {
-                self.decompress_buf.clear();
-
-                if let Err(err) = decompress(
-                    CompressionSettings::new(encoding, buffer_settings.buffer_size),
-                    &mut self.buf,
-                    &mut self.decompress_buf,
-                    len,
-                ) {
-                    let message = if let Direction::Response(status) = self.direction {
-                        format!(
-                            "Error decompressing: {err}, while receiving response with status: {status}"
-                        )
-                    } else {
-                        format!("Error decompressing: {err}, while sending request")
-                    };
-                    return Err(Status::internal(message));
+                #[cfg(any(feature = "gzip", feature = "deflate", feature = "zstd"))]
+                {
+                    self.decompress_buf.clear();
+
+                    if let Err(err) = decompress(
+                        CompressionSettings::new(encoding, buffer_settings.buffer_size),
+                        &mut self.buf,
+                        &mut self.decompress_buf,
+                        len,
+                    ) {
+                        let message = if let Direction::Response(status) = self.direction {
+                            format!(
+                                "Error decompressing: {err}, while receiving response with status: {status}"
+                            )
+                        } else {
+                            format!("Error decompressing: {err}, while sending request")
+                        };
+                        return Err(Status::internal(message));
+                    }
+                    let decompressed_len = self.decompress_buf.len();
+                    DecodeBuf::new(&mut self.decompress_buf, decompressed_len)
+                }
+                #[cfg(not(any(feature = "gzip", feature = "deflate", feature = "zstd")))]
+                {
+                    // This branch is unreachable when no compression features are enabled
+                    // because CompressionEncoding has no variants
+                    unreachable!("Compression encoding without compression features")
                 }
-                let decompressed_len = self.decompress_buf.len();
-                DecodeBuf::new(&mut self.decompress_buf, decompressed_len)
             } else {
                 DecodeBuf::new(&mut self.buf, len)
             };

tonic/src/codec/encode.rs

@@ -12,6 +12,7 @@ use std::{
     pin::Pin,
     task::{ready, Context, Poll},
 };
+#[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
 use tokio::task::JoinHandle;
 use tokio_stream::{adapters::Fuse, Stream, StreamExt};
 
@@ -38,6 +39,7 @@ struct EncodedBytes<T, U> {
     buf: BytesMut,
     uncompression_buf: BytesMut,
     error: Option<Status>,
+    #[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
     #[pin]
     compression_task: Option<JoinHandle<Result<CompressionResult, Status>>>,
 }
@@ -74,6 +76,7 @@ impl<T: Encoder, U: Stream> EncodedBytes<T, U> {
             buf,
             uncompression_buf,
             error: None,
+            #[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
             compression_task: None,
         }
     }
@@ -113,7 +116,10 @@ where
         uncompression_buf: &mut BytesMut,
         compression_encoding: Option<CompressionEncoding>,
         max_message_size: Option<usize>,
-        compression_task: &mut Pin<&mut Option<JoinHandle<Result<CompressionResult, Status>>>>,
+        #[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
+        compression_task: &mut Pin<
+            &mut Option<JoinHandle<Result<CompressionResult, Status>>>,
+        >,
         buffer_settings: &BufferSettings,
     ) -> Result<bool, Status> {
         let compression_settings = compression_encoding
@@ -127,6 +133,8 @@ where
 
             let uncompressed_len = uncompression_buf.len();
 
+            // Check if we should use spawn_blocking (only when tokio is available)
+            #[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
             if let Some(spawn_threshold) = settings.spawn_blocking_threshold {
                 if uncompressed_len >= spawn_threshold
                     && uncompressed_len >= settings.compression_threshold
@@ -156,6 +164,7 @@ where
         Ok(false)
     }
 
+    #[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
     fn poll_compression_task(
         compression_task: &mut Pin<&mut Option<JoinHandle<Result<CompressionResult, Status>>>>,
         buf: &mut BytesMut,
@@ -226,6 +235,7 @@ where
             buf,
             uncompression_buf,
             error,
+            #[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
             mut compression_task,
         } = self.project();
         let buffer_settings = encoder.buffer_settings();
@@ -235,17 +245,20 @@ where
         }
 
         // Check if we have an in-flight compression task
-        match Self::poll_compression_task(
-            &mut compression_task,
-            buf,
-            *max_message_size,
-            &buffer_settings,
-            cx,
-        ) {
-            Poll::Ready(Some(result)) => return Poll::Ready(Some(result)),
-            Poll::Pending => return Poll::Pending,
-            Poll::Ready(None) => {
-                // Task completed, continue processing
+        #[cfg(any(feature = "transport", feature = "channel", feature = "server"))]
+        {
+            match Self::poll_compression_task(
+                &mut compression_task,
+                buf,
+                *max_message_size,
+                &buffer_settings,
+                cx,
+            ) {
+                Poll::Ready(Some(result)) => return Poll::Ready(Some(result)),
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(None) => {
+                    // Task completed, continue processing
+                }
             }
         }
 
@@ -268,32 +281,53 @@ where
                         uncompression_buf,
                         *compression_encoding,
                         *max_message_size,
+                        #[cfg(any(
+                            feature = "transport",
+                            feature = "channel",
+                            feature = "server"
+                        ))]
                         &mut compression_task,
                         &buffer_settings,
                     ) {
                         Ok(true) => {
-                            // We just spawned/armed the blocking compression task.
-                            // Poll it once right away so it can capture our waker.
-                            match Self::poll_compression_task(
-                                &mut compression_task,
-                                buf,
-                                *max_message_size,
-                                &buffer_settings,
-                                cx,
-                            ) {
-                                Poll::Ready(Some(result)) => {
-                                    return Poll::Ready(Some(result));
-                                }
-                                Poll::Ready(None) => {
-                                    if buf.len() >= buffer_settings.yield_threshold {
-                                        return Poll::Ready(Some(Ok(buf
-                                            .split_to(buf.len())
-                                            .freeze())));
+                            #[cfg(any(
+                                feature = "transport",
+                                feature = "channel",
+                                feature = "server"
+                            ))]
+                            {
+                                // We just spawned/armed the blocking compression task.
+                                // Poll it once right away so it can capture our waker.
+                                match Self::poll_compression_task(
+                                    &mut compression_task,
+                                    buf,
+                                    *max_message_size,
+                                    &buffer_settings,
+                                    cx,
+                                ) {
+                                    Poll::Ready(Some(result)) => {
+                                        return Poll::Ready(Some(result));
+                                    }
+                                    Poll::Ready(None) => {
+                                        if buf.len() >= buffer_settings.yield_threshold {
+                                            return Poll::Ready(Some(Ok(buf
+                                                .split_to(buf.len())
+                                                .freeze())));
+                                        }
+                                    }
+                                    Poll::Pending => {
+                                        return Poll::Pending;
                                     }
                                 }
-                                Poll::Pending => {
-                                    return Poll::Pending;
-                                }
+                            }
+                            #[cfg(not(any(
+                                feature = "transport",
+                                feature = "channel",
+                                feature = "server"
+                            )))]
+                            {
+                                // This shouldn't happen when tokio is not available
+                                unreachable!("spawn_blocking returned true without tokio")
                             }
                         }
                         Ok(false) => {