use tokio::spawn;
use tokio::sync::mpsc::{channel, Sender};
use tokio::task::JoinHandle;

use crate::chopper::error::Result;
use crate::chopper::types::{Header, Row};
use crate::driver::sink::SinkDriver;
use crate::output::builder::OutputFactoryBuilder;
use crate::output::factory::OutputFactory;
use crate::output::output::Output;

pub async fn drive_output(
    output: &Output,
    header: &Header,
) -> Result<(Sender<Box<Row>>, JoinHandle<()>)> {
    drive_output_with_factory(&OutputFactoryBuilder::new().build()?, output, header).await
}

pub async fn drive_output_with_factory(
    factory: &OutputFactory,
    output: &Output,
    header: &Header,
) -> Result<(Sender<Box<Row>>, JoinHandle<()>)> {
    let sink = factory.create_sink(&header, output)?;
    let (tx, rx) = channel(256);
    let mut driver = SinkDriver::new(rx, sink);
    let handle = spawn(async move { driver.run().await });
    Ok((tx, handle))
}
