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: interface.rs
    path of file: ./repos/nushell/crates/nu-plugin/src/plugin
    
    the code of the file until where you have to start completion: //! Implements the stream multiplexing interface for both the plugin side and the engine side.

use std::{
    io::Write,
    sync::{
        atomic::{AtomicBool, Ordering::Relaxed},
        Arc, Mutex,
    },
    thread,
};

use nu_protocol::{ListStream, PipelineData, RawStream, ShellError};

use crate::{
    plugin::Encoder,
    protocol::{
        ExternalStreamInfo, ListStreamInfo, PipelineDataHeader, RawStreamInfo, StreamMessage,
    },
    sequence::Sequence,
};

mod stream;

mod engine;
pub use engine::EngineInterface;
pub(crate) use engine::{EngineInterfaceManager, ReceivedPluginCall};

mod plugin;
pub(crate) use plugin::{PluginInterface, PluginInterfaceManager};

use self::stream::{StreamManager, StreamManagerHandle, StreamWriter, WriteStreamMessage};

#[cfg(test)]
mod test_util;

#[cfg(test)]
mod tests;

/// The maximum number of list stream values to send without acknowledgement. This should be tuned
/// with consideration for memory usage.
const LIST_STREAM_HIGH_PRESSURE: i32 = 100;

/// The maximum number of raw stream buffers to send without acknowledgement. This should be tuned
/// with consideration for memory usage.
const RAW_STREAM_HIGH_PRESSURE: i32 = 50;

/// Read input/output from the stream.
pub(crate) trait PluginRead<T> {
    /// Returns `Ok(None)` on end of stream.
    fn read(&mut self) -> Result<Option<T>, ShellError>;
}

impl<R, E, T> PluginRead<T> for (R, E)
where
    R: std::io::BufRead,
    E: Encoder<T>,
{
    fn read(&mut self) -> Result<Option<T>, ShellError> {
        self.1.decode(&mut self.0)
    }
}

impl<R, T> PluginRead<T> for &mut R
where
    R: PluginRead<T>,
{
    fn read(&mut self) -> Result<Option<T>, ShellError> {
        (**self).read()
    }
}

/// Write input/output to the stream.
///
/// The write should be atomic, without interference from other threads.
pub(crate) trait PluginWrite<T>: Send + Sync {
    fn write(&self, data: &T) -> Result<(), ShellError>;

    /// Flush any internal buffers, if applicable.
    fn flush(&self) -> Result<(), ShellError>;
}

impl<E, T> PluginWrite<T> for (std::io::Stdout, E)
where
    E: Encoder<T>,
{
    fn write(&self, data: &T) -> Result<(), ShellError> {
        let mut lock = self.0.lock();
        self.1.encode(data, &mut lock)
    }

    fn flush(&self) -> Result<(), ShellError> {
        self.0.lock().flush().map_err(|err| ShellError::IOError {
            msg
    
    pub(crate) fn write(self) -> Result<(), ShellError> {
        match self {
            // If no stream was contained in the PipelineData, do nothing.
            PipelineDataWriter::None => Ok(()),
            // Write a list stream.
            PipelineDataWriter::ListStream(mut writer, stream) => {
                writer.write_all(stream)?;
                Ok(())
            }
            // Write all three possible streams of an ExternalStream on separate threads.
            PipelineDataWriter::ExternalStream {
                stdout,
                stderr,
                exit_code,
            } => {
                thread::scope(|scope| {
                    let stderr_thread = stderr
                        .map(|(mut writer, stream)| {
                            thread::Builder::new()
                                .name("plugin stderr writer".into())
                                .spawn_scoped(scope, move || {
                                    writer.write_all(raw_stream_iter(stream))
                                })
                        })
                        .transpose()?;
                    let exit_code_thread = exit_code
                        .map(|(mut writer, stream)| {
                            thread::Builder::new()
                                .name("plugin exit_code writer".into())
                                .spawn_scoped(scope, move || writer.write_all(stream))
                        })
                        .transpose()?;
                    // Optimize for stdout: if only stdout is present, don't spawn any other
                    // threads.
                    if let Some((mut writer, stream)) = stdout {
                        writer.write_all(raw_stream_iter(stream))?;
                    }
                    let panicked = |thread_name: &str| {
                        Err(ShellError::NushellFailed {
                            msg: format!(
                                "{thread_name} thread panicked in PipelineDataWriter::write"
                            ),
                        })
                    };
                    stderr_thread
                        .map(|t| t.join().unwrap_or_else(|_| panicked("stderr")))
                        .transpose()?;
                    exit_code_thread
                        .map(|t| t.join().unwrap_or_else(|_| panicked("exit_code")))
                        .transpose()?;
                    Ok(())
                })
            }
        }
    }