// Copyright 2017, CZ.NIC z.s.p.o. (http://www.nic.cz/)
//
// This file is part of the pakon system.
//
// Pakon is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
//  (at your option) any later version.
//
// Pakon is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with Pakon.  If not, see <http://www.gnu.org/licenses/>.

//! Utilities for gathering flow updates and turning them into flow slices.

#![deny(missing_docs)]

extern crate int_compute;
extern crate libdata;
extern crate libflow;
#[cfg(test)]
extern crate libquery;
extern crate libutils;
extern crate futures;
#[cfg(test)]
#[macro_use]
extern crate serde_derive;
#[macro_use]
extern crate slog;

use std::cell::RefCell;
use std::collections::{BTreeMap, BTreeSet};
use std::mem;
use std::rc::Rc;
use std::time::{Duration, SystemTime};

use futures::unsync::mpsc::UnboundedSender;

use slog::Logger;

use int_compute::{Event, Query as ComputeQuery};
use libdata::column::{FlowTags, RefHeaders, Tags};
use libdata::stats::Sizes as SizeStats;
use libflow::update::{Cork, Key, Status, Update};
use libflow::slice::{Flow as FlowSlice, Time as TimeSlice};
use libutils::tunable::{SLICE_INACTIVE_LIMIT, SLICE_KEEP_ENDED, SLICE_LENGTH_SECONDS};

/// An internal „fat incomplete flow slice“
///
/// This represents the active, live flow. We need some more information during the tracking and
/// construction than when we keep the older slices and this represents the active head.
struct FlowHead {
    /// The keys we are able to look the flow by.
    ///
    /// In addition to these keys, a flow can be looked up by an internal handle (eg.
    /// `Key::InternalHandle`), but it is not saved here, to prevent circular references and save
    /// on memory, as each flow has exactly one such handle.
    keys: BTreeSet<Key>,
    /// The tags on the flow.
    tags: FlowTags,
    /// The last *absolute* snapshot of the statistics.
    last_stats: SizeStats,
    /// The statistics accumulated since the last slicing.
    current_stats: SizeStats,
    /// The current status of the flow.
    status: Status,
    /// When the flow started (our own time).
    start_time: SystemTime,
    /// When the flow ended (for flows that already ended).
    end_time: Option<SystemTime>,
    /// For how many slices the flow made no activity.
    inactive_for: u16,
    /// Did the flow start since the last slicing?
    ///
    /// Eg. is it yet unsliced?
    ///
    /// # Note
    ///
    /// Even when the new flow is created in Ongoing state, it starts here. This relates to our
    /// known start of the flow, not a real one.
    started_here: bool,
}

impl FlowHead {
    /// Creates a new flow.
    ///
    /// It sets some basic parameters to sane values. These are left empty and populated later on,
    /// by updates (including the initial update which is expected to happen right afterwards):
    ///
    /// * keys
    /// * tags
    ///
    /// The start time is set to now. If the flow is just started, it includes the passed
    /// statistics in the current slice, otherwise it ignores all the data that happened beforehand
    /// (that's to cope with seeing a long-running flow just after the daemon starts).
    ///
    /// # Params
    ///
    /// * status: The status of the flow.
    /// * current_stats: The initial statistics (absolute). This may be zero.
    fn new(status: Status, current_stats: SizeStats) -> Self {
        let stats = if status == Status::Start {
            // If the flow starts just now, contain all the stats
            current_stats.clone()
        } else {
            // Otherwise, when it started some long time ago, just ignore the unknown part and
            // start from zero
            SizeStats::default()
        };
        Self {
            keys: BTreeSet::new(),
            tags: FlowTags::new(Tags::new()),
            last_stats: current_stats,
            current_stats: stats,
            status,
            start_time: SystemTime::now(),
            end_time: None,
            inactive_for: 0,
            started_here: true,
        }
    }
    /// Slices the flow and produce the slice.
    ///
    /// This cuts the flow, resets the statistics for the next slice and returns the just
    /// terminated slice.
    fn slice(&mut self, mut start: SystemTime, mut end: SystemTime) -> FlowSlice {
        if self.start_time > start {
            start = self.start_time;
        }
        if let Some(my_end) = self.end_time {
            if my_end < end {
                end = my_end;
            }
        }
        let stats = self.current_stats.expand(start, end,
                                              Duration::from_secs(SLICE_LENGTH_SECONDS),
                                              self.started_here);
        self.current_stats = SizeStats::default();
        self.started_here = false;
        FlowSlice::new(self.tags.clone(), stats)
    }
}

/// A data structure that tracs multiple active flows.
///
/// This keeps the currently active flows, cuts them into slices and gathers information passed to
/// it, updating the flows as it goes.
///
/// It produces two results. First, every time a slice is closed, the slice is returned (and is
/// left for the caller to take care of). The other is a stream of events happening on the flows to
/// allow other parts of the program to react to them.
pub struct Gather {
    /// The logger used to report activity.
    logger: Logger,
    /// The last time we closed a slice.
    last_close: SystemTime,
    /// Storage of all our active flows.
    flows: Vec<Rc<RefCell<FlowHead>>>,
    /// An index to look up flows by keys.
    ///
    /// # Note
    ///
    /// While a key of a flow needs to be unique, it is possible (and common) for a flow to have
    /// multiple keys. Therefore, a flow will be present here multiple times.
    index: BTreeMap<Key, Rc<RefCell<FlowHead>>>,
    /// Where we put generated events.
    event_sink: UnboundedSender<Event>,
}

impl Gather {
    /// Creates a new gather component.
    ///
    /// The component is empty (there are no flows active at the moment).
    ///
    /// It makes little sense to have multiple gather components, but it is not enforced in any
    /// way.
    ///
    /// # Params
    ///
    /// * `logger`: Where the gatherer shall log.
    /// * `event_sink`: A channel where the gatherer should put events generated by updates and
    ///   some other things. The receiver end can be freely closed, the gatherer simply ignores all
    ///   errors when sending there.
    pub fn new(logger: Logger, event_sink: UnboundedSender<Event>) -> Self {
        debug!(logger, "Creating new gather component");
        Gather {
            logger,
            last_close: SystemTime::now(),
            flows: Vec::new(),
            index: BTreeMap::new(),
            event_sink,
        }
    }
    /// Updates a flow.
    ///
    /// This applies the passed update to a flow. If the flow is not found, it creates a fresh new
    /// one.
    ///
    /// First, the keys are used to find the flow. If no key matches, a new flow is created.
    ///
    /// Then, all the keys that are not yet known are also linked to the flow (either found by
    /// other key or just created).
    ///
    /// All the tags passed in the update are added to the flow. If the tag is already present, the
    /// old one is replaced.
    ///
    /// If the statistics are present in the update, they are also applied.
    ///
    /// # Params
    ///
    /// * update: The description of update to apply.
    ///
    /// # Panics
    ///
    /// It is considered a contract violation if a key is used by multiple flows that are live at
    /// the same time. If such condition is discovered, it panics.
    ///
    /// Also, if an update with `Key::InternalHandle` and some other key is passed in, it may
    /// panic.
    ///
    /// # TODO
    ///
    /// Consider what happens when a flow is reported as terminated, we still hold it for a while
    /// and the ID is reused.  This is likely something that can happen (eg. the tupple gets
    /// reused, or conntrack ID is reused).
    pub fn update(&mut self, mut update: Update) {
        trace!(self.logger, "Applying update {:?}", update);
        // First, try to find the flow
        let mut flow = None;
        for key in &update.keys {
            if let Some(found) = self.index.get(key) {
                flow = Some(Rc::clone(found));
                break;
            } else if let Key::InternalHandle(_) = *key {
                warn!(self.logger, "An internal key references non-existent flow, ignoring");
                return;
            }
        }
        // If it's not there, add a fresh new one
        let flow = flow.unwrap_or_else(|| {
            trace!(self.logger, "Corresponding flow not found, adding a new one");
            let stats = update.stats
                .take() // Take it, as we don't need to update it with these stats once again
                .unwrap_or_default();
            let new = FlowHead::new(update.status, stats);
            // Every flow has at least that one internal key, by itself
            let internal_key = Key::InternalHandle(new.tags.clone());
            let flow_tags = new.tags.clone();
            let flow = Rc::new(RefCell::new(new));
            self.index.insert(internal_key, Rc::clone(&flow));
            self.flows.push(Rc::clone(&flow));
            drop(self.event_sink.unbounded_send(Event::FlowStarted(flow_tags)));
            flow
        });
        let mut inner = flow.borrow_mut();
        // Now, apply all the updates to whatever we either found or created
        // First, make sure we have all the keys present
        let key_cnt = update.keys.len();
        for key in update.keys.drain(..) {
            if let Key::InternalHandle(_) = key {
                if key_cnt > 1 {
                    panic!("An update with internal handle and some other key");
                }
            } else if !inner.keys.contains(&key) {
                inner.keys.insert(key.clone());
                assert!(self.index.insert(key.clone(), Rc::clone(&flow)).is_none(),
                        "Collision on a key {:?}", key);
            }
        }
        // Now, status and tags
        if inner.status != Status::End {
            inner.status = update.status;
            inner.inactive_for = 0;
        }
        if !update.tags.is_empty() {
            let event = Event::FlowUpdated {
                flow: inner.tags.clone(),
                columns: update.tags
                    .idents()
                    .cloned()
                    .collect(),
            };
            inner.tags.borrow_mut().extend(update.tags);
            drop(self.event_sink.unbounded_send(event));
        }
        // And if we got a new statistics, use them.
        if let Some(stats) = update.stats {
            let mut last = stats.clone();
            mem::swap(&mut last, &mut inner.last_stats);
            if stats.sub_safe(&last) {
                let difference = stats - last;
                inner.current_stats += difference;
            } else {
                warn!(self.logger,
                      "Stats {:?} are smaller than previous snapshot {:?}",
                      stats,
                      last);
            }
        }
    }
    /// Closes a slice.
    ///
    /// This slices all the active flows and produces flow slices for them.
    ///
    /// It also does some other book keeping, like timing out old flows and dropping them.
    pub fn close_slice(&mut self) -> TimeSlice {
        let logger = &self.logger;
        let index = &mut self.index;
        debug!(logger, "Closing a time slice");
        // Time management
        let now = SystemTime::now();
        let before = self.last_close;
        self.last_close = now;
        // Get rid of dead flows and make all the others older.
        trace!(logger, "Flow head bookkeeping");
        let sink = &self.event_sink;
        self.flows.retain(|head| {
            let mut h = head.borrow_mut();
            if h.status == Status::Start {
                trace!(logger, "Activating flow {:?}", h.tags);
                // Just started. Mark it as ongoing in the next interval, but active.
                h.status = Status::Ongoing;
                h.inactive_for = 1;
                true
            } else if h.status == Status::Ongoing && h.inactive_for >= SLICE_INACTIVE_LIMIT {
                trace!(logger, "Timing out an inactive flow {:?}", h.tags);
                // Inactive for too long, call it terminated.
                h.status = Status::End;
                h.inactive_for = 1;
                true
            } else if h.status == Status::End && h.inactive_for >= SLICE_KEEP_ENDED {
                trace!(logger, "Dropping an old flow {:?}", h.tags);
                // Remove them from the index
                for key in &h.keys {
                    if let Key::InternalHandle(_) = *key {
                        panic!("Internal handle in list of keys");
                    }
                    index.remove(key);
                }
                // The internal handle is *not* part of the keys (to avoid circular references and
                // save on memory).
                index.remove(&Key::InternalHandle(h.tags.clone()));
                drop(sink.unbounded_send(Event::FlowTerminated(h.tags.clone())));
                // Ended long time ago. Get rid of it.
                false
            } else {
                h.inactive_for += 1;
                true
            }
        });
        trace!(logger, "Creating flow slices");
        // Produce the flow slices and a time slice out of the data we have
        let slices = self.flows
            .iter()
            .filter_map(|head| {
                let mut h = head.borrow_mut();
                if h.status == Status::End && h.inactive_for > 1 {
                    // We want to report the ended flows just one last time, but preserve them for
                    // a while internally, so any stray data doesn't cause havoc
                    None
                } else {
                    let slice = h.slice(before, now);
                    Some(slice)
                }
            })
            .collect();
        TimeSlice::new(before, now, slices)
    }
    /// Pushes a cork through the event channel.
    ///
    /// Sometimes the caller needs to know when all existing events passed through the channel and
    /// were processed. To mark such place, a cork may be used. This method pushes such cork
    /// through the event channel.
    pub fn push_cork(&self, cork: Cork) {
        drop(self.event_sink.unbounded_send(Event::CorkRequest(cork)));
    }
}

impl ComputeQuery for Gather {
    fn query<'a, 'b, 'r>(&'a self, filter: &'b RefHeaders<'b>)
        -> Box<Iterator<Item = FlowTags> + 'r>
    where
        'a: 'r,
        'b: 'r,
    {
        let iter = self.flows
            .iter()
            .filter_map(move |flow| {
                // Layers upon layers of RefCells…
                let inner = flow.borrow();
                let tags = inner.tags.borrow();
                for (ident, values) in filter {
                    let value = tags.get(ident);
                    if !values.contains(&value) {
                        return None;
                    }
                }
                Some(inner.tags.clone())
            });
        Box::new(iter)
    }
}

#[cfg(test)]
mod tests {
    use futures::unsync::mpsc;
    use futures::{Future, Stream};

    use super::*;

    use libutils;
    use libdata::column::{Headers, Ident, Value, Type as ColumnType};
    use libdata::flow::{Bytes, Count, Direction, IpProto};
    use libdata::stats::Size as SizeStat;
    use libflow::update;
    use libquery::{InTimeInterval, ValueSrc};

    struct TestData {
        sizes: SizeStats,
    }

    impl TestData {
        fn new() -> Self {
            Self {
                sizes: SizeStats {
                    dir_in: SizeStat {
                        packets: Count(2),
                        size: Bytes(1024),
                    },
                    dir_out: SizeStat {
                        packets: Count(3),
                        size: Bytes(512),
                    },
                }
            }
        }
    }

    /// Tests creation and slicing of a flow.
    #[test]
    fn flow_head() {
        let data = TestData::new();
        // First, one where we missed the start
        let start = SystemTime::now();
        let head_aged = FlowHead::new(Status::Ongoing, data.sizes.clone());
        assert!(head_aged.keys.is_empty());
        assert_eq!(0, head_aged.tags.borrow().len());
        assert_eq!(data.sizes, head_aged.last_stats);
        assert_eq!(SizeStats::default(), head_aged.current_stats);
        assert_eq!(Status::Ongoing, head_aged.status);
        assert!(start <= head_aged.start_time && head_aged.start_time <= SystemTime::now());
        assert!(head_aged.end_time.is_none());
        assert_eq!(0, head_aged.inactive_for);
        // This might look a bit odd, as we create it as Ongoing, which means that it actually
        // started some time ago. However, the started_here means if our tracking of the flow
        // started in this slice, not the actual slice.
        assert!(head_aged.started_here);
        // Another one, where we've seen the start
        let mut head = FlowHead::new(Status::Start, data.sizes.clone());
        // As we start a new flow right now, it is not empty, but we count the statistics in
        assert_eq!(data.sizes, head.current_stats);
        assert_eq!(Status::Start, head.status);
        // Now when we cut the flow, we get a slice
        let end = start + Duration::from_secs(15);
        let slice = head.slice(start, end);
        // Check the slice looks fine
        assert_eq!((head.start_time, end), (&slice).interval());
        assert_eq!(&data.sizes.expand(head.start_time, end, Duration::from_secs(5), true),
                   (&slice).stats());
        // No tag there, so check one at random
        assert!((&slice).value(&Ident::of::<Direction>()).is_none());
        // Check how the slice looks like after the split.
        assert!(!head.started_here);
        assert_eq!(SizeStats::default(), head.current_stats);
        // The flow head shares the tag storage with the flow, so we can add tags later on.
        head.tags.borrow_mut().insert(Direction::In);
        // The tag appears on the flow slice
        assert_eq!(Value::from(Direction::In), (&slice).value(&Ident::of::<Direction>()).unwrap());
    }

    #[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
    struct Key1(usize);
    impl ColumnType for Key1 {
        fn name() -> String { "test-key-1".to_owned() }
    }

    #[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
    struct Key2(usize);
    impl ColumnType for Key2 {
        fn name() -> String { "test-key-2".to_owned() }
    }

    /// Test manipulating updates and flows in the gather.
    #[test]
    fn updates() {
        let data = TestData::new();
        let logger = libutils::test_logger();
        let (sender, receiver) = mpsc::unbounded();
        let mut iter = receiver.wait();
        let mut gather = Gather::new(logger, sender);
        assert!(gather.flows.is_empty());
        assert!(gather.index.is_empty());
        // Let's close a slice on an empty gather first
        let time = gather.close_slice();
        assert_eq!(0, (&time).into_iter().count());
        // Put a flow inside
        let init_f1 = Update {
            keys: vec![
                Key::Simple(Key1(1).into()),
            ],
            status: Status::Start,
            tags: Tags::new(),
            stats: Some(data.sizes.clone()),
        };
        gather.update(init_f1);
        // It's inside (it has an internal key in addition to the one provided)
        assert_eq!(1, gather.flows.len());
        assert_eq!(2, gather.index.len());
        assert_eq!(Status::Start, gather.flows[0].borrow().status);
        assert_eq!(1, gather.flows[0].borrow().keys.len());
        if let Some(Ok(Event::FlowStarted(flow))) = iter.next() {
            assert_eq!(0, flow.borrow().len());
        } else {
            panic!("Wrong event");
        }
        // As there are no tags inside, there's no update event after the start one
        // When we close it, we get one flow out
        let time = gather.close_slice();
        assert_eq!(1, (&time).into_iter().count());
        // The flow inside is no longer new
        assert_eq!(Status::Ongoing, gather.flows[0].borrow().status);
        assert!(!gather.flows[0].borrow().started_here);
        // The statistics are now empty, as they got used for that flow
        assert_eq!(SizeStats::default(), gather.flows[0].borrow().current_stats);
        assert_eq!(1, gather.flows[0].borrow().inactive_for);
        // If we another key and some tags, it gets there
        let both_keys = vec![
            Key::Simple(Key1(1).into()),
            Key::Simple(Key2(2).into()),
        ];
        let mut tags = Tags::new();
        tags.insert(Direction::In);
        let update_f1 = Update {
            keys: both_keys.clone(),
            status: Status::Ongoing,
            tags,
            stats: None,
        };
        gather.update(update_f1);
        // It's still the same flow
        assert_eq!(1, gather.flows.len());
        // But by touching it gets active now again
        assert_eq!(0, gather.flows[0].borrow().inactive_for);
        assert_eq!(2, gather.flows[0].borrow().keys.len());
        // But it has 2 different keys
        assert_eq!(3, gather.index.len());
        // And the flow (even its previous slices) has the new tag.
        let ident = Ident::of::<Direction>();
        for flow in &time {
            assert_eq!(Value::from(Direction::In), flow.value(&ident).unwrap());
        }
        if let Some(Ok(Event::FlowUpdated { flow, columns })) = iter.next() {
            assert_eq!(1, flow.borrow().len());
            assert_eq!(1, columns.len());
            assert!(columns.contains(&ident));
        } else {
            panic!("Wrong event");
        }
        // Now, let's put another update there, this time with stats (the same ones)
        let update_f1 = Update {
            keys: both_keys.clone(),
            status: Status::Ongoing,
            tags: Tags::new(),
            stats: Some(data.sizes.clone()),
        };
        gather.update(update_f1);
        // The stats of the flow don't change
        assert_eq!(SizeStats::default(), gather.flows[0].borrow().current_stats);
        assert_eq!(data.sizes, gather.flows[0].borrow().last_stats);
        // Now, create some larger stats
        let larger = data.sizes.clone() + data.sizes.clone();
        let update_f1 = Update {
            // We use just a single key, so we check they are interchangeable
            keys: vec![
                Key::Simple(Key2(2).into()),
            ],
            status: Status::Ongoing,
            tags: Tags::new(),
            stats: Some(larger.clone()),
        };
        gather.update(update_f1);
        // Check nothing broke with using the single key
        assert_eq!(1, gather.flows.len());
        assert_eq!(3, gather.index.len());
        // The amount of data is larger in the flow now
        assert_eq!(data.sizes, gather.flows[0].borrow().current_stats);
        assert_eq!(larger, gather.flows[0].borrow().last_stats);
        // These last two updates didn't provoke an event, as they didn't change the flow head.
        // Let's add another flow in there.
        let mut tags2 = Tags::new();
        tags2.insert(IpProto::Tcp);
        tags2.insert(Direction::In);
        let init_f2 = Update {
            keys: vec![
                Key::Simple(Key1(2).into()),
            ],
            status: Status::Ongoing,
            tags: tags2,
            stats: None,
        };
        gather.update(init_f2);
        // The new flow gets born
        assert_eq!(2, gather.flows.len());
        assert_eq!(5, gather.index.len());
        {
            let f = gather.flows[1].borrow();
            assert_eq!(SizeStats::default(), f.current_stats);
            assert_eq!(SizeStats::default(), f.last_stats);
            assert_eq!(Status::Ongoing, f.status);
            assert_eq!(0, f.inactive_for);
        }
        // We get *two* events now, as that creates a new flow and produces an update to it
        if let Some(Ok(Event::FlowStarted(flow))) = iter.next() {
            // Even the first one already contains the relevant columns
            let borrow = flow.borrow();
            assert_eq!(2, borrow.len());
            assert!(borrow.contains(&Ident::of::<Direction>()));
            assert!(borrow.contains(&Ident::of::<IpProto>()));
        } else {
            panic!("Wrong event");
        }
        if let Some(Ok(Event::FlowUpdated { flow, columns })) = iter.next() {
            // Even the first one already contains the relevant columns
            let borrow = flow.borrow();
            assert_eq!(2, borrow.len());
            assert!(borrow.contains(&Ident::of::<Direction>()));
            assert!(borrow.contains(&Ident::of::<IpProto>()));
            assert_eq!(2, columns.len());
            assert!(columns.contains(&Ident::of::<Direction>()));
            assert!(columns.contains(&Ident::of::<IpProto>()));
        } else {
            panic!("Wrong event");
        }
        // Now, we close the first flow.
        let close_f1 = Update {
            keys: both_keys.clone(),
            status: Status::End,
            tags: Tags::new(),
            stats: None,
        };
        gather.update(close_f1);
        // It is still present, but closed
        assert_eq!(2, gather.flows.len());
        assert_eq!(5, gather.index.len());
        assert_eq!(Status::End, gather.flows[0].borrow().status);
        assert_eq!(0, gather.flows[0].borrow().inactive_for);
        // Now, close it again. We get two flows out.
        let time = gather.close_slice();
        assert_eq!(2, (&time).into_iter().count());
        // And both flows are still present in there. We can even touch the first flow and nothing
        // interesting happens.
        assert_eq!(2, gather.flows.len());
        assert_eq!(5, gather.index.len());
        // The closing doesn't produce an event (yet). We want to check nothing is produced right
        // now, therefore we cork it and check the cork.
        gather.push_cork(update::cork().0);
        if let Some(Ok(Event::CorkRequest(_))) = iter.next() { } else {
            panic!("There should have been a cork");
        }
        let dead_f1 = Update {
            keys: both_keys.clone(),
            // Even Ongoing status won't wake it up
            status: Status::Ongoing,
            tags: Tags::new(),
            stats: None,
        };
        gather.update(dead_f1);
        assert_eq!(2, gather.flows.len());
        assert_eq!(5, gather.index.len());
        assert_eq!(Status::End, gather.flows[0].borrow().status);
        // When we close again, we get just the other flow.
        let mut time = gather.close_slice();
        assert_eq!(1, (&time).into_iter().count());
        // Now, a lot of time passes (eg. it is closed many times). The first closed flow drops
        // out.
        for _ in 0..SLICE_KEEP_ENDED {
            time = gather.close_slice();
        }
        // Now it contains just one flow
        assert_eq!(1, (&time).into_iter().count());
        assert_eq!(1, gather.flows.len());
        assert_eq!(2, gather.index.len());
        assert!(gather.index.contains_key(&Key::Simple(Key1(2).into())));
        assert_eq!(1, gather.flows[0].borrow().keys.len());
        // And we get an event about terminated flow
        if let Some(Ok(Event::FlowTerminated(flow))) = iter.next() {
            // It's the first flow, that has a single tag.
            assert_eq!(1, flow.borrow().len());
        } else {
            panic!("Wrong event");
        }
        // And when we wait even longer, even the other flow drops out, because it is inactive.
        // (it was already inactive during the first waiting, so now it is inactive + keep ended)
        for _ in 0..SLICE_INACTIVE_LIMIT {
            time = gather.close_slice();
        }
        assert_eq!(0, (&time).into_iter().count());
        assert_eq!(0, gather.flows.len());
        assert_eq!(0, gather.index.len());
        // The other flow produces an event
        if let Some(Ok(Event::FlowTerminated(flow))) = iter.next() {
            assert_eq!(2, flow.borrow().len());
        } else {
            panic!("Wrong event");
        }
        // Now when we drop the gather and get an end of the stream, there are no more events
        drop(gather);
        assert!(iter.next().is_none());
    }

    /// An update with internal key is ignored if the flow doesn't exist
    #[test]
    fn internal_key_missing() {
        let key = Key::InternalHandle(FlowTags::new(Tags::new()));
        let update = Update {
            keys: vec![key],
            status: Status::Start,
            tags: Tags::new(),
            stats: None,
        };
        let logger = libutils::test_logger();
        let (sender, receiver) = mpsc::unbounded();
        let mut gather = Gather::new(logger, sender);
        gather.update(update);
        assert_eq!(0, gather.flows.len());
        assert_eq!(0, gather.index.len());
        drop(gather); // So the sender part is closed
        assert_eq!(0, receiver.wait().count());
    }

    /// Tests corks are properly inserted where we ask them to.
    #[test]
    fn gather_cork() {
        let data = TestData::new();
        let logger = libutils::test_logger();
        let (sender, receiver) = mpsc::unbounded();
        let mut gather = Gather::new(logger, sender);
        // Put a flow inside
        let init_f1 = Update {
            keys: vec![
                Key::Simple(Key1(1).into()),
            ],
            status: Status::Start,
            tags: Tags::new(),
            stats: Some(data.sizes.clone()),
        };
        gather.update(init_f1);
        // Push a cork after the update
        let (cork, handle) = update::cork();
        gather.push_cork(cork);
        // Now take out all the events and compare them.
        let mut iter = receiver.wait();
        if let Some(Ok(Event::FlowStarted(_))) = iter.next() { } else {
            panic!("The first event should be FlowStarted");
        }
        // This also drops the cork
        if let Some(Ok(Event::CorkRequest(_))) = iter.next() { } else {
            panic!("The second event should be a CorkRequest");
        }
        handle.wait().unwrap();
        // The stream terminates when we drop the gather and there are no more events
        drop(gather);
        assert!(iter.next().is_none());
    }

    /// Tests the query interface for internal computations
    #[test]
    fn query() {
        let logger = libutils::test_logger();
        let (sender, _receiver) = mpsc::unbounded();
        let mut gather = Gather::new(logger, sender);
        // First, a smoke test on an empty gather, that nothing breaks.
        let empty_query = Headers::new();
        let empty_query = RefHeaders::from(&empty_query);
        assert_eq!(0, gather.query(&empty_query).count());
        // Put some stuff inside the gather.
        let flows: Vec<Vec<Value>> = vec![
            vec![Direction::In.into()],
            vec![Direction::In.into()],
            vec![Direction::Out.into()],
            vec![IpProto::Tcp.into()],
            vec![Direction::In.into(), IpProto::Tcp.into()],
            vec![],
        ];
        for (id, flow) in flows.into_iter().enumerate() {
            let mut tags = Tags::new();
            for v in flow {
                tags.insert(v);
            }
            let update = Update {
                keys: vec![
                    Key::Simple(Key1(id).into()),
                ],
                status: Status::Start,
                tags,
                stats: None,
            };
            gather.update(update);
        }
        // Just a sanity check we inserted correctly
        assert_eq!(6, gather.flows.len());
        // If the query is empty, there are no restrictions and we get everything
        assert_eq!(6, gather.query(&empty_query).count());
        // The direction is either In or unset [0, 1, 3, 4, 5]
        let mut in_unset = Headers::new();
        in_unset.insert(Direction::In);
        in_unset.insert_empty(Ident::of::<Direction>());
        let in_unset = RefHeaders::from(&in_unset);
        assert_eq!(5, gather.query(&in_unset).count());
        // Combine two criteria [5]
        let mut two_crits = Headers::new();
        two_crits.insert(Direction::In);
        two_crits.insert(IpProto::Tcp);
        let two_crits = RefHeaders::from(&two_crits);
        assert_eq!(1, gather.query(&two_crits).count());
        // This one doesn't match anything
        let mut tcp_out = Headers::new();
        tcp_out.insert(Direction::Out);
        tcp_out.insert(IpProto::Tcp);
        let tcp_out = RefHeaders::from(&tcp_out);
        assert_eq!(0, gather.query(&tcp_out).count());
        // Check the content of one random returned value
        let single = gather.query(&two_crits).next().unwrap();
        let borrowed = single.borrow();
        assert_eq!(2, borrowed.len());
        assert_eq!(Some(&Value::from(IpProto::Tcp)), borrowed.get(&Ident::of::<IpProto>()));
        assert_eq!(Some(&Value::from(Direction::In)), borrowed.get(&Ident::of::<Direction>()));
    }
}