async fn put<S>(&self, location: &str, bytes: S, length: usize) -> InternalResult<()>
    where
        S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
    {
        let temporary_non_streaming = bytes
            .map_ok(|b| bytes::BytesMut::from(&b[..]))
            .try_concat()
            .await
            .expect("Should have been able to collect streaming data")
            .to_vec();

        ensure!(
            temporary_non_streaming.len() == length,
            DataDoesNotMatchLength {
                actual: temporary_non_streaming.len(),
                expected: length,
            }
        );

        let location_copy = location.to_string();
        let bucket_name = self.bucket_name.clone();

        let _ = tokio::task::spawn_blocking(move || {
            cloud_storage::Object::create(
                &bucket_name,
                &temporary_non_streaming,
                &location_copy,
                "application/octet-stream",
            )
        })
        .await
        .context(UnableToPutDataToGcs {
            bucket: &self.bucket_name,
            location,
        })?;

        Ok(())
    }
/// Save the provided bytes to the specified location.
Save the provided bytes to the specified location.
Save the provided bytes to the specified location .
async fn put<S>(&self, location: &str, bytes: S, length: usize) -> InternalResult<()>
    where
        S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
    {
        let temporary_non_streaming = bytes
            .map_ok(|b| bytes::BytesMut::from(&b[..]))
            .try_concat()
            .await
            .expect("Should have been able to collect streaming data")
            .to_vec();
        ensure!(
            temporary_non_streaming.len() == length,
            DataDoesNotMatchLength {
                actual: temporary_non_streaming.len(),
                expected: length,
            }
        );
        let location_copy = location.to_string();
        let bucket_name = self.bucket_name.clone();
        let _ = tokio::task::spawn_blocking(move || {
            cloud_storage::Object::create(
                &bucket_name,
                &temporary_non_streaming,
                &location_copy,
                "application/octet-stream",
            )
        })
        .await
        .context(UnableToPutDataToGcs {
            bucket: &self.bucket_name,
            location,
        })?;
        Ok(())
    }
async fn put < S > ( & self , location : & str , bytes : S , length : usize ) -> InternalResult < ( ) > where S : Stream < Item = io :: Result < Bytes > > + Send + Sync + ' static , { let temporary_non_streaming = bytes . map_ok ( | b | bytes :: BytesMut :: from ( & b [ .. ] ) ) . try_concat ( ) . await . expect ( " " ) . to_vec ( ) ; ensure ! ( temporary_non_streaming len ( ) length DataDoesNotMatchLength { actual temporary_non_streaming len ( ) expected length } ) ; let location_copy = location . to_string ( ) ; let bucket_name = self . bucket_name . clone ( ) ; let _ = tokio :: task :: spawn_blocking ( move | | { cloud_storage :: Object :: create ( & bucket_name , & temporary_non_streaming , & location_copy , " " , ) } ) . await . context ( UnableToPutDataToGcs { bucket : & self . bucket_name , location , } ) ? ; Ok ( ( ) ) }
Save the provided bytes to the specified location.
Save the provided bytes to the specified location .
&self
location
&str
bytes
S
length
usize