#[cfg(feature = "source_n_triples")]
pub mod n_triples;
#[cfg(feature = "source_turtle")]
pub mod turtle;

use crate::prelude::*;
use crate::terms::BlankNode;

use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::Mutex;
use std::task::{Context, Poll};

use tracing::debug;

use tokio::io::{AsyncBufRead, AsyncBufReadExt};

use futures::{Stream, StreamExt};

#[derive(Debug)]
pub enum TripleParser<R: AsyncBufRead + AsyncBufReadExt> {
    #[cfg(feature = "source_n_triples")]
    NTriples(self::n_triples::NTriplesParser<tokio::io::Lines<R>>),
    #[cfg(feature = "source_turtle")]
    Turtle(self::turtle::TurtleParser<async_utf8_decoder::Utf8Decoder<tokio_util::compat::Compat<R>>>),
    Phantom(PhantomData<R>),
}

impl<R: AsyncBufRead + AsyncBufReadExt> TripleParser<R> {
    pub fn guess_media_type(stream: &mut R) -> Option<String> {
        // FIXME implement or remove feature
        None
    }

    pub fn from_media_type(
        media_type: &str,
        stream: R,
        state: Arc<Mutex<BasicProfile>>,
    ) -> Result<Self, SourceError> {
        debug!("Determining parser for media type {}", media_type);
        match media_type {
            #[cfg(feature = "source_n_triples")]
            "application/n-triples" => Ok(Self::NTriples(self::n_triples::NTriplesParser::new(
                stream, state,
            ))),
            #[cfg(feature = "source_turtle")]
            "text/turtle" => Ok(Self::Turtle(self::turtle::TurtleParser::new(stream, state))),
            _ => Err(SourceError::UnsupportedSourceConfig),
        }
    }
}

impl<R: AsyncBufRead + Unpin> Stream for TripleParser<R> {
    type Item = Result<Arc<Triple>, SourceError>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.get_mut() {
            #[cfg(feature = "source_n_triples")]
            Self::NTriples(parser) => parser.poll_next_unpin(cx),
            #[cfg(feature = "source_turtle")]
            Self::Turtle(parser) => parser.poll_next_unpin(cx),
            _ => unreachable!(),
        }
    }
}
