chunk-streamer 0.3.0

use crate::chunk_receiver::ChunkReceiver;
use crate::chunk_sender::ChunkSender;
use async_trait::async_trait;
use autonomi::client::GetError;
use autonomi::{Chunk, ChunkAddress};
use self_encryption::{streaming_decrypt, DataMap};
use tokio::sync::mpsc::channel;
use crate::chunk_getter_wrapper::ChunkGetterWrapper;

#[async_trait]
pub trait ChunkGetter: Clone + Send + Sync + 'static {
    async fn chunk_get(&self, address: &ChunkAddress) -> Result<Chunk, GetError>;
}

pub struct ChunkStreamer<T> {
    id: String,
    data_map: DataMap,
    chunk_getter: T,
    download_threads: usize,
}

impl<T: ChunkGetter> ChunkStreamer<T> {
    pub fn new(id: String, data_map: DataMap, chunk_getter: T, download_threads: usize) -> ChunkStreamer<T> {
        ChunkStreamer { id, data_map, chunk_getter, download_threads }
    }
    
    pub fn open(&self, range_from: u64, range_to: u64) -> ChunkReceiver {
        let (sender, receiver) = channel(self.download_threads);
        let chunk_sender = ChunkSender::new(sender, self.id.clone(), self.chunk_getter.clone(), self.data_map.clone());
        tokio::spawn( Box::pin(async move { chunk_sender.send(range_from, range_to).await; }));
        ChunkReceiver::new(receiver, self.id.clone())
    }

    pub async fn get_stream_size(&self) -> usize {
        let local_data_map = self.data_map.clone();
        let local_chunk_getter = self.chunk_getter.clone();

        let join_handle = tokio::task::spawn_blocking(move || {
            let chunk_getter_wrapper = ChunkGetterWrapper::new(local_chunk_getter);
            let get_chunk_functor = chunk_getter_wrapper.get_chunk_functor();
            let stream = streaming_decrypt(&local_data_map, get_chunk_functor)
                .expect("failed to execute streaming_decrypt");
            stream.file_size()
        });
        join_handle.await.unwrap_or(0)
    }
}