use async_trait::async_trait;
use tokio::sync::mpsc::Sender;

use crate::chopper::error::{Error, Result};
use crate::chopper::functor::Functor;
use crate::chopper::types::{FieldValue, Header, Row};

pub struct RowFilterEqualValue {
    column_index: usize,
    value: FieldValue,
}

impl RowFilterEqualValue {
    pub fn new(header: &Header, column_name: &str, value: FieldValue) -> Result<Self> {
        let field_names = header.field_names();
        for i in 0..field_names.len() {
            if field_names[i].eq_ignore_ascii_case(column_name) {
                // set column index
                let filter = RowFilterEqualValue {
                    column_index: i,
                    value,
                };
                return Ok(filter);
            }
        }
        Err(Error::from(format!(
            "RowFilterEqualValueConfig -- field name [{}] not found",
            column_name
        )))
    }
}

#[async_trait]
impl Functor for RowFilterEqualValue {
    async fn write(&mut self, row: Box<Row>, output: &mut Sender<Box<Row>>) -> Result<()> {
        let field_value: &FieldValue = row.field_values.get(self.column_index).unwrap();
        if field_value.eq(&self.value) {
            output.send(row).await.unwrap();
        }
        Ok(())
    }
}
