#![allow(unsafe_code)]

use std::collections::hash_map::{RandomState, Entry};
use std::collections::HashMap;
use std::borrow::{Borrow, BorrowMut};
use std::hash::Hash;
use std::ops::{Deref, DerefMut};
use crate::runtime::{RwLock, RwLockReadGuard, RwLockWriteGuard};

/// SyncMap impl the Send and Sync

/// it use of RwLock,so it's safe! but we went convert lifetime ,so use some lifetime convert unsafe method(but it is safe)

pub struct SyncMap<K, V> where K: Eq + Hash {
    pub shard: RwLock<HashMap<K, V, RandomState>>,
}

impl<'a, K: 'a + Eq + Hash, V: 'a> SyncMap<K, V> where K: Eq + Hash {
    pub fn new() -> Self {
        Self {
            shard: RwLock::new(HashMap::new())
        }
    }

    pub fn with_capacity(capacity: usize) -> Self {
        Self {
            shard: RwLock::new(HashMap::with_capacity(capacity))
        }
    }

    pub async fn insert(&self, key: K, value: V) -> Option<V> {
        let mut w = self.shard.write().await;
        w.insert(key, value)
    }

    pub async fn remove<Q>(&self, key: &Q) -> Option<V>
        where
            K: Borrow<Q>,
            Q: Hash + Eq + ?Sized {
        let mut w = self.shard.write().await;
        w.remove(key)
    }

    pub async fn clear(&self){
        let mut w = self.shard.write().await;
        w.clear();
    }

    pub async fn shrink_to_fit(&self) {
        let mut w = self.shard.write().await;
        w.shrink_to_fit();
    }

    pub async fn reserve(&self, additional: usize) {
        let mut w = self.shard.write().await;
        w.reserve(additional)
    }

    pub async fn read(&self) -> RwLockReadGuard<'_, HashMap<K, V, RandomState>> {
        self.shard.read().await
    }

    pub async fn write(&self) -> RwLockWriteGuard<'_, HashMap<K, V, RandomState>> {
        self.shard.write().await
    }

    pub async fn get<Q>(&'a self, k: &Q) -> Option<Ref<'a, K, V>>
        where K: Borrow<Q>,
              Q: Hash + Eq + ?Sized {
        let mut get_ref = Ref::new(self.shard.read().await, None);
        unsafe {
            let v = get_ref._guard.get(k);
            if v.is_some() {
                let vptr = change_lifetime_const(v.unwrap());
                get_ref.v = Option::from(vptr);
                Some(get_ref)
            } else {
                None
            }
        }
    }

    pub async fn get_mut<Q>(&'a self, k: &Q) -> Option<RefMut<'a, K, V>>
        where K: Borrow<Q>,
              Q: Hash + Eq + ?Sized {
        let mut get_ref = RefMut::new(self.shard.write().await, None);
        unsafe {
            let v = get_ref._guard.get_mut(k);
            if v.is_some() {
                let vptr = change_lifetime_mut(v.unwrap());
                get_ref.v = Option::Some(vptr);
                Some(get_ref)
            } else {
                None
            }
        }
    }
}

pub unsafe fn change_lifetime_const<'a, 'b, T>(x: &'a T) -> &'b T {
    &*(x as *const T)
}

pub unsafe fn change_lifetime_mut<'a, 'b, T>(x: &'a mut T) -> &'b mut T {
    &mut *(x as *mut T)
}

#[derive(Debug)]
pub struct Ref<'a, K, V>
    where K: Eq + Hash {
    _guard: RwLockReadGuard<'a, HashMap<K, V, RandomState>>,
    v: Option<&'a V>,
}

impl<'a, K, V> Ref<'a, K, V> where K: Eq + Hash {
    pub fn new(guard: RwLockReadGuard<'a, HashMap<K, V, RandomState>>, v: Option<&'a V>) -> Self {
        let s = Self {
            _guard: guard,
            v: v,
        };
        s
    }
    pub fn value(&self) -> &V {
        self.v.unwrap()
    }
}

impl<'a, K: Eq + Hash, V> Deref for Ref<'a, K, V> {
    type Target = V;

    fn deref(&self) -> &V {
        &self.value()
    }
}

#[derive(Debug)]
pub struct RefMut<'a, K, V, S = RandomState> {
    _guard: RwLockWriteGuard<'a, HashMap<K, V, S>>,
    v: Option<&'a mut V>,
}

impl<'a, K: Eq + Hash, V> RefMut<'a, K, V> {
    pub fn new(guard: RwLockWriteGuard<'a, HashMap<K, V, RandomState>>, v: Option<&'a mut V>) -> Self {
        let s = Self {
            _guard: guard,
            v: v,
        };
        s
    }

    pub fn value(&self) -> &V {
        self.v.as_ref().unwrap()
    }

    pub fn value_mut(&mut self) -> &mut V {
        self.v.as_mut().unwrap()
    }
}


impl<'a, K: Eq + Hash, V> Deref for RefMut<'a, K, V> {
    type Target = V;

    fn deref(&self) -> &V {
        self.value()
    }
}

impl<'a, K: Eq + Hash, V> DerefMut for RefMut<'a, K, V> {
    fn deref_mut(&mut self) -> &mut V {
        self.value_mut()
    }
}


#[cfg(test)]
mod test {
    use std::collections::HashMap;
    use std::sync::Arc;
    use std::ops::Deref;
    use futures_util::StreamExt;
    use crate::sync::sync_map::SyncMap;
    use std::time::SystemTime;
    use chrono::Local;


    #[test]
    fn test_map() {
        let m = Arc::new(SyncMap::new());
        async_std::task::block_on(async {
            let v = m.insert(1, "default".to_string()).await;
            let r = m.get(&1).await;
            let rv = r.unwrap().v;
            println!("r:{:?}", &rv);
            assert_eq!("default", format!("{}", &rv.unwrap()));

            drop(rv);

            let mut mut_v = m.get_mut(&1).await.unwrap();
            *mut_v = "changed".to_string();
            drop(mut_v);
            let r = m.get(&1).await;
            println!("r:{:?}", &r.as_ref().unwrap().deref());
            assert_eq!("changed", format!("{}", &r.as_ref().unwrap().deref()));
        });
    }

    #[test]
    fn test_map_for() {
        let m = Arc::new(SyncMap::new());
        async_std::task::block_on(async {
            let mut lock= m.write().await;
            lock.insert(1,1);
            drop(lock);
            let mut lock= m.read().await;
            for (k,v) in lock.deref(){
               println!("k:{},v:{}",k,v);
            }
        });
    }


    //bench on windows10 40 nano/op.  It depends on the runtime(tokio/async_std) speed

    //test command:

    //cargo test --release --color=always --package rbatis-core --lib sync::sync_map::test::bench_test --no-fail-fast -- --exact -Z unstable-options --format=json --show-output

    #[test]
    fn bench_test(){
        let mut m = Arc::new(SyncMap::new());
        async_std::task::block_on(async {
            let s = m.insert(1, "default".to_string()).await;
            drop(s);

            let total = 100000;
            let now=Local::now().timestamp_millis();
            for current in 0..total{
                let v=  m.get(&1).await;
                if current == total - 1 {
                    let end = Local::now().timestamp_millis();
                    print_use_time(total,now as i64,end as i64);
                    break;
                }
            }
            m.shrink_to_fit().await;
            println!("done");
        });
    }

    fn print_use_time(total: i32, start: i64, end: i64) {
        let mut time = (end - start) as f64;
        time = time / 1000.0;
        println!("use Time: {} s,each:{} nano/op", time, time * 1000000000.0 / (total as f64));
    }
}