map_iced 0.1.0

//! This module administrates the iced image handles for potentially several different widgets.

use crate::gui_system::internal_math::{BoundingRectangle, TileChange, TilePosition};
use crate::tile_cache::cache_core::{CachingResultMessage, CachingSystem};
use fxhash::{FxHashMap, FxHashSet};
use iced::advanced::image::Handle;
use std::mem::take;
use tokio::sync::mpsc::Receiver;

/// Contains the image entry.
#[derive(Debug, Clone)]
struct ImageEntry {
    /// The real image if already existing.
    image: Option<Handle>,
    /// The counter how often the image is used.
    usage_counter: u32,
}

/// The update messages that can be drained from the system.
#[derive(Clone, Debug)]
pub enum CacheUpdateMessage {
    /// An internal error in the caching system has occurred and can be obtained here.
    ErrorMessage { text: String },
    /// A new tile has arrived that the indicated client is waiting for.
    RelevantTilesArrived { client: u32 },
    /// The files of the cache are deleted, see [`TileCache::clear_cache`].
    CacheCleared,
}

/// The tiles to draw as a  result for a specific client.
pub struct TilesToDraw {
    pub position: TilePosition,
    pub image: Handle,
}

/// The high level tile cache, that encapsulates the real cache with image and file level, as well
/// as the access system for the tiles.
pub struct TileCache {
    /// The core we have access too.
    core: CachingSystem,
    /// This stores the requests and delays delivery until the system is initialized.
    pending_requests: Vec<TilePosition>,
    /// Stores the currently associated image for a certain tile coordinate. Contains the image and a usage counter.
    content_tiles: FxHashMap<TilePosition, ImageEntry>,
    /// The subscription list which client has subscribed for which region.
    subscription_region: FxHashMap<u32, BoundingRectangle>,
    /// Stores, if we have passed the initialization phase.
    is_initialized: bool,
    /// The accumulated error string.
    error_msg: String,
    /// Special error message from tile error.
    tile_error_msg: String,
    /// The hash set of clients, whose subscription got affected.
    client_notifications: FxHashSet<u32>,
    /// Tiles for which a request is currently outstanding at the core.
    tiles_in_flight: FxHashSet<TilePosition>,
    /// A clear requested before the initialization, with its refetch flag.
    pending_clear: Option<bool>,
    /// The refetch flag of the clear running at the core, if any.
    running_clear: Option<bool>,
    /// The clear has finished and still has to be reported.
    cache_cleared: bool,
    /// Tiles in flight when a refetching clear finished. Their answer may stem from
    /// a file read before the deletion, so they get requested once more instead.
    request_again: FxHashSet<TilePosition>,
}

impl TileCache {
    /// Creates a tile cache from the low level system handed over.
    pub fn new(requestor: CachingSystem) -> Self {
        // Kick off the initialization process.
        requestor.initialize();
        Self {
            core: requestor,
            pending_requests: Vec::new(),
            content_tiles: FxHashMap::default(),
            subscription_region: FxHashMap::default(),
            is_initialized: false,
            error_msg: "".to_string(),
            tile_error_msg: "".to_string(),
            client_notifications: FxHashSet::default(),
            tiles_in_flight: FxHashSet::default(),
            pending_clear: None,
            running_clear: None,
            cache_cleared: false,
            request_again: FxHashSet::default(),
        }
    }

    /// Extracts the receiver out of the class. Can only be done one time.
    pub fn get_receiver(&mut self) -> Option<Receiver<CachingResultMessage>> {
        self.core.get_receiver()
    }

    /// Gets the copyright text.
    pub fn get_copyright_text(&self) -> String {
        self.core.get_copyright_text()
    }

    /// Gets the internal update messages that have been accumulated,
    pub fn drain_result_messages(&mut self) -> Vec<CacheUpdateMessage> {
        let mut result: Vec<CacheUpdateMessage> = Vec::new();

        if !self.error_msg.is_empty() || !self.tile_error_msg.is_empty() {
            let mut text = take(&mut self.error_msg);
            text.push_str(&take(&mut self.tile_error_msg));
            result.push(CacheUpdateMessage::ErrorMessage {
                text: text.trim().into(),
            });
        }

        for &client in &self.client_notifications {
            result.push(CacheUpdateMessage::RelevantTilesArrived { client });
        }
        self.client_notifications.clear();
        if take(&mut self.cache_cleared) {
            result.push(CacheUpdateMessage::CacheCleared);
        }
        result
    }

    /// Asks for the  current number of tiles missing can be used for
    /// enabling / disabling buttons.
    pub fn number_of_tiles_failed(&self) -> u32 {
        if !self.is_initialized {
            return 0;
        }
        self.content_tiles
            .iter()
            .filter(|(pos, image)| image.image.is_none() && !self.tiles_in_flight.contains(pos))
            .count() as u32
    }

    /// Gets invoked from the outside to send request for failed tiles again.
    /// A failed tile is a tile with no image data in the cache and whose position is not in tiles in flight.
    pub fn retry_failed_tiles(&mut self) {
        if !self.is_initialized {
            return;
        }
        self.request_missing_tiles();
    }

    /// Requests all tiles without image data that are not in flight.
    fn request_missing_tiles(&mut self) {
        let missing_pos = self
            .content_tiles
            .iter()
            .filter_map(|(pos, image)| {
                (image.image.is_none() && !self.tiles_in_flight.contains(pos)).then_some(*pos)
            })
            .collect::<Vec<TilePosition>>();

        for pos in missing_pos {
            self.tiles_in_flight.insert(pos);
            self.core
                .request_tile(pos.zoom, pos.x, pos.y)
                .expect("Inconsistent tile initialization state");
        }
    }

    /// Deletes the files of the cache. With `refetch` the tiles in use get dropped
    /// and loaded again once the files are gone, otherwise they stay in memory and
    /// nothing gets written to disc any more. Before the initialization the clear
    /// waits for it. [`CacheUpdateMessage::CacheCleared`] reports the end.
    pub fn clear_cache(&mut self, refetch: bool) {
        if !self.is_initialized {
            // Of several early requests, one not refetching wins: it stops the writing.
            self.pending_clear = Some(self.pending_clear.unwrap_or(true) && refetch);
            return;
        }
        self.running_clear = Some(self.running_clear.unwrap_or(true) && refetch);
        self.core.clear_cache(refetch);
    }

    /// Gets called to process the messages, effectively those that have been pulled out the receiver
    pub fn process_caching_message(&mut self, message: CachingResultMessage) {
        match message {
            CachingResultMessage::Error { message: text } => {
                self.error_msg += &*(text + "\n");
            }
            CachingResultMessage::InitializationCompleted => {
                self.is_initialized = true;
                if let Some(refetch) = self.pending_clear.take() {
                    self.clear_cache(refetch);
                }
                self.flush_pending_requests()
            }
            CachingResultMessage::CacheCleared => {
                self.cache_cleared = true;
                // The tiles are only requested again now, a download finishing
                // earlier could have been deleted by the clear.
                if self.running_clear.take() == Some(true) {
                    for entry in self.content_tiles.values_mut() {
                        entry.image = None;
                    }
                    self.client_notifications
                        .extend(self.subscription_region.keys().copied());
                    self.request_again
                        .extend(self.tiles_in_flight.iter().copied());
                    self.request_missing_tiles();
                }
            }
            CachingResultMessage::TileData {
                x,
                y,
                level,
                data,
                decoded,
            } => {
                let pos = TilePosition { x, y, zoom: level };
                if self.settle_request(pos) {
                    return;
                }
                // Eventually we have a late arriver nobody is interested in anymore.
                let Some(cache_entry) = self.content_tiles.get_mut(&pos) else {
                    return;
                };
                debug_assert!(cache_entry.image.is_none(), "The image should be empty now");
                // RGBA is uploaded by iced in the frame it is first drawn, encoded data
                // waits for iced's decode worker. The encoded path stays as fallback.
                cache_entry.image = Some(match decoded {
                    Some(tile) => Handle::from_rgba(tile.width, tile.height, tile.pixels),
                    None => Handle::from_bytes(data),
                });
                for (client, region) in &self.subscription_region {
                    if region.contains_position(&pos) {
                        self.client_notifications.insert(*client);
                    }
                }
            }
            CachingResultMessage::TileFailed {
                x,
                y,
                level,
                message,
            } => {
                let pos = TilePosition { x, y, zoom: level };
                if self.settle_request(pos) {
                    return;
                }
                if self.tile_error_msg.is_empty() {
                    self.tile_error_msg = message;
                }
            }
        }
    }

    /// Settles the request for `pos`, no matter if anybody is still interested. Returns
    /// true if the answer has to be ignored, because the tile is in `request_again`;
    /// then it is requested anew if still in use.
    fn settle_request(&mut self, pos: TilePosition) -> bool {
        if !self.request_again.remove(&pos) {
            self.tiles_in_flight.remove(&pos);
            return false;
        }
        if self.content_tiles.contains_key(&pos) {
            self.core
                .request_tile(pos.zoom, pos.x, pos.y)
                .expect("Inconsistent tile initialization state");
        } else {
            self.tiles_in_flight.remove(&pos);
        }
        true
    }

    /// The client present wants to completely unsubscribe.
    pub fn completely_unsubscribe(&mut self, client: u32) {
        let Some(region) = self.subscription_region.remove(&client) else {
            return;
        };

        for x in region.get_iterator() {
            self.decrement_usage(&x);
        }
    }

    /// We register for a new entrance area.
    pub fn register_new_interest_area(&mut self, client: u32, area: BoundingRectangle) {
        let mut tile_change = TileChange::default();
        // Do we already have something?
        if let Some(old_area) = self.subscription_region.remove(&client) {
            tile_change = old_area.generate_deletion_creation_list(&area);
        } else {
            // This is a new region completely register it.
            tile_change.added = area.get_iterator().collect();
        }
        for pos in tile_change.added {
            self.increment_usage(&pos);
        }
        for pos in tile_change.deleted {
            self.decrement_usage(&pos);
        }

        self.subscription_region.insert(client, area);
    }

    /// We get all the images for a specific client we are subscribed to.
    pub fn get_all_images_for_client(&self, client: u32) -> Vec<TilesToDraw> {
        let Some(bounding_rect) = self.subscription_region.get(&client) else {
            return vec![];
        };
        bounding_rect
            .get_iterator()
            .filter_map(|position| {
                self.content_tiles
                    .get(&position)
                    .and_then(|image| image.image.clone())
                    .map(|image| TilesToDraw { position, image })
            })
            .collect()
    }

    /// Deals with the fact, that someone stops using a specific tile, eventually it will get deleted.
    fn decrement_usage(&mut self, position: &TilePosition) {
        // Get the entry, here we also take care of, that it eventually does not exist.
        let Some(entry) = self.content_tiles.get_mut(position) else {
            return;
        };
        entry.usage_counter = entry.usage_counter.saturating_sub(1);
        if entry.usage_counter == 0 {
            self.content_tiles.remove(position);
        }
    }

    /// Increments the usage count, if it does not exist yet, we register it and send the delivery request.
    fn increment_usage(&mut self, position: &TilePosition) {
        if let Some(entry) = self.content_tiles.get_mut(position) {
            entry.usage_counter += 1;
        } else {
            // Here we have a new position.
            self.content_tiles.insert(
                *position,
                ImageEntry {
                    image: None,
                    usage_counter: 1,
                },
            );
            // Eventually a request from a dropped subscription is still outstanding.
            if !self.tiles_in_flight.insert(*position) {
                return;
            }
            if self.is_initialized {
                self.core
                    .request_tile(position.zoom, position.x, position.y)
                    .expect("Inconsistent tile initialization state");
            } else {
                self.pending_requests.push(*position);
            }
        }
    }

    /// Flushes all pending requests used at the end of the
    /// initialization process.
    fn flush_pending_requests(&mut self) {
        for p in take(&mut self.pending_requests) {
            self.core
                .request_tile(p.zoom, p.x, p.y)
                .expect("Inconsistent start state of tile cache.");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::tile_cache::cache_core::generate_dummy_cache;
    use std::assert_matches;

    #[tokio::test]
    async fn base_test() {
        const TILE_DIMENSION: u32 = 10;
        let test_rect = BoundingRectangle::new(&[
            TilePosition {
                x: 0,
                y: 0,
                zoom: 5,
            },
            TilePosition {
                x: TILE_DIMENSION - 1,
                y: TILE_DIMENSION - 1,
                zoom: 5,
            },
        ]);
        let low_level = generate_dummy_cache(tempfile::tempdir().unwrap(), 100_000);
        let mut high_level = TileCache::new(low_level);
        let mut receiver = high_level.get_receiver().unwrap();

        // Register, drop and register again before the answers arrive.
        high_level.register_new_interest_area(0, test_rect);
        high_level.completely_unsubscribe(0);
        high_level.register_new_interest_area(0, test_rect);

        let message = receiver.recv().await.unwrap();
        assert_matches!(message, CachingResultMessage::InitializationCompleted);
        high_level.process_caching_message(message);

        let mut counter = 0;
        while let Some(message) = receiver.recv().await {
            counter += 1;
            assert_matches!(message, CachingResultMessage::TileData { level: 5, .. });
            high_level.process_caching_message(message);
            if counter == TILE_DIMENSION * TILE_DIMENSION {
                break;
            }
        }
        assert_eq!(high_level.number_of_tiles_failed(), 0);

        let client_images = high_level.get_all_images_for_client(0);
        assert_eq!(
            client_images.len(),
            TILE_DIMENSION as usize * TILE_DIMENSION as usize
        );
        assert_eq!(client_images[0].position.zoom, 5);

        high_level.completely_unsubscribe(0);
        let client_images = high_level.get_all_images_for_client(0);
        assert!(client_images.is_empty());
    }

    /// Processes tile messages till `count` tiles have arrived.
    async fn receive_tiles(
        high_level: &mut TileCache,
        receiver: &mut Receiver<CachingResultMessage>,
        count: usize,
    ) {
        for _ in 0..count {
            let message = receiver.recv().await.unwrap();
            assert_matches!(message, CachingResultMessage::TileData { .. });
            high_level.process_caching_message(message);
        }
    }

    #[tokio::test]
    async fn clear_cache_refetch_test() {
        let test_rect = BoundingRectangle::new(&[
            TilePosition {
                x: 0,
                y: 0,
                zoom: 4,
            },
            TilePosition {
                x: 2,
                y: 2,
                zoom: 4,
            },
        ]);
        let low_level = generate_dummy_cache(tempfile::tempdir().unwrap(), 100_000);
        let mut high_level = TileCache::new(low_level);
        let mut receiver = high_level.get_receiver().unwrap();
        // Requested before the initialization, so it waits for it.
        high_level.clear_cache(true);
        high_level.register_new_interest_area(0, test_rect);

        let message = receiver.recv().await.unwrap();
        assert_matches!(message, CachingResultMessage::InitializationCompleted);
        high_level.process_caching_message(message);

        // The tiles requested meanwhile may answer before or after the clear, all of
        // them are requested again after it.
        let mut cleared = false;
        while !cleared || !high_level.tiles_in_flight.is_empty() {
            let message = receiver.recv().await.unwrap();
            match message {
                CachingResultMessage::CacheCleared => cleared = true,
                CachingResultMessage::TileData { .. } => {}
                other => panic!("Unexpected {other:?}"),
            }
            high_level.process_caching_message(message);
        }
        assert_matches!(
            high_level.drain_result_messages().last(),
            Some(CacheUpdateMessage::CacheCleared)
        );
        assert_eq!(high_level.number_of_tiles_failed(), 0);
        assert_eq!(high_level.get_all_images_for_client(0).len(), 9);

        // A clear after the initialization drops the images and loads them again.
        high_level.clear_cache(true);
        let message = receiver.recv().await.unwrap();
        assert_matches!(message, CachingResultMessage::CacheCleared);
        high_level.process_caching_message(message);
        assert!(high_level.get_all_images_for_client(0).is_empty());
        receive_tiles(&mut high_level, &mut receiver, 9).await;
        assert_eq!(high_level.get_all_images_for_client(0).len(), 9);
        assert!(high_level.tiles_in_flight.is_empty());
    }
}