You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: input_file.rs
    path of file: ./repos/teloxide/crates/teloxide-core/src/types
    
    the code of the file until where you have to start completion: use bytes::{Bytes, BytesMut};
use futures::{
    future::{ready, Either},
    stream,
};
use once_cell::sync::OnceCell;
use rc_box::ArcBox;
use reqwest::{multipart::Part, Body};
use serde::Serialize;
use takecell::TakeCell;
use tokio::{
    io::{AsyncRead, AsyncReadExt, ReadBuf},
    sync::watch,
};
use tokio_util::codec::{Decoder, FramedRead};

use std::{
    borrow::Cow,
    convert::{Infallible, TryFrom},
    fmt,
    future::Future,
    io, iter, mem,
    path::PathBuf,
    pin::Pin,
    sync::Arc,
    task,
};

use crate::types::InputSticker;

/// This object represents the contents of a file to be uploaded.
///
/// [The official docs](https://core.telegram.org/bots/api#inputfile).
#[derive(Debug, Clone)]
pub struct InputFile {
    id: OnceCell<Arc<str>>,
    file_name: Option<Cow<'static, str>>,
    inner: InnerFile,
}

#[derive(Clone)]
enum InnerFile {
    Read(Read),
    File(PathBuf),
    Bytes(bytes::Bytes),
    Url(url::Url),
    FileId(String),
}

use InnerFile::*;

impl InputFile {
    /// Creates an `InputFi
    
    async fn into_shared_body(mut self) -> Body {
        match self.inner.take() {
            // Read `dyn AsyncRead` into a buffer
            Some(mut read_ref) => {
                // Chunk size, arbitrary chosen to be 1KiB
                const CHUNK: usize = 1024;

                let mut chunks = Vec::new();
                let mut bytes = BytesMut::with_capacity(CHUNK);

                let res = loop {
                    match (&mut read_ref).read_buf(&mut bytes).await {
                        // eof
                        Ok(0) if bytes.len() < bytes.capacity() => {
                            chunks.push(bytes.freeze());

                            break Ok(chunks);
                        }

                        // No space left in bytes, allocate a new chunk
                        Ok(0) => {
                            chunks.push(bytes.freeze());
                            bytes = BytesMut::with_capacity(CHUNK);
                        }

                        // keep reading into the same chunk
                        Ok(_) => {}

                        // i/o error
                        Err(err) => break Err(Arc::new(err)),
                    }
                };

                // Initialize `buf` with the result.
                // Error indicates that the `buf` was already initialized, but this can't happen
                // since we synchronize through other means.
                let r = self.buf.set(res);
                debug_assert!(r.is_ok());

                // Notify other tasks that `buf` is initialized.
                // Error indicates that there is no one to notify anymore, but we don't care.
                let _ = self.notify.send(());
            }

            // Wait until `dyn AsyncRead` is read into a buffer, if it hasn't been read yet
            None if self.buf.get().is_none() => {
                // Error indicates that the sender was dropped, by we hold `Arc<Sender>`, so
                // this can't happen
                let _ = self.wait.changed().await;
            }

            // Someone else has already initialized the buffer
            None => {}
        };

        let buf = self.buf;
        // unwrap: `OnceCell` is initialized in the match above before sending
        // notification, so at this point it's already initialized.
        match buf.get().unwrap() {
            Ok(_) => {
                // We can't use `.iter()` here, because the iterator must capture `buf`
                let mut i = 0;
                let iter = iter::from_fn(move || match buf.get().unwrap() {
                    Ok(buf) if i >= buf.len() => None,
                    Ok(buf) => {
                        let res = buf[i].clone();
                        i += 1;
                        Some(Ok::<_, Infallible>(res))
                    }
                    // We've just checked in the above match, it's `Ok(_)`
                    Err(_) => unreachable!(),
                });

                Body::wrap_stream(stream::iter(iter))
            }

            Err(err) => {
                let err = Err::<Bytes, _>(Arc::clone(err));
                Body::wrap_stream(stream::iter(iter::once(err)))
            }
        }
    }