An alternative approach to removing `UnknownBufferType`.

This is a potential alternative to #359. This PR is based on #359.

This approach opts for a dynamically checked sample type approach with
the aim of minimising compile time and binary size.

You can read more discussion on this [here](https://github.com/RustAudio/cpal/pull/359#issuecomment-575931461)

Implemented backends:

- [x] null
- [x] ALSA
- [ ] CoreAudio
- [ ] WASAPI
- [ ] ASIO
- [ ] Emscripten

examples/beep.rs

@@ -19,27 +19,15 @@ fn main() -> Result<(), anyhow::Error> {
         (sample_clock * 440.0 * 2.0 * 3.141592 / sample_rate).sin()
     };
 
-    let err_fn = |err| {
+    let err_fn = |err| eprintln!("an error occurred on stream: {}", err);
-        eprintln!("an error occurred on stream: {}", err);
+
+    let data_fn = move |data: &mut cpal::Data| match data.sample_format() {
+        cpal::SampleFormat::F32 => write_data::<f32>(data, channels, &mut next_value),
+        cpal::SampleFormat::I16 => write_data::<i16>(data, channels, &mut next_value),
+        cpal::SampleFormat::U16 => write_data::<u16>(data, channels, &mut next_value),
     };
 
-    let stream = match format.data_type {
+    let stream = device.build_output_stream(&format, data_fn, err_fn)?;
-        cpal::SampleFormat::F32 => device.build_output_stream(
-            &format,
-            move |mut data| write_data::<f32>(&mut *data, channels, &mut next_value),
-            err_fn,
-        ),
-        cpal::SampleFormat::I16 => device.build_output_stream(
-            &format,
-            move |mut data| write_data::<i16>(&mut *data, channels, &mut next_value),
-            err_fn,
-        ),
-        cpal::SampleFormat::U16 => device.build_output_stream(
-            &format,
-            move |mut data| write_data::<u16>(&mut *data, channels, &mut next_value),
-            err_fn,
-        ),
-    }?;
 
     stream.play()?;
 
@@ -48,10 +36,11 @@ fn main() -> Result<(), anyhow::Error> {
     Ok(())
 }
 
-fn write_data<T>(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32)
+fn write_data<T>(output: &mut cpal::Data, channels: usize, next_sample: &mut dyn FnMut() -> f32)
 where
     T: cpal::Sample,
 {
+    let output = output.as_slice_mut::<T>().expect("unexpected sample type");
     for frame in output.chunks_mut(channels) {
         let value: T = cpal::Sample::from::<f32>(&next_sample());
         for sample in frame.iter_mut() {

examples/feedback.rs

@@ -47,13 +47,10 @@ fn main() -> Result<(), anyhow::Error> {
         producer.push(0.0).unwrap();
     }
 
-    // Build streams.
+    let input_data_fn = move |data: &cpal::Data| {
-    println!("Attempting to build both streams with `{:?}`.", format);
-    let input_stream = input_device.build_input_stream(
-        &format,
-        move |data| {
         let mut output_fell_behind = false;
-            for &sample in data.iter() {
+        let data = data.as_slice::<f32>().expect("unexpected sample type");
+        for &sample in data {
             if producer.push(sample).is_err() {
                 output_fell_behind = true;
             }
@@ -61,16 +58,12 @@ fn main() -> Result<(), anyhow::Error> {
         if output_fell_behind {
             eprintln!("output stream fell behind: try increasing latency");
         }
-        },
+    };
-        |err| {
+
-            eprintln!("an error occurred on stream: {}", err);
+    let output_data_fn = move |data: &mut cpal::Data| {
-        },
-    )?;
-    let output_stream = output_device.build_output_stream(
-        &format,
-        move |mut data| {
         let mut input_fell_behind = None;
-            for sample in data.iter_mut() {
+        let data = data.as_slice_mut::<f32>().expect("unexpected sample type");
+        for sample in data {
             *sample = match consumer.pop() {
                 Ok(s) => s,
                 Err(err) => {
@@ -82,11 +75,12 @@ fn main() -> Result<(), anyhow::Error> {
         if let Some(err) = input_fell_behind {
             eprintln!("input stream fell behind: {:?}: try increasing latency", err);
         }
-        },
+    };
-        move |err| {
+
-            eprintln!("an error occurred on output stream: {}", err);
+    // Build streams.
-        },
+    println!("Attempting to build both streams with `{:?}`.", format);
-    )?;
+    let input_stream = input_device.build_input_stream(&format, input_data_fn, err_fn)?;
+    let output_stream = output_device.build_output_stream(&format, output_data_fn, err_fn)?;
     println!("Successfully built streams.");
 
     // Play the streams.
@@ -105,3 +99,7 @@ fn main() -> Result<(), anyhow::Error> {
     println!("Done!");
     Ok(())
 }
+
+fn err_fn(err: cpal::StreamError) {
+    eprintln!("an error occurred on stream: {}", err);
+}

examples/record_wav.rs

@@ -40,23 +40,15 @@ fn main() -> Result<(), anyhow::Error> {
         eprintln!("an error occurred on stream: {}", err);
     };
 
-    let stream = match format.data_type {
+    let data_fn = move |data: &cpal::Data| {
-        cpal::SampleFormat::F32 => device.build_input_stream(
+        match data.sample_format() {
-            &format,
+            cpal::SampleFormat::F32 => write_input_data::<f32, f32>(data, &writer_2),
-            move |data| write_input_data::<f32, f32>(&*data, &writer_2),
+            cpal::SampleFormat::I16 => write_input_data::<i16, i16>(data, &writer_2),
-            err_fn,
+            cpal::SampleFormat::U16 => write_input_data::<u16, i16>(data, &writer_2),
-        ),
+        }
-        cpal::SampleFormat::I16 => device.build_input_stream(
+    };
-            &format,
+
-            move |data| write_input_data::<i16, i16>(&*data, &writer_2),
+    let stream = device.build_input_stream(&format, data_fn, err_fn)?;
-            err_fn,
-        ),
-        cpal::SampleFormat::U16 => device.build_input_stream(
-            &format,
-            move |data| write_input_data::<u16, i16>(&*data, &writer_2),
-            err_fn,
-        ),
-    }?;
 
     stream.play()?;
 
@@ -87,11 +79,12 @@ fn wav_spec_from_format(format: &cpal::Format) -> hound::WavSpec {
 
 type WavWriterHandle = Arc<Mutex<Option<hound::WavWriter<BufWriter<File>>>>>;
 
-fn write_input_data<T, U>(input: &[T], writer: &WavWriterHandle)
+fn write_input_data<T, U>(input: &cpal::Data, writer: &WavWriterHandle)
 where
     T: cpal::Sample,
     U: cpal::Sample + hound::Sample,
 {
+    let input = input.as_slice::<T>().expect("unexpected sample format");
     if let Ok(mut guard) = writer.try_lock() {
         if let Some(writer) = guard.as_mut() {
             for &sample in input.iter() {

src/host/alsa/mod.rs

@@ -5,15 +5,13 @@ use crate::{
     BackendSpecificError,
     BuildStreamError,
     ChannelCount,
+    Data,
     DefaultFormatError,
     DeviceNameError,
     DevicesError,
     Format,
-    InputData,
-    OutputData,
     PauseStreamError,
     PlayStreamError,
-    Sample,
     SampleFormat,
     SampleRate,
     StreamError,
@@ -90,37 +88,31 @@ impl DeviceTrait for Device {
         Device::default_output_format(self)
     }
 
-    fn build_input_stream<T, D, E>(
+    fn build_input_stream<D, E>(
         &self,
         format: &Format,
         data_callback: D,
         error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        T: Sample,
+        D: FnMut(&Data) + Send + 'static,
-        D: FnMut(InputData<T>) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
-        // TODO: Consider removing `data_type` field from `Format` and removing this.
-        assert_eq!(format.data_type, T::FORMAT, "sample format mismatch");
         let stream_inner = self.build_stream_inner(format, alsa::SND_PCM_STREAM_CAPTURE)?;
         let stream = Stream::new_input(Arc::new(stream_inner), data_callback, error_callback);
         Ok(stream)
     }
 
-    fn build_output_stream<T, D, E>(
+    fn build_output_stream<D, E>(
         &self,
         format: &Format,
         data_callback: D,
         error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        T: Sample,
+        D: FnMut(&mut Data) + Send + 'static,
-        D: FnMut(OutputData<T>) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
-        // TODO: Consider removing `data_type` field from `Format` and removing this.
-        assert_eq!(format.data_type, T::FORMAT, "sample format mismatch");
         let stream_inner = self.build_stream_inner(format, alsa::SND_PCM_STREAM_PLAYBACK)?;
         let stream = Stream::new_output(Arc::new(stream_inner), data_callback, error_callback);
         Ok(stream)
@@ -564,14 +556,12 @@ struct StreamWorkerContext {
     buffer: Vec<u8>,
 }
 
-fn input_stream_worker<T>(
+fn input_stream_worker(
     rx: TriggerReceiver,
     stream: &StreamInner,
-    data_callback: &mut (dyn FnMut(InputData<T>) + Send + 'static),
+    data_callback: &mut (dyn FnMut(&Data) + Send + 'static),
     error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
-) where
+) {
-    T: Sample,
-{
     let mut ctxt = StreamWorkerContext::default();
     loop {
         match poll_descriptors_and_prepare_buffer(&rx, stream, &mut ctxt, error_callback) {
@@ -583,7 +573,7 @@ fn input_stream_worker<T>(
                     StreamType::Input,
                     "expected input stream, but polling descriptors indicated output",
                 );
-                process_input::<T>(
+                process_input(
                     stream,
                     &mut ctxt.buffer,
                     available_frames,
@@ -595,14 +585,12 @@ fn input_stream_worker<T>(
     }
 }
 
-fn output_stream_worker<T>(
+fn output_stream_worker(
     rx: TriggerReceiver,
     stream: &StreamInner,
-    data_callback: &mut (dyn FnMut(OutputData<T>) + Send + 'static),
+    data_callback: &mut (dyn FnMut(&mut Data) + Send + 'static),
     error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
-) where
+) {
-    T: Sample,
-{
     let mut ctxt = StreamWorkerContext::default();
     loop {
         match poll_descriptors_and_prepare_buffer(&rx, stream, &mut ctxt, error_callback) {
@@ -614,7 +602,7 @@ fn output_stream_worker<T>(
                     StreamType::Output,
                     "expected output stream, but polling descriptors indicated input",
                 );
-                process_output::<T>(
+                process_output(
                     stream,
                     &mut ctxt.buffer,
                     available_frames,
@@ -729,15 +717,13 @@ fn poll_descriptors_and_prepare_buffer(
 }
 
 // Read input data from ALSA and deliver it to the user.
-fn process_input<T>(
+fn process_input(
     stream: &StreamInner,
     buffer: &mut [u8],
     available_frames: usize,
-    data_callback: &mut (dyn FnMut(InputData<T>) + Send + 'static),
+    data_callback: &mut (dyn FnMut(&Data) + Send + 'static),
     error_callback: &mut dyn FnMut(StreamError),
-) where
+) {
-    T: Sample,
-{
     let result = unsafe {
         alsa::snd_pcm_readi(
             stream.channel,
@@ -750,28 +736,34 @@ fn process_input<T>(
         error_callback(BackendSpecificError { description }.into());
         return;
     }
-    let buffer = unsafe { cast_input_buffer::<T>(buffer) };
+    let sample_format = stream.sample_format;
-    let input_data = InputData { buffer };
+    let data = buffer.as_mut_ptr() as *mut ();
-    data_callback(input_data);
+    let len = buffer.len() / sample_format.sample_size();
+    let data = unsafe {
+        Data::from_parts(data, len, sample_format)
+    };
+    data_callback(&data);
 }
 
 // Request data from the user's function and write it via ALSA.
 //
 // Returns `true`
-fn process_output<T>(
+fn process_output(
     stream: &StreamInner,
     buffer: &mut [u8],
     available_frames: usize,
-    data_callback: &mut (dyn FnMut(OutputData<T>) + Send + 'static),
+    data_callback: &mut (dyn FnMut(&mut Data) + Send + 'static),
     error_callback: &mut dyn FnMut(StreamError),
-) where
+) {
-    T: Sample,
-{
     {
         // We're now sure that we're ready to write data.
-        let buffer = unsafe { cast_output_buffer::<T>(buffer) };
+        let sample_format = stream.sample_format;
-        let output_data = OutputData { buffer };
+        let data = buffer.as_mut_ptr() as *mut ();
-        data_callback(output_data);
+        let len = buffer.len() / sample_format.sample_size();
+        let mut data = unsafe {
+            Data::from_parts(data, len, sample_format)
+        };
+        data_callback(&mut data);
     }
     loop {
         let result = unsafe {
@@ -805,21 +797,20 @@ fn process_output<T>(
 }
 
 impl Stream {
-    fn new_input<T, D, E>(
+    fn new_input<D, E>(
         inner: Arc<StreamInner>,
         mut data_callback: D,
         mut error_callback: E,
     ) -> Stream
     where
-        T: Sample,
+        D: FnMut(&Data) + Send + 'static,
-        D: FnMut(InputData<T>) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
         let (tx, rx) = trigger();
         // Clone the handle for passing into worker thread.
         let stream = inner.clone();
         let thread = thread::spawn(move || {
-            input_stream_worker::<T>(rx, &*stream, &mut data_callback, &mut error_callback);
+            input_stream_worker(rx, &*stream, &mut data_callback, &mut error_callback);
         });
         Stream {
             thread: Some(thread),
@@ -828,21 +819,20 @@ impl Stream {
         }
     }
 
-    fn new_output<T, D, E>(
+    fn new_output<D, E>(
         inner: Arc<StreamInner>,
         mut data_callback: D,
         mut error_callback: E,
     ) -> Stream
     where
-        T: Sample,
+        D: FnMut(&mut Data) + Send + 'static,
-        D: FnMut(OutputData<T>) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
         let (tx, rx) = trigger();
         // Clone the handle for passing into worker thread.
         let stream = inner.clone();
         let thread = thread::spawn(move || {
-            output_stream_worker::<T>(rx, &*stream, &mut data_callback, &mut error_callback);
+            output_stream_worker(rx, &*stream, &mut data_callback, &mut error_callback);
         });
         Stream {
             thread: Some(thread),
@@ -1080,17 +1070,3 @@ fn check_errors(err: libc::c_int) -> Result<(), String> {
 
     Ok(())
 }
-
-/// Cast a byte slice into a (immutable) slice of desired type.
-/// Safety: it's up to the caller to ensure that the input slice has valid bit representations.
-unsafe fn cast_input_buffer<T>(v: &[u8]) -> &[T] {
-    debug_assert!(v.len() % std::mem::size_of::<T>() == 0);
-    std::slice::from_raw_parts(v.as_ptr() as *const T, v.len() / std::mem::size_of::<T>())
-}
-
-/// Cast a byte slice into a mutable slice of desired type.
-/// Safety: it's up to the caller to ensure that the input slice has valid bit representations.
-unsafe fn cast_output_buffer<T>(v: &mut [u8]) -> &mut [T] {
-    debug_assert!(v.len() % std::mem::size_of::<T>() == 0);
-    std::slice::from_raw_parts_mut(v.as_mut_ptr() as *mut T, v.len() / std::mem::size_of::<T>())
-}

src/host/null/mod.rs

@@ -1,11 +1,10 @@
 use crate::{
     BuildStreamError,
+    Data,
     DefaultFormatError,
     DevicesError,
     DeviceNameError,
     Format,
-    InputData,
-    OutputData,
     PauseStreamError,
     PlayStreamError,
     StreamError,
@@ -71,28 +70,28 @@ impl DeviceTrait for Device {
         unimplemented!()
     }
 
-    fn build_input_stream<T, D, E>(
+    fn build_input_stream<D, E>(
         &self,
         _format: &Format,
         _data_callback: D,
         _error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(InputData<T>) + Send + 'static,
+        D: FnMut(&Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
         unimplemented!()
     }
 
     /// Create an output stream.
-    fn build_output_stream<T, D, E>(
+    fn build_output_stream<D, E>(
         &self,
         _format: &Format,
         _data_callback: D,
         _error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(OutputData<T>) + Send + 'static,
+        D: FnMut(&mut Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
         unimplemented!()

src/lib.rs

@@ -152,7 +152,6 @@ pub use platform::{
     HostId, Stream, SupportedInputFormats, SupportedOutputFormats,
 };
 pub use samples_formats::{Sample, SampleFormat};
-use std::ops::{Deref, DerefMut};
 
 mod error;
 mod host;
@@ -193,34 +192,90 @@ pub struct SupportedFormat {
     pub data_type: SampleFormat,
 }
 
-/// Represents a buffer containing audio data that may be read.
+/// Represents a buffer of audio data, delivered via a user's stream data callback function.
 ///
-/// This struct implements the `Deref` trait targeting `[T]`. Therefore this buffer can be read the
+/// Input stream callbacks receive `&Data`, while output stream callbacks expect `&mut Data`.
-/// same way as reading from a `Vec` or any other kind of Rust array.
-// TODO: explain audio stuff in general
-// TODO: Consider making this an `enum` with `Interleaved` and `NonInterleaved` variants.
 #[derive(Debug)]
-pub struct InputData<'a, T: 'a>
+pub struct Data {
-where
+    data: *mut (),
-    T: Sample,
+    len: usize,
-{
+    sample_format: SampleFormat,
-    buffer: &'a [T],
 }
 
-/// Represents a buffer that must be filled with audio data. The buffer in unfilled state may
+impl Data {
-/// contain garbage values.
+    // Internal constructor for host implementations to use.
-///
+    pub(crate) unsafe fn from_parts(
-/// This struct implements the `Deref` and `DerefMut` traits to `[T]`. Therefore writing to this
+        data: *mut (),
-/// buffer is done in the same way as writing to a `Vec` or any other kind of Rust array.
+        len: usize,
-// TODO: explain audio stuff in general
+        sample_format: SampleFormat,
-// TODO: Consider making this an `enum` with `Interleaved` and `NonInterleaved` variants.
+    ) -> Self {
-#[must_use]
+        Data { data, len, sample_format }
-#[derive(Debug)]
+    }
-pub struct OutputData<'a, T: 'a>
+
-where
+    /// The sample format of the internal audio data.
+    pub fn sample_format(&self) -> SampleFormat {
+        self.sample_format
+    }
+
+    /// The full length of the buffer in samples.
+    ///
+    /// The returned length is the same length as the slice of type `T` that would be returned via
+    /// `as_slice` given a sample type that matches the inner sample format.
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    /// The raw slice of memory representing the underlying audio data as a slice of bytes.
+    ///
+    /// It is up to the user to interprate the slice of memory based on `Data::sample_format`.
+    pub fn bytes(&self) -> &[u8] {
+        let len = self.len * self.sample_format.sample_size();
+        unsafe {
+            std::slice::from_raw_parts(self.data as *const u8, len)
+        }
+    }
+
+    /// The raw slice of memory representing the underlying audio data as a slice of bytes.
+    ///
+    /// It is up to the user to interprate the slice of memory based on `Data::sample_format`.
+    pub fn bytes_mut(&mut self) -> &mut [u8] {
+        let len = self.len * self.sample_format.sample_size();
+        unsafe {
+            std::slice::from_raw_parts_mut(self.data as *mut u8, len)
+        }
+    }
+
+    /// Access the data as a slice of sample type `T`.
+    ///
+    /// Returns `None` if the sample type does not match the expected sample format.
+    pub fn as_slice<T>(&self) -> Option<&[T]>
+    where
         T: Sample,
-{
+    {
-    buffer: &'a mut [T],
+        if T::FORMAT == self.sample_format {
+            unsafe {
+                Some(std::slice::from_raw_parts(self.data as *const T, self.len))
+            }
+        } else {
+            None
+        }
+    }
+
+    /// Access the data as a slice of sample type `T`.
+    ///
+    /// Returns `None` if the sample type does not match the expected sample format.
+    pub fn as_slice_mut<T>(&mut self) -> Option<&mut [T]>
+    where
+        T: Sample,
+    {
+        if T::FORMAT == self.sample_format {
+            unsafe {
+                Some(std::slice::from_raw_parts_mut(self.data as *mut T, self.len))
+            }
+        } else {
+            None
+        }
+    }
 }
 
 impl SupportedFormat {
@@ -306,40 +361,6 @@ impl SupportedFormat {
     }
 }
 
-impl<'a, T> Deref for InputData<'a, T>
-where
-    T: Sample,
-{
-    type Target = [T];
-
-    #[inline]
-    fn deref(&self) -> &[T] {
-        self.buffer
-    }
-}
-
-impl<'a, T> Deref for OutputData<'a, T>
-where
-    T: Sample,
-{
-    type Target = [T];
-
-    #[inline]
-    fn deref(&self) -> &[T] {
-        self.buffer
-    }
-}
-
-impl<'a, T> DerefMut for OutputData<'a, T>
-where
-    T: Sample,
-{
-    #[inline]
-    fn deref_mut(&mut self) -> &mut [T] {
-        self.buffer
-    }
-}
-
 impl From<Format> for SupportedFormat {
     #[inline]
     fn from(format: Format) -> SupportedFormat {

src/platform/mod.rs

@@ -255,15 +255,14 @@ macro_rules! impl_platform_host {
                 }
             }
 
-            fn build_input_stream<T, D, E>(
+            fn build_input_stream<D, E>(
                 &self,
                 format: &crate::Format,
                 data_callback: D,
                 error_callback: E,
             ) -> Result<Self::Stream, crate::BuildStreamError>
             where
-                T: crate::Sample,
+                D: FnMut(&crate::Data) + Send + 'static,
-                D: FnMut(crate::InputData<T>) + Send + 'static,
                 E: FnMut(crate::StreamError) + Send + 'static,
             {
                 match self.0 {
@@ -275,15 +274,14 @@ macro_rules! impl_platform_host {
                 }
             }
 
-            fn build_output_stream<T, D, E>(
+            fn build_output_stream<D, E>(
                 &self,
                 format: &crate::Format,
                 data_callback: D,
                 error_callback: E,
             ) -> Result<Self::Stream, crate::BuildStreamError>
             where
-                T: crate::Sample,
+                D: FnMut(&mut crate::Data) + Send + 'static,
-                D: FnMut(crate::OutputData<T>) + Send + 'static,
                 E: FnMut(crate::StreamError) + Send + 'static,
             {
                 match self.0 {

src/traits.rs

@@ -2,17 +2,15 @@
 
 use {
     BuildStreamError,
+    Data,
     DefaultFormatError,
     DeviceNameError,
     DevicesError,
     Format,
-    InputData,
     InputDevices,
-    OutputData,
     OutputDevices,
     PauseStreamError,
     PlayStreamError,
-    Sample,
     StreamError,
     SupportedFormat,
     SupportedFormatsError,
@@ -120,27 +118,25 @@ pub trait DeviceTrait {
     fn default_output_format(&self) -> Result<Format, DefaultFormatError>;
 
     /// Create an input stream.
-    fn build_input_stream<T, D, E>(
+    fn build_input_stream<D, E>(
         &self,
         format: &Format,
         data_callback: D,
         error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        T: Sample,
+        D: FnMut(&Data) + Send + 'static,
-        D: FnMut(InputData<T>) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static;
 
     /// Create an output stream.
-    fn build_output_stream<T, D, E>(
+    fn build_output_stream<D, E>(
         &self,
         format: &Format,
         data_callback: D,
         error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        T: Sample,
+        D: FnMut(&mut Data) + Send + 'static,
-        D: FnMut(OutputData<T>) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static;
 }