use std::collections::HashSet;
use std::sync::RwLock;
use std::hash::{Hash, Hasher};

use crate::{Signal, Handler, MutHandler, Remote, WeakRemote};

/**
 * A receiver reference, uniquely identified.
 */
#[derive(Clone)]
struct Receiver<S: Signal> {
    /**
     * The hard pointer is used to compare receivers.
     * We know we cannot have two receivers at the same address at the same time in this table,
     * so it is safe to compare them that way.
     */
    ptr: *const (),

    /**
     * The weak remote actor reference.
     */
    actor: WeakRemote<Handler<S>>
}

impl<S: Signal> PartialEq for Receiver<S> {
	fn eq(&self, other: &Receiver<S>) -> bool {
		self.ptr == other.ptr
	}
}

impl<S: Signal> Eq for Receiver<S> {
	// ...
}

impl<S: Signal> Hash for Receiver<S> {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.ptr.hash(state);
    }
}

/**
 * A mutable receiver reference, uniquely identified.
 */
#[derive(Clone)]
struct MutReceiver<S: Signal> {
    /**
     * The hard pointer is used to compare receivers.
     * We know we cannot have two receivers at the same address at the same time in this table,
     * so it is safe to compare them that way.
     */
    ptr: *const (),

    /**
     * The weak remote actor reference.
     */
    actor: WeakRemote<MutHandler<S>>
}

impl<S: Signal> PartialEq for MutReceiver<S> {
	fn eq(&self, other: &MutReceiver<S>) -> bool {
		self.ptr == other.ptr
	}
}

impl<S: Signal> Eq for MutReceiver<S> {
	// ...
}

impl<S: Signal> Hash for MutReceiver<S> {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.ptr.hash(state);
    }
}


/**
 * Signal broadcaster.
 */
pub struct Broadcaster<S: Signal + Clone> {
    receivers: RwLock<HashSet<Receiver<S>>>,
    receivers_mut: RwLock<HashSet<MutReceiver<S>>>
}

impl<S: 'static + Signal + Clone> Broadcaster<S> {
    /**
     * Create a new broadcaster.
     */
    pub fn new() -> Broadcaster<S> {
        Broadcaster {
            receivers: RwLock::new(HashSet::new()),
            receivers_mut: RwLock::new(HashSet::new())
        }
    }

    /**
     * Add a signal receiver.
     * The receiver will be deleted once it has died, after the next signal emission.
     */
    pub fn register(&self, actor: Remote<dyn 'static + Handler<S>>) {
        //let actor = actor.clone() as Remote<Handler<S>>;
        let mut receivers = self.receivers.write().unwrap();
        let receiver = Receiver {
            ptr: actor.ptr() as *const (),
            actor: actor.downgrade()// as WeakRemote<dyn Handler<S>>
        };
        receivers.insert(receiver);
    }

    /**
     * Add a signal receiver.
     * The receiver will be deleted once it has died, after the next signal emission.
     */
    pub fn register_mut(&self, actor: Remote<dyn 'static + MutHandler<S>>) {
        //let actor = actor.clone() as Remote<Handler<S>>;
        let mut receivers_mut = self.receivers_mut.write().unwrap();
        let receiver = MutReceiver {
            ptr: actor.ptr() as *const (),
            actor: actor.downgrade()// as WeakRemote<dyn MutHandler<S>>
        };
        receivers_mut.insert(receiver);
    }

    /**
     * Emit a signal.
     */
    pub fn emit(&self, signal: S) {
        {
            let mut receivers = self.receivers.write().unwrap();
            let mut dead_receivers: Vec<Receiver<S>> = Vec::new();
            for receiver in receivers.iter() {
                match receiver.actor.upgrade() {
                    Some(actor) => {
                        actor.send(signal.clone());
                    },
                    None => dead_receivers.push(receiver.clone())
                }
            }

            for dead_receiver in dead_receivers.iter() {
                receivers.remove(dead_receiver);
            }
        }
        {
            let mut receivers_mut = self.receivers_mut.write().unwrap();
            let mut dead_receivers: Vec<MutReceiver<S>> = Vec::new();
            for receiver in receivers_mut.iter() {
                match receiver.actor.upgrade() {
                    Some(actor) => {
                        actor.send_mut(signal.clone());
                    },
                    None => dead_receivers.push(receiver.clone())
                }
            }

            for dead_receiver in dead_receivers.iter() {
                receivers_mut.remove(dead_receiver);
            }
        }
    }
}

pub enum Subscription<S: Signal> {
    Const(Remote<dyn 'static + Handler<S>>),
    Mut(Remote<dyn 'static + MutHandler<S>>)
}

impl<S: Signal> Signal for Subscription<S> {
    type Response = ();
}

/**
 * An object capable of emitting signals using a broadcaster.
 * Receivers object can receive such signals.
 */
pub trait Emitter<S: Signal> {
    /**
     * Subscribe to an emitter.
     */
    fn subscribe(&self, actor: Remote<dyn 'static + Handler<S>>);

    /**
     * Subscribe mutabily to an emitter.
     */
    fn subscribe_mut(&self, actor: Remote<dyn 'static + MutHandler<S>>);
}

impl<T, S: Signal> Handler<Subscription<S>> for T where T: Emitter<S> {
    fn handle(&self, s: Subscription<S>) {
        match s {
            Subscription::Const(actor) => self.subscribe(actor),
            Subscription::Mut(actor) => self.subscribe_mut(actor)
        }
    }
}