// 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/>.

//! Column descriptions and values.
//!
//! A column is a kind of value tagged onto a flow (o part of it). As these can be arbitrary data,
//! this module contains an abstraction allowing to manipulate them.
//!
//! The basic type is a [`Value`](struct.Value.html), which represents one opaque tag. The other
//! important type is [`Ident`](struct.Ident.html). That one describes the kind (identity) of a
//! value. In a nutshell, `Ident` can talk about a remote IP address, while `Value` holds an actual
//! instance of that IP address. Of course, a `Value` knows its own identity.
//!
//! To use a data type as a public tag, two things need to be performed:
//!
//! * Implement the `Type` trait for it. This allows creation of values out of this type and
//!   having an identity for it.
//! * Register it inside a global registry. It can either be done here in the source, if it is
//!   globally available (eg. multiple modules might want to use it, like the flow direction), or
//!   it can be registered from the outside by [`register`](fn.register.html) or
//!   [`register_endpoint`](fn.register_endpoint.html).
//!
//! If the column is to be used only as a key for internal use (eg. users won't be able to query
//! for it), the registration is not needed.
//!
//! There are also some other helpful types that build on top of these.

use std::any::{Any, TypeId};
use std::borrow::Borrow;
use std::cell::RefCell;
use std::cmp::Ordering;
use std::collections::{hash_map, BTreeSet, HashMap};
use std::fmt::{Debug, Formatter, Result as FmtResult};
use std::hash::{Hash, Hasher};
use std::marker::PhantomData;
use std::net::IpAddr;
use std::ops::Deref;
use std::rc::Rc;
use std::sync::RwLock;

use eui48::MacAddress;
use erased_serde::{self, Deserializer as EDeserializer, Error as EError, Serialize as ESerialize};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde::de::{DeserializeSeed, Error as DeError, MapAccess, SeqAccess, Visitor};

use flow::{Direction, IpProto, IpProtoRaw, MacName, Name, Names, Port};

/// A trait for the types that can be made into the generic [`Value`](struct.Value.html)
///
/// Implementing this trait allows a type to act as the generic opaque `Value`. This mostly means
/// providing some other common traits, but also assigning a name for the tag, which is used for
/// serializing and deserializing the tag's identity.
///
/// Note that there can be only one tag of each type. Therefore, if a flow needs to be tagged with
/// something that is eg. a `String`, create a newtype for that.
///
/// Some data are relevant for the local and remote endpoints. There are the
/// [`Local`](struct.Local.html) and [`Remote`](struct.Remote.html) wrappers for that and an
/// [`EndpointType`](trait.EndpointType.html) trait.
///
/// Note that for some actions to be usable for a type, the type needs to be registered in
/// addition to implementing this trait. See [`register`](fn.register.html).
// https://doc.rust-lang.org/nomicon/hrtb.html for the magic with `for`
pub trait Type
where
    Self: Clone + Debug + for<'de> Deserialize<'de> + Ord + Serialize + Sized + 'static
{
    /// A name of the tag.
    ///
    /// The name is used when serializing to JSON. It must be unique between all registered types.
    fn name() -> String;
}

/// A trait to implement endpoint types.
///
/// The endpoint types for type `I` are simply [`Local<I>`](struct.Local.html) and
/// [`Remote<I>`](struct.Remote.html). But as these wrappers and the [`Type`](trait.Type.html) are
/// in this crate, it is not possible to implement the `Type` trait directly by another crate.
///
/// This works around for that problem, by implementing the `Type` trait for both wrappers for `I:
/// EndpointType`.
pub trait EndpointType
where
    Self: Clone + Debug + for<'de> Deserialize<'de> + Ord + Serialize + Sized + 'static
{
    /// The name of the bare tag.
    ///
    /// The tag will be prepended with `"local-"` and `"remote-"` respectively for the wrappers.
    fn name() -> String;
}

impl<I: EndpointType> Type for Local<I> {
    fn name() -> String {
        format!("local-{}", I::name())
    }
}

impl<I: EndpointType> Type for Remote<I> {
    fn name() -> String {
        format!("remote-{}", I::name())
    }
}

/// A wrapper for local-endpoint columns.
///
/// If there's an information that ties to each endpoint of communication (eg. an IP address), use
/// this generic newtype to create the local-endpoint version of the column.
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
pub struct Local<T>(pub T);
/// A wrapper for remote-endpoint columns.
///
/// If there's an information that ties to each endpoint of communication (eg. an IP address), use
/// this generic newtype to create the remote-endpoint version of the column.
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
pub struct Remote<T>(pub T);

macro_rules! coln {
    ($($name: expr => $ty: ty;)*) => {
        $(
            impl Type for $ty {
                fn name() -> String { $name.to_owned() }
            }
        )*
    }
}

macro_rules! endp {
    ($($name: expr => $ty: ty;)*) => {
        $(
            impl EndpointType for $ty {
                fn name() -> String { $name.to_owned() }
            }
        )*
    }
}

// Implementation of the Type trait for bunch of basic types
coln! {
    "ip-proto" => IpProto;
    "ip-proto-raw" => IpProtoRaw;
    "direction" => Direction;
}

endp! {
    "mac" => MacAddress;
    "mac-name" => MacName;
    "ip" => IpAddr;
    "name-primary" => Name;
    "name-set" => Names;
    "port" => Port;
}

/// A helper trait to do the runtime reflection.
///
/// The `Value` is just a nicely clothed trait object inside. But because a trait object can access
/// methods only of a single trait (not of any traits the one trait depends on), we cheat a bit
/// here. We create a helper trait that encompasses all the functionality we need and smuggle the
/// calls to the other traits through that.
///
/// We implement it automatically for the `Type` objects.
trait Inner: Any {
    /// Turn into the Any trait object.
    ///
    /// That's the only one that can be downcasted.
    fn as_any(&self) -> &Any;
    /// Provide identity of the data.
    fn ident(&self) -> Ident {
        Ident(TypeId::of::<Self>())
    }
    /// Perform a comparison with onether object.
    ///
    /// Note that the other object doesn't have to be of the same type. It is expected the
    /// implementation compares the identities in such case.
    fn cmp(&self, other: &Inner) -> Ordering;
    /// Create a new boxed trait object.
    fn clone(&self) -> Box<Inner>;
    /// Perform the debug formatting.
    fn debug(&self, f: &mut Formatter) -> FmtResult;
    /// Return an erased Serialize trait object.
    fn serde(&self) -> &ESerialize;
}

/// The actual implementation of the helper trait for the column types.
impl<I: Type> Inner for I {
    fn as_any(&self) -> &Any {
        self
    }
    fn cmp(&self, other: &Inner) -> Ordering {
        let self_id = self.ident();
        let other_id = other.ident();
        match self_id.cmp(&other_id) {
            Ordering::Equal =>  {
                // If they are of the same type, we compare the values
                let other_self = other
                    .as_any()
                    .downcast_ref::<Self>()
                    .unwrap();
                Ord::cmp(self, other_self)
            },
            // Otherwise, we compare the identities (types)
            idcmp => idcmp,
        }
    }
    fn clone(&self) -> Box<Inner> {
        Box::new(Clone::clone(self))
    }
    fn debug(&self, f: &mut Formatter) -> FmtResult {
        Debug::fmt(self, f)
    }
    fn serde(&self) -> &ESerialize {
        self
    }
}

/// Another helper trait.
///
/// This acts in a similar way as the [`Inner`](trait.Inner.html), except we want to be able to
/// deserialize values before we have an instance of them. Therefore we have „dummy“ factory trait
/// objects to do that.
trait Decoder {
    fn deserialize(&self, deserializer: &mut EDeserializer) -> Result<Option<Value>, EError>;
}

/// A dummy factory for the `Decoder` trait.
struct Dec<C>(PhantomData<*const C>);

// It's safe, we only have that phantom data to carry the type relation
unsafe impl<C> Sync for Dec<C> {}
unsafe impl<C> Send for Dec<C> {}

impl<C: Type> Decoder for Dec<C> {
    fn deserialize(&self, deserializer: &mut EDeserializer) -> Result<Option<Value>, EError> {
        let column = erased_serde::deserialize::<Option<C>>(deserializer)?;
        Ok(column.map(Value::from))
    }
}

/// A registrar of the different column types and their types.
///
/// Used internally to store all the known types and convert the names, etc.
struct Registrar {
    /// A conversion table from identity to the name.
    id2name: HashMap<Ident, String>,
    /// A conversion table from name to the identity.
    name2id: HashMap<String, Ident>,
    /// A table of factory trait objects to deserialize values of the given column.
    id2build: HashMap<Ident, Box<Decoder + Send + Sync>>,
}

impl Registrar {
    /// A constructor of an empty registrar.
    fn new() -> Self {
        Self {
            id2name: HashMap::new(),
            name2id: HashMap::new(),
            id2build: HashMap::new(),
        }
    }
    /// Registers a new column type.
    ///
    /// # Panics
    ///
    /// If the same type or name is already taken. However, it is possible to register the same
    /// type multiple times.
    fn reg<C: Type>(&mut self) {
        let name = C::name();
        let ident = Ident::of::<C>();
        let builder = Box::new(Dec(PhantomData::<*const C>));
        if let Some(old_name) = self.id2name.insert(ident, name.clone()) {
            assert_eq!(name, old_name, "The same type already registered under a different name");
        }
        if let Some(old_ident) = self.name2id.insert(name, ident) {
            assert_eq!(ident, old_ident,
                       "Some other type already registered under the same name");
        }
        self.id2build.insert(ident, builder);
    }
    /// Register a new endpoint column type.
    ///
    /// Both `Local<I>` and `Remote<I>` are registered.
    ///
    /// # Panics
    ///
    /// If the same type or name is already taken.
    fn rege<I: EndpointType>(&mut self) {
        self.reg::<Local<I>>();
        self.reg::<Remote<I>>();
    }
    /// Looks up a decoder for the given ident.
    ///
    /// # Panics
    ///
    /// If the ident is not known. The corresponding column type hasn't been registered.
    fn decoder(&self, i: &Ident) -> &(Decoder + Sync + Send) {
        self.id2build
            .get(i)
            .expect("Ident not registered")
            .borrow()
    }
    /// Looks up a name for the given ident.
    ///
    /// # Panics
    ///
    /// If the ident is not known. The corresponding column type hasn't been registered.
    fn name(&self, i: &Ident) -> &str {
        self.id2name.get(i).expect("Ident not registered")
    }
    /// Looks up an ident for the given name.
    ///
    /// Unlike the other look up methods, this one returns `Err` in case it is not known, since
    /// such names may come from outside of the application.
    // It is OK to use () as the error type here, as this is just a private interface.
    fn ident(&self, n: &str) -> Result<Ident, ()> {
        self.name2id
            .get(n)
            .cloned()
            .ok_or(())
    }
}

lazy_static! {
    /// The global [`Registrar`](struct.Registrar.html).
    static ref REGISTRAR: RwLock<Registrar> = {
        RwLock::new(Registrar::new())
    };
}

/// Register a type to be used as a column.
///
/// Some functions (eg. querying for the data by a user) need the type used as a column to be
/// registered, in addition to implementing the [`Type`](trait.Type.html) trait.
///
/// Also, if you want a type that is present on both local and remote endpoint, see the
/// [`register_endpoint`](fn.register_endpoint.html).
///
/// ```
/// # #[macro_use]
/// # extern crate serde_derive;
/// # extern crate libdata;
/// #
/// # use libdata::column::{self, Type};
/// #
/// # fn main() {
/// #[derive(Clone, Debug, Deserialize, Eq, Ord, Serialize, PartialEq, PartialOrd)]
/// struct SomeColumn(String);
///
/// impl Type for SomeColumn {
///     fn name() -> String {
///         "some-column".to_owned()
///     }
/// }
///
/// column::register::<SomeColumn>();
/// # }
/// ```
pub fn register<C: Type>() {
    REGISTRAR.write()
        .unwrap()
        .reg::<C>();
}

/// Register a type to be used as endpoint columns.
///
/// This is similar to [`register`](fn.register.html), but for type `I` passed it registers
/// `Local<I>` and `Remote<I>`. This helps when the value shall be present on each endpoint of a
/// flow.
///
/// ```
/// # #[macro_use]
/// # extern crate serde_derive;
/// # extern crate libdata;
/// #
/// # use libdata::column::{self, EndpointType};
/// #
/// # fn main() {
/// #[derive(Clone, Debug, Deserialize, Eq, Ord, Serialize, PartialEq, PartialOrd)]
/// struct SomeColumn(String);
///
/// impl EndpointType for SomeColumn {
///     fn name() -> String {
///         "some-column".to_owned()
///     }
/// }
///
/// column::register_endpoint::<SomeColumn>();
/// # }
/// ```
pub fn register_endpoint<I: EndpointType>() {
    REGISTRAR.write()
        .unwrap()
        .rege::<I>();
}

/// An identity of a column.
///
/// While the [`Value`](struct.Value.html) structure holds the actual data (one „cell“ of the
/// table), this is something like the name of the column. This specifies we want eg. a remote IP
/// address without having the concrete IP address in hand.
#[derive(Copy, Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct Ident(TypeId);

impl Ident {
    /// Decodes a value of the corresponding column type.
    ///
    /// It also can decode `null` as the `None` variant, eg. it assumes the value is optional.
    ///
    /// # Panics
    ///
    /// If the ident hasn't been registered.
    pub fn decode_value<'de, D>(&self, des: D) -> Result<Option<Value>, D::Error>
    where
        D: Deserializer<'de>
    {
        REGISTRAR.read()
            .unwrap()
            .decoder(self)
            .deserialize(&mut EDeserializer::erase(des))
            .map_err(D::Error::custom)
    }
    /// Provides the identity of the given column type.
    pub fn of<C: Type>() -> Self {
        Ident(TypeId::of::<C>())
    }
}

impl Serialize for Ident {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        REGISTRAR.read()
            .unwrap()
            .name(self)
            .serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for Ident {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let s: String = Deserialize::deserialize(deserializer)?;
        static VARIANTS: &[&'static str] = &["..."];
        REGISTRAR.read()
            .unwrap()
            .ident(&s)
            .map_err(|()| D::Error::unknown_variant(&s, VARIANTS))
    }
}

/// A column value.
///
/// The value represents some bit of information. It is an abstract handle, meaning the concrete
/// type is not known from the outside. It is possible to ask it for its type identity, so it's
/// possible to find out if two values are of the same type or ask if it is of a specific type. It
/// is possible to get access to the typed data (if the type is known and matches), but it should
/// be used only when necessary. Prefer the generic handling when possible.
///
/// However, it implements a lot of traits that allow for manipulation in generic manner.
///
/// This allows tagging flows with data without knowing the types of information used to tag it
/// upfront.
///
/// Note that the values act sane in regards to the normal traits. Eg. the `Ord` implementation can
/// compare even values of different underlying types.
pub struct Value(Box<Inner>);

impl Value {
    /// Provides the identity of the column type.
    pub fn ident(&self) -> Ident {
        self.0.ident()
    }
    /// Tries to turn the value into the type held inside.
    ///
    /// This allows accessing the data held inside as the type it actually is. It returns `None` if
    /// the requested type doesn't match the actual one.
    ///
    /// Generally, you should use this only if you know the type. Most of the handling can be done
    /// in a generic way, without extracting the data inside.
    pub fn downcast_ref<T: 'static>(&self) -> Option<&T> {
        self.0.as_any().downcast_ref()
    }
}

impl Clone for Value {
    fn clone(&self) -> Self {
        Value(self.0.clone())
    }
}

impl Debug for Value {
    fn fmt(&self, f: &mut Formatter) -> FmtResult {
        self.0.debug(f)
    }
}

impl Eq for Value {}

impl<C: Type> From<C> for Value {
    fn from(c: C) -> Self {
        Value(Box::new(c))
    }
}

impl Ord for Value {
    fn cmp(&self, other: &Self) -> Ordering {
        self.0.cmp(other.0.as_ref())
    }
}

impl PartialEq for Value {
    fn eq(&self, other: &Self) -> bool {
        self.cmp(other) == Ordering::Equal
    }
}

impl PartialOrd for Value {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Serialize for Value {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        self.0.serde().serialize(serializer)
    }
}

/// Tags of a single entity.
///
/// This represents a set of tags. However, the set contains at most one value of each type. This
/// is well suited for things like the *header* information about flows, for example.
///
/// This data storage is cummulative ‒ it is not possible to remove tags (only to replace them).
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct Tags(HashMap<Ident, Value>);

impl Tags {
    /// Creates a new tag storage.
    pub fn new() -> Self {
        Tags(HashMap::new())
    }
    /// Inserts a value into the set.
    ///
    /// If a value of that column type is already present, the old one is replaced and returned.
    pub fn insert<V: Into<Value>>(&mut self, value: V) -> Option<Value> {
        let value = value.into();
        self.0.insert(value.ident(), value)
    }
    /// Checks if a tag of the given type is present.
    pub fn contains(&self, ident: &Ident) -> bool {
        self.0.contains_key(ident)
    }
    /// Returns the tag value.
    ///
    /// This looks up the value of the given type, if any is present.
    pub fn get(&self, ident: &Ident) -> Option<&Value> {
        self.0.get(ident)
    }
    /// Consumes `other` and inserts all its content into `self`.
    ///
    /// If both `self` and `other` contain value of the same type, the one in `self` is
    /// overwritten and `other` wins.
    ///
    /// # Params
    ///
    /// * `other`: The other tag storage to incorporate into `self`.
    pub fn extend(&mut self, other: Self) {
        self.0.extend(other.0)
    }
    /// Returns how many tags there are inside.
    pub fn len(&self) -> usize {
        self.0.len()
    }
    /// Checks if the storage is empty.
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
    /// Iterates over the identities in the tags.
    pub fn idents(&self) -> hash_map::Keys<Ident, Value> {
        self.0
            .keys()
    }
}

/// A reference counted handle to [`Tags`](struct.Tags.html)
///
/// This is an easily clonable handle towards `Tags`. The `Tags` inside can be modified.
///
/// Note that comparison and hash identity is based on the address of the tags. Eg. it doesn't
/// matter what they contain, only the handles referencing the same tags are considered equal and
/// the hash and comparison order doesn't change when modifying the content. This allows using the
/// `FlowTags` as an unique handle to a flow.
///
/// Due to the `Defer` trait it can be manipulated simply as the internal `RefCell<Tags>`.
#[derive(Clone, Debug)]
pub struct FlowTags(Rc<RefCell<Tags>>);

impl FlowTags {
    /// Creates a new FlowTags out of simple tags.
    pub fn new(tags: Tags) -> Self {
        FlowTags(Rc::new(RefCell::new(tags)))
    }
}

impl Deref for FlowTags {
    type Target = Rc<RefCell<Tags>>;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl PartialEq for FlowTags {
    fn eq(&self, other: &Self) -> bool {
        Rc::ptr_eq(self, other)
    }
}

impl Eq for FlowTags {}

impl Hash for FlowTags {
    fn hash<H: Hasher>(&self, hasher: &mut H) {
        let addr = self.0.as_ref() as *const _;
        addr.hash(hasher);
    }
}

impl PartialOrd for FlowTags {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for FlowTags {
    fn cmp(&self, other: &Self) -> Ordering {
        let self_addr = self.0.as_ref() as *const _;
        let other_addr = other.0.as_ref() as *const _;
        self_addr.cmp(&other_addr)
    }
}

/// A set of values of the same type.
///
/// This is an inner type of [`Headers`](struct.Headers.html). It is expected that the set contains
/// only values of a single type. However, it is not enforced at the type level now (but unless you
/// create it yourself, you can assume this property).
pub type Header = BTreeSet<Option<Value>>;

/// A multi-set of tags.
///
/// This is similar to the [`Tags`](struct.Tags.html) type, but allows having multiple values of
/// each type (or even an empty slot in each type). This is what you'd get when you combine data
/// from multiple flows in additive manner.
///
/// However, values are deduplicated ‒ eg. there may be multiple remote IP addresses, but each one
/// only in one instance.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
pub struct Headers(HashMap<Ident, Header>);

impl Headers {
    /// Creates a new empty `Headers`.
    pub fn new() -> Self {
        Headers(HashMap::new())
    }
    /// Inserts an optional value into the set.
    ///
    /// Returns false if it is already present. The contained value is not updated.
    ///
    /// # Panics
    ///
    /// If `Some(value)` is inserted and the ident doesn't match.
    pub fn insert_opt(&mut self, ident: Ident, value: Option<Value>) -> bool {
        if let Some(ref val) = value {
            assert_eq!(ident, val.ident());
        }
        self.0
            .entry(ident)
            .or_insert_with(Header::new)
            .insert(value)
    }
    /// Inserts another value.
    ///
    /// If the value is already present, it is not updated and `false` is returned instead.
    pub fn insert<V: Into<Value>>(&mut self, value: V) -> bool {
        let value = value.into();
        let ident = value.ident();
        self.insert_opt(ident, Some(value))
    }
    /// Inserts a hole of the given type.
    ///
    /// Each type of column can contain a single `None` value. This inserts the `None` into the
    /// given type. Returns `false` if it was already present.
    pub fn insert_empty(&mut self, ident: Ident) -> bool {
        self.insert_opt(ident, None)
    }
    /// Iterates through the set by types.
    pub fn iter(&self) -> hash_map::Iter<Ident, Header> {
        self.0.iter()
    }
}

// This one is a bit tricky. That's because we deserialize the `Value` objects deep down. But it
// needs a hint to what type to expect.
//
// The hint is provided by the key in the map (the key is turned into an Ident). We need to pass
// the preparsed ident as the hint two levels down (though the array and into the value) to parse
// it. So that's the reason for the temporary Deserialize/DeserializeSeed/Visitor objects around.
impl<'de> Deserialize<'de> for Headers {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        struct SeedSingle<'a>(Ident, &'a mut Headers);
        impl<'a, 'de> DeserializeSeed<'de> for SeedSingle<'a> {
            type Value = ();
            fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
            where
                D: Deserializer<'de>
            {
                let val = self.0.decode_value(deserializer)?;
                self.1.insert_opt(self.0, val);
                Ok(())
            }
        }

        struct SeedHeader<'a>(Ident, &'a mut Headers);
        impl<'a, 'de> Visitor<'de> for SeedHeader<'a> {
            type Value = ();
            fn expecting(&self, formatter: &mut Formatter) -> FmtResult {
                write!(formatter, "a sequence of {:?}", self.0)
            }
            fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<(), A::Error> {
                while let Some(()) = seq.next_element_seed(SeedSingle(self.0, self.1))? {}
                Ok(())
            }
        }
        impl<'a, 'de> DeserializeSeed<'de> for SeedHeader<'a> {
            type Value = ();
            fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
            where
                D: Deserializer<'de>
            {
                deserializer.deserialize_seq(self)
            }
        }

        struct Map;
        impl<'de> Visitor<'de> for Map {
            type Value = Headers;
            fn expecting(&self, formatter: &mut Formatter) -> FmtResult {
                formatter.write_str("an object with flow sets")
            }
            fn visit_map<A: MapAccess<'de>>(self, mut access: A) -> Result<Headers, A::Error> {
                let mut res = Headers::new();
                while let Some(ident) = access.next_key::<Ident>()? {
                    access.next_value_seed(SeedHeader(ident, &mut res))?;
                }
                Ok(res)
            }
        }

        deserializer.deserialize_map(Map)
    }
}

impl<'a> IntoIterator for &'a Headers {
    type Item = (&'a Ident, &'a Header);
    type IntoIter = hash_map::Iter<'a, Ident, Header>;
    fn into_iter(self) -> hash_map::Iter<'a, Ident, Header> {
        self.iter()
    }
}

/// Just like [`Header`](type.Header.html), but with references instead of owned values.
///
/// This allows looking up values when only `Option<&Value>` is avaliable without cloning the
/// value.
pub type RefHeader<'a> = BTreeSet<Option<&'a Value>>;

/// Just like [`Headers`](struct.Headers.htlm), but with references instead of owned values.
///
/// This allows looking up values when only `Option<&Value>` is available.
///
/// # Examples
///
/// ```
/// # extern crate libdata;
/// # use std::net::*;
/// # use libdata::column::*;
/// # fn main() {
/// let val = Value::from(Local("127.0.0.1".parse::<IpAddr>().unwrap()));
/// let mut headers = Headers::new();
/// headers.insert(val.clone());
///
/// let ref_headers = RefHeaders::from(&headers);
/// assert_eq!(1, ref_headers.iter().count());
///
/// let some_val = Some(val);
/// let ref_val = some_val.as_ref();
/// for (ident, vals) in &ref_headers {
///     assert_eq!(Ident::of::<Local<IpAddr>>(), *ident);
///     assert!(vals.contains(&ref_val));
/// }
/// # }
/// ```
pub struct RefHeaders<'a>(HashMap<Ident, RefHeader<'a>>);

impl<'a> RefHeaders<'a> {
    /// Iterates through the `(Ident, RefHeader)` pairs.
    pub fn iter(&self) -> hash_map::Iter<Ident, RefHeader<'a>> {
        self.0.iter()
    }
}

impl<'a> From<&'a Headers> for RefHeaders<'a> {
    fn from(headers: &'a Headers) -> Self {
        let inner = headers.iter()
            .map(|(ident, header)| {
                let converted = header.iter()
                    .map(Option::as_ref)
                    .collect();
                (*ident, converted)
            })
            .collect();
        RefHeaders(inner)
    }
}

impl<'a> IntoIterator for &'a RefHeaders<'a> {
    type Item = (&'a Ident, &'a RefHeader<'a>);
    type IntoIter = hash_map::Iter<'a, Ident, RefHeader<'a>>;
    fn into_iter(self) -> hash_map::Iter<'a, Ident, RefHeader<'a>> {
        self.iter()
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashSet;

    use serde_json::{self, Deserializer};

    use test_help;

    use super::*;

    /// Makes sure the relevant types are registered for the tests.
    fn register_types() {
        register::<Direction>();
        register::<IpProto>();
        register_endpoint::<IpAddr>();
        register_endpoint::<MacAddress>();
        register_endpoint::<Name>();
        register_endpoint::<Names>();
        register_endpoint::<Port>();
    }

    /// Tests serializing and deserializing the identities.
    ///
    /// Also some other manipulation ‒ comparing and registration.
    #[test]
    fn ident() {
        register_types();
        fn t<T: Type>(name: &str) {
            test_help::deser(name, &Ident::of::<T>());
            test_help::ser(name, &Ident::of::<T>());
            // It is stable in the type we give it
            assert_eq!(Ident::of::<T>(), Ident::of::<T>());
        }
        t::<Remote<IpAddr>>("\"remote-ip\"");
        t::<Local<Name>>("\"local-name-primary\"");
        t::<Direction>("\"direction\"");
        // Different types are not equal
        assert!(Ident::of::<Direction>() != Ident::of::<IpProtoRaw>());
        // Some negative answers
        serde_json::from_str::<Ident>("\"nothing\"").unwrap_err();
        serde_json::from_str::<Ident>("\"local-nothing\"").unwrap_err();
        // But we can plug them in by registration
        #[derive(Clone, Debug, Deserialize, Eq, Ord, Serialize, PartialEq, PartialOrd)]
        struct Nothing;
        impl Type for Nothing {
            fn name() -> String { "nothing".to_owned() }
        }
        impl EndpointType for Nothing {
            fn name() -> String { "nothing".to_owned() }
        }
        register::<Nothing>();
        register_endpoint::<Nothing>();
        t::<Nothing>("\"nothing\"");
        t::<Local<Nothing>>("\"local-nothing\"");
        // Multiple registrations of the same type is OK
        register::<Nothing>();
        register_endpoint::<Nothing>();
    }

    /// Tests serializing and deserializing the values.
    ///
    /// And some other manipulation.
    #[test]
    fn value() {
        register_types();
        fn t<T: Type>(inner: T, serialized: &str) {
            let ident = Ident::of::<T>();
            let value = Value::from(inner);
            assert_eq!(ident, value.ident());
            let mut deserializer = Deserializer::from_str(serialized);
            let decoded = ident.decode_value(&mut deserializer).unwrap();
            assert_eq!(value, decoded.unwrap());
            let encoded = serde_json::to_string(&value).unwrap();
            assert_eq!(serialized, encoded);
            let mut deserializer_none = Deserializer::from_str("null");
            let none = ident.decode_value(&mut deserializer_none).unwrap();
            assert_eq!(None, none);
        }
        let addr = "192.0.2.1".parse::<IpAddr>().unwrap();
        t(Remote(addr), "\"192.0.2.1\"");
        t(Local(Name::new("example.com".to_owned())), "\"example.com\"");
        t(Direction::In, "\"IN\"");
        t(IpProto::Tcp, "\"TCP\"");
        t(Remote(Port(42)), "42");
    }

    /// Tests downcasting the values.
    #[test]
    fn value_downcast() {
        let value = Value::from(Direction::In);
        assert_eq!(Some(&Direction::In), value.downcast_ref());
        assert!(value.downcast_ref::<IpProto>().is_none());
    }

    /// Tests comparing the values (even across different types).
    #[test]
    fn value_cmp() {
        assert!(Value::from(Local(Port(0))) < Value::from(Local(Port(42))));
        assert!(Value::from(Remote(Port(0))) != Value::from(Local(Port(0))));
    }

    /// Tests manipulation of the tags.
    #[test]
    fn tags() {
        let mut tags = Tags::new();
        assert_eq!(0, tags.len());
        assert!(tags.is_empty());
        let ip = "192.0.2.1".parse::<IpAddr>().unwrap();
        assert!(tags.insert(Local(ip)).is_none());
        assert!(tags.insert(Value::from(Remote(ip))).is_none());
        assert!(tags.insert(Direction::In).is_none());
        assert!(!tags.is_empty());
        assert_eq!(3, tags.len());
        assert!(tags.contains(&Ident::of::<Local<IpAddr>>()));
        assert!(!tags.contains(&Ident::of::<Remote<Port>>()));
        assert_eq!(Some(Value::from(Direction::In)), tags.insert(Direction::Out));
        assert_eq!(3, tags.len());
        assert_eq!(&Value::from(Direction::Out), tags.get(&Ident::of::<Direction>()).unwrap());
        assert!(tags.get(&Ident::of::<IpProto>()).is_none());
        let idents = tags.idents().collect::<HashSet<_>>();
        let idents_expected = [
                Ident::of::<Local<IpAddr>>(),
                Ident::of::<Remote<IpAddr>>(),
                Ident::of::<Direction>()
            ];
        let idents_expected = idents_expected
            .iter()
            .collect::<HashSet<_>>();
        assert_eq!(idents_expected, idents);
    }

    /// Tests extending tags.
    #[test]
    fn tags_extend() {
        let mut tags = Tags::new();
        let empty = Tags::new();
        tags.extend(empty);
        assert!(tags.is_empty());
        let mut additional = Tags::new();
        additional.insert(Direction::In);
        additional.insert(IpProto::Tcp);
        tags.extend(additional.clone());
        assert_eq!(additional, tags);
        // Repeating the extention does nothing bad
        tags.extend(additional.clone());
        assert_eq!(additional, tags);
        // Now insert something more
        // It partially overlaps.
        let mut others = Tags::new();
        others.insert(IpProto::Udp);
        others.insert(Remote(Port(22)));
        tags.extend(others);
        let mut expected = Tags::new();
        expected.insert(IpProto::Udp);
        expected.insert(Direction::In);
        expected.insert(Remote(Port(22)));
        assert_eq!(expected, tags);
    }

    /// Tests serializing and deserializing the headers.
    #[test]
    fn headers() {
        register_types();
        let input = r#"{
            "direction": ["IN"],
            "local-name-set": [
                [],
                ["example.com", "example.org"]
            ],
            "remote-mac": [null, "00-11-22-33-44-55"],
            "remote-ip": [null, "192.0.2.1"]
        }"#;
        let mut headers = Headers::new();
        headers.insert_opt(Ident::of::<Direction>(), Some(Direction::In.into()));
        headers.insert(Local(Names::new()));
        let names = [
                "example.com",
                "example.org",
            ]
            .iter()
            .map(|n| Name::new(n.clone()))
            .collect::<Names>();
        headers.insert(Local(names));
        headers.insert_empty(Ident::of::<Remote<MacAddress>>());
        headers.insert(Remote("00:11:22:33:44:55".parse::<MacAddress>().unwrap()));
        headers.insert_empty(Ident::of::<Remote<IpAddr>>());
        headers.insert(Remote("192.0.2.1".parse::<IpAddr>().unwrap()));
        test_help::deser(input, &headers);
        test_help::ser(input, &headers);
        assert_eq!(4, headers.iter().count());
    }


    /// A strange test-case for the deserialization of headers.
    ///
    /// The actual headers may be split across multiple dict values with the same key. They get
    /// accumulated together.
    #[test]
    fn headers_deser_odd() {
        register_types();
        let input = r#"{
            "direction": ["IN"],
            "direction": ["OUT"],
            "direction": [],
            "remote-mac": []
        }"#;
        let mut headers = Headers::new();
        headers.insert(Direction::In);
        headers.insert(Direction::Out);
        test_help::deser(&input, &headers);
    }

    /// Check hashing and comparing of the flow tags
    ///
    /// As it acts a bit strange.
    #[test]
    fn flow_tags() {
        macro_rules! test {
            ($kind: expr) => {
                let mut set = $kind;
                let tags = FlowTags::new(Tags::new());
                set.insert(tags.clone());
                assert_eq!(1, set.len());
                let tags2 = FlowTags::new(Tags::new());
                // Even when they are the same content, these are considered different in the hash
                set.insert(tags2.clone());
                assert_eq!(2, set.len());
                // Modifying the content doesn't change the hash
                tags.borrow_mut().insert(Direction::In);
                assert!(set.contains(&tags));
            }
        }
        test!(HashSet::new());
        test!(BTreeSet::new());
    }
}