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

use crate::chopper::types::Row;
use crate::driver::joiner::Joiner;

pub async fn drive_joiner(
    output_tx: Sender<Box<Row>>,
    input_count: usize,
) -> (Vec<Sender<Box<Row>>>, JoinHandle<()>) {
    let mut joiner_txs: Vec<Sender<Box<Row>>> = Vec::new();
    let mut joiner_rxs: Vec<Receiver<Box<Row>>> = Vec::new();
    for _ in 0..input_count {
        let (joiner_tx, joiner_rx) = channel(256);
        joiner_txs.push(joiner_tx);
        joiner_rxs.push(joiner_rx);
    }
    let mut joiner = Joiner::new(joiner_rxs, output_tx);
    (joiner_txs, spawn(async move { joiner.run().await }))
}
