rostrum 14.0.1

use anyhow::Result;
use bitcoincash::Txid;
use std::collections::HashSet;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::chaindef::{BlockHash, Transaction};
use crate::thread;
use crate::util::Channel;

static GLOBAL_SHUTDOWN_SIGNAL: AtomicBool = AtomicBool::new(false);

// Type of signals we listen for in the main loop of the program.
pub enum SignalNotification {
    NewTxs(HashSet<Txid>),
    NewFullTx(Transaction),
    NewBlock(BlockHash),
    Timeout,
}

pub struct Waiter {
    signal_channel: Channel<SignalNotification>,
}

/// Signals we receive via p2p connections
pub struct NetworkNotifier {
    pub block_channel: Channel<BlockHash>,
    pub tx_channel: Channel<HashSet<Txid>>,
    pub full_tx_channel: Channel<Transaction>,
}

impl NetworkNotifier {
    pub fn new() -> NetworkNotifier {
        let block_channel = Channel::<BlockHash>::bounded(100);
        let tx_channel = Channel::<HashSet<Txid>>::bounded(5);
        let full_tx_channel = Channel::<Transaction>::bounded(5);
        NetworkNotifier {
            block_channel,
            tx_channel,
            full_tx_channel,
        }
    }
}

impl Default for NetworkNotifier {
    fn default() -> Self {
        Self::new()
    }
}

async fn notify(network_notifier: Arc<NetworkNotifier>) -> Channel<SignalNotification> {
    let c = Channel::bounded(100_000);
    let s = c.sender();

    thread::spawn_task("network_signal", async move {
        loop {
            if Waiter::shutdown_check().is_err() {
                break;
            }

            let mut tx_receiver = network_notifier.tx_channel.receiver().await;
            let mut block_receiver = network_notifier.block_channel.receiver().await;
            let mut full_tx_receiver = network_notifier.full_tx_channel.receiver().await;

            tokio::select! {
                Some(txs) = tx_receiver.recv() => {
                    // Best effort
                    if s.try_send(SignalNotification::NewTxs(txs)).is_err() {
                        debug!("Failed to send NewTxs notification");
                    }
                }
                Some(blockhash) = block_receiver.recv() => {
                    // Best effort
                    if s.try_send(SignalNotification::NewBlock(blockhash)).is_err() {
                        debug!("Failed to send NewBlock notification");
                    }
                }
                Some(tx) = full_tx_receiver.recv() => {
                    if s.try_send(SignalNotification::NewFullTx(tx)).is_err() {
                        debug!("Failed to send NewTxBroadcast notification");
                    }
                }
            }
        }

        Ok(())
    });
    c
}

impl Waiter {
    #[cfg(not(windows))]
    fn start_shutdown_listener() {
        info!("Listening for shutdown signal");
        let posix_signals = &[
            signal_hook::consts::SIGINT,
            signal_hook::consts::SIGTERM,
            signal_hook::consts::SIGUSR1, // allow external triggering (e.g. via bitcoind `blocknotify`)
        ];
        let mut signals = signal_hook::iterator::Signals::new(posix_signals)
            .expect("failed to register signal hook");

        crate::thread::spawn("posix_signal", move || {
            for signal in signals.forever() {
                if signal == signal_hook::consts::SIGUSR1 {
                    // Ignore. Used to trigger block update in earlier versions.
                    continue;
                }
                info!("Shutdown triggered via signal {}", signal);
                GLOBAL_SHUTDOWN_SIGNAL.store(true, Ordering::Relaxed);
            }
            Ok(())
        });
    }

    #[cfg(windows)]
    fn start_shutdown_listener() {
        let (tx, rx) = std::sync::mpsc::channel();
        ctrlc::set_handler(move || tx.send(()).expect("Could not send signal on channel."))
            .expect("Error setting Ctrl-C handler");

        crate::thread::spawn("win_signal", move || {
            rx.recv().expect("Could not receive from channel.");
            info!("Shutdown triggered via signal");
            GLOBAL_SHUTDOWN_SIGNAL.store(true, Ordering::Relaxed);
            Ok(())
        });
    }

    pub async fn start(network_notifier: Arc<NetworkNotifier>) -> Waiter {
        let waiter = Waiter {
            signal_channel: notify(network_notifier).await,
        };
        Waiter::start_shutdown_listener();
        waiter
    }
    pub fn shutdown_check() -> Result<()> {
        if GLOBAL_SHUTDOWN_SIGNAL.load(Ordering::Relaxed) {
            bail!("Shutdown triggered");
        }
        Ok(())
    }
    pub async fn wait(&self, duration: Duration) -> Result<SignalNotification> {
        Self::shutdown_check()?;
        let signal =
            tokio::time::timeout(duration, self.signal_channel.receiver().await.recv()).await;
        match signal {
            Ok(Some(s)) => Ok(s),
            Ok(None) => bail!("signal hook channel disconnected"),
            Err(_e) => Ok(SignalNotification::Timeout),
        }
    }

    pub async fn wait_or_shutdown(duration: Duration) -> Result<()> {
        let start = Instant::now();
        while start.elapsed() < duration {
            Self::shutdown_check()?;
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        Ok(())
    }

    pub fn shutdown(reason: &str) {
        error!("Shutdown called due to {}", reason);
        GLOBAL_SHUTDOWN_SIGNAL.store(true, Ordering::Relaxed);
    }
}