//
// Copyright (c) 2019 RepliXio Ltd. All rights reserved.
// Use is subject to license terms.
//

use std::fmt;
use std::iter::once;
use std::iter::FromIterator;
use std::ops::{Add, AddAssign, Range};
use std::vec;

use itertools::Either::{Left, Right};
use libc::{c_int, iovec};
use serde::ser::SerializeSeq;
use serde::{Deserialize, Serialize, Serializer};
use serde_bytes::Bytes;
use slog;

use super::utils::intersection;
use super::Iovec;
use super::SgList;
use super::Zle;

/// High Level wrapper for multiple data representation methods.
/// It has 2 fields:
///
/// 1. data - a vector of heterogeneous data representations.
///
/// 2. observers -
///     An observer is some generic type that implements Clone.
///
///     An observer can be explicitly associated with an sg-data
///     By a call to the method `observe()`
///     Or by the creation of a shallow copy of an observed sg-data,
///     I.e. a call to the method `copy_masked()`.
///
///     An observer is used as a helper to ensure that an sg-data can safely be dropped,
///     Since an Sg-Data is dropped with it's observers, they will not access it.
///
#[derive(Debug, Deserialize)]
pub struct SgData<O>
where
    O: Clone,
{
    data: Vec<SgEntry>,

    #[serde(skip, default = "Vec::default")]
    observers: Vec<O>,
}

/// High Level wrapper for multiple data representation methods.
#[derive(Deserialize, PartialEq)]
enum SgEntry {
    /// Plain Vec<u8> buffer
    Direct(Vec<u8>),
    /// Classic Scatter Gather list as it comes from C (array of `iovec` elements)
    SgList(SgList),
    /// Special case for single `iovec`
    Iovec(Iovec),
    /// Zero Length Encoding (ZLE) - a buffer of zeroes of a specified length
    Zle(Zle),
}

impl SgEntry {
    /// Iterates over all the underlying byte slices of this `SgData`.
    pub fn iter(&self) -> impl Iterator<Item = &[u8]> {
        match self {
            SgEntry::Direct(vec) => Left(Right(once(vec.as_slice()))),
            SgEntry::SgList(sglist) => unsafe { Right(Right(sglist.iter_slices())) },
            SgEntry::Iovec(iov) => Right(Left(once(iov.as_slice()))),
            SgEntry::Zle(zle) => Left(Left(zle.iter())),
        }
    }

    /// Iterates over all the underlying mutable byte slices of this `SgData`.
    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut [u8]> {
        match self {
            SgEntry::Direct(ref mut vec) => Right(Left(once(vec.as_mut_slice()))),
            SgEntry::SgList(sglist) => unsafe { Right(Right(sglist.iter_slices_mut())) },
            SgEntry::Iovec(iov) => Left(once(iov.as_mut_slice())),
            SgEntry::Zle(_) => panic!("Cannot borrow ZLE slice as mut."),
        }
    }

    /// Returns number of elements in the underlying object, NOT the total number of bytes.
    fn size(&self) -> usize {
        match self {
            SgEntry::SgList(sglist) => sglist.size(),
            SgEntry::Zle(zle) => zle.size(),
            SgEntry::Iovec(_) | SgEntry::Direct(_) => 1,
        }
    }

    /// Checks whether this instance is built from `Iovec`
    fn is_iovec(&self) -> bool {
        match self {
            SgEntry::Iovec(_) => true,
            _ => false,
        }
    }

    /// Checks whether this instance is built from `SgList`
    fn is_sglist(&self) -> bool {
        match self {
            SgEntry::SgList(_) => true,
            _ => false,
        }
    }
}

impl fmt::Debug for SgEntry {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SgEntry::SgList(sglist) => write!(f, "SgEntry::{:?}", sglist),
            SgEntry::Direct(vec) => write!(f, "SgEntry::Vec<u8>[{}])", vec.len()),
            SgEntry::Iovec(iov) => write!(f, "SgEntry::{:?}", iov),
            SgEntry::Zle(zle) => write!(f, "SgEntry::{:?}", zle),
        }
    }
}

impl From<SgList> for SgEntry {
    fn from(sglist: SgList) -> Self {
        SgEntry::SgList(sglist)
    }
}

impl From<Vec<u8>> for SgEntry {
    fn from(vec: Vec<u8>) -> Self {
        SgEntry::Direct(vec)
    }
}

impl From<Iovec> for SgEntry {
    fn from(iov: Iovec) -> Self {
        SgEntry::Iovec(iov)
    }
}

impl From<Zle> for SgEntry {
    fn from(zle: Zle) -> Self {
        SgEntry::Zle(zle)
    }
}

impl FromIterator<u8> for SgEntry {
    fn from_iter<T>(iter: T) -> Self
    where
        T: IntoIterator<Item = u8>,
    {
        let vec = iter.into_iter().collect::<Vec<_>>();
        SgEntry::Direct(vec)
    }
}

impl Serialize for SgEntry {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            SgEntry::SgList(sglist) => {
                // While serializing `SgList` variant mimics `Direct`.
                serializer.serialize_newtype_variant("SgEntry", 1, "Direct", sglist)
            }
            SgEntry::Direct(buf) => {
                serializer.serialize_newtype_variant("SgEntry", 1, "Direct", buf)
            }
            SgEntry::Iovec(vec) => {
                // While serializing `Iovec` variant mimics `Direct`
                serializer.serialize_newtype_variant("SgEntry", 1, "Direct", vec)
            }
            SgEntry::Zle(zle) => {
                // While serializing `Zle` variant mimics `Direct`
                serializer.serialize_newtype_variant("SgEntry", 1, "Direct", zle)
            }
        }
    }
}

impl slog::Value for SgEntry {
    fn serialize(
        &self,
        _record: &slog::Record<'_>,
        key: slog::Key,
        serializer: &mut dyn slog::Serializer,
    ) -> slog::Result {
        serializer.emit_str(key, "SgEntry<>")
    }
}

impl<O> SgData<O>
where
    O: Clone,
{
    /// Create an `SgData` with specified capacity
    pub fn with_capacity(capcity: usize) -> Self {
        let mut vec = Vec::<u8>::with_capacity(capcity);
        unsafe {
            vec.set_len(capcity);
        };
        vec.into()
    }

    /// Iterates over all the underlying byte slices of this `SgData`.
    pub fn iter(&self) -> impl Iterator<Item = &[u8]> {
        self.data.iter().flat_map(SgEntry::iter)
    }

    /// Iterates over all the mutable underlying byte slices of this `SgData`.
    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut [u8]> {
        self.data.iter_mut().flat_map(SgEntry::iter_mut)
    }

    /// Iterate over `slices` `ranges`
    fn iter_ranges<'a>(&'a self) -> impl Iterator<Item = Range<usize>> + 'a {
        let mut offset = 0;

        self.iter().map(move |slice| {
            let start = offset;
            offset += slice.len();
            start..offset
        })
    }

    /// Given a `mask` iterates over `slices` that are visible after masking
    pub fn iter_masked<'a>(&'a self, mask: &'a Range<usize>) -> impl Iterator<Item = &[u8]> + 'a {
        self.iter()
            .zip(self.iter_ranges())
            .filter_map(move |(slice, range)| {
                intersection(&range, mask).map(|intersect| {
                    let len = intersect.end - intersect.start;
                    let start = intersect.start - range.start;
                    &slice[start..(start + len)]
                })
            })
    }

    /// Get an iterator over `SgData`s bytes - useful for comparing values.
    fn iter_bytes(&self) -> impl Iterator<Item = &u8> {
        self.iter().flatten()
    }

    /// Returns number of elements in the underlying object, NOT the total number of bytes.
    pub fn size(&self) -> usize {
        self.data.iter().fold(0, |sum, entry| sum + entry.size())
    }

    /// Returns the total number of bytes in the underlying object.
    pub fn capacity(&self) -> usize {
        self.iter().fold(0, |sum, slice| sum + slice.len())
    }

    /// Checks this SgData for containing any byte slice
    pub fn is_empty(&self) -> bool {
        self.size() == 0
    }

    #[deprecated(since = "0.2.0", note = "use drain_into() instead")]
    #[allow(clippy::cast_sign_loss)]
    pub unsafe fn clone_into(self, iovec: *mut iovec, count: c_int) -> Self {
        self.drain_into(iovec, count as usize)
    }

    /// Moves the data contained in this `SgData` instance into `iovec`.
    /// Consumes this instance and creates a new one, based on the buffers
    /// described by `iovec` and `count`.
    ///
    /// observers are not drained since the data is moved to a new sg-list
    pub unsafe fn drain_into(self, iovec: *const iovec, count: usize) -> Self {
        let sglist = SgList::new(iovec, count);
        assert_eq!(sglist.capacity(), self.capacity());

        sglist.iter_slices_range_mut().for_each(|(buf, range)| {
            self.iter_masked(&range).fold(buf, |buf_slice, slice| {
                buf_slice[..slice.len()].copy_from_slice(slice);
                &mut buf_slice[slice.len()..]
            });
        });

        sglist.into()
    }

    /// Copies self into a Vector
    /// `mask` is an optional parameter: An exclusive range - i.e - from start to end - 1
    pub fn to_vec(&self) -> Vec<u8> {
        self.iter()
            .fold(Vec::with_capacity(self.capacity()), |mut vec, slice| {
                vec.extend_from_slice(slice);
                vec
            })
    }

    /// Applies the `mask` to the current object and creates a new `SgData` object which describes
    /// only masked region
    /// `observers` are not cloned since this is a deep-copy cloning
    pub fn clone_masked(&self, mask: Range<usize>) -> Self {
        let capacity = self
            .iter_masked(&mask)
            .fold(0, |capacity, slice| capacity + slice.len());

        self.iter_masked(&mask)
            .fold(Vec::with_capacity(capacity), |mut vec, slice| {
                vec.extend_from_slice(slice);
                vec
            })
            .into()
    }

    /// Applies the `mask` to the current object and creates a new `SgData` object which describes
    /// only masked region
    /// `observers` are cloned since this is a shallow-copy clonning,
    /// and we want to make sure the data is not used-after-free.
    pub fn copy_masked(&self, mask: Range<usize>) -> Self {
        Self {
            data: self
                .iter_masked(&mask)
                .map(Iovec::from)
                .map(SgEntry::from)
                .collect(),
            observers: self.observers.clone(),
        }
    }
    /// Checks whether this instance is built from `Iovec`
    pub fn is_iovec(&self) -> bool {
        self.data.iter().all(SgEntry::is_iovec)
    }

    /// Checks whether this instance is built from `SgList`
    pub fn is_sglist(&self) -> bool {
        self.data.len() == 1 && self.data[0].is_sglist()
    }

    #[deprecated(since = "0.2.2", note = "use Extend::extend() instead")]
    pub fn append(&mut self, other: &mut Self) {
        self.extend(other.iter());
    }

    /// Creates an efficient (ZLE) zeroed buffer
    pub fn zero(capacity: usize) -> Self {
        Self {
            data: vec![SgEntry::from(Zle::new(capacity))],
            observers: Vec::default(),
        }
    }

    /// let `observer` know when the data associated with this sg-data is dropped.
    pub fn observe(&mut self, observer: &O) {
        self.observers.push(observer.clone())
    }
}

impl<O> From<SgEntry> for SgData<O>
where
    O: Clone,
{
    fn from(entry: SgEntry) -> Self {
        Self {
            data: vec![entry],
            observers: Vec::default(),
        }
    }
}

impl<O> From<SgList> for SgData<O>
where
    O: Clone,
{
    fn from(sglist: SgList) -> Self {
        Self {
            data: vec![SgEntry::from(sglist)],
            observers: Vec::default(),
        }
    }
}

impl<O> From<Vec<u8>> for SgData<O>
where
    O: Clone,
{
    fn from(vec: Vec<u8>) -> Self {
        Self {
            data: vec![SgEntry::from(vec)],
            observers: Vec::default(),
        }
    }
}

impl<O> From<Vec<Vec<u8>>> for SgData<O>
where
    O: Clone,
{
    fn from(sgvec: Vec<Vec<u8>>) -> Self {
        Self {
            data: sgvec.into_iter().map(SgEntry::from).collect(),
            observers: Vec::default(),
        }
    }
}

impl<O> From<Iovec> for SgData<O>
where
    O: Clone,
{
    fn from(elt: Iovec) -> Self {
        Self {
            data: vec![SgEntry::from(elt)],
            observers: Vec::default(),
        }
    }
}

impl<O> From<Vec<Iovec>> for SgData<O>
where
    O: Clone,
{
    fn from(vec: Vec<Iovec>) -> Self {
        Self {
            data: vec.into_iter().map(SgEntry::from).collect(),
            observers: Vec::default(),
        }
    }
}

impl<O> From<Zle> for SgData<O>
where
    O: Clone,
{
    fn from(zle: Zle) -> Self {
        SgEntry::from(zle).into()
    }
}

impl<O> From<SgData<O>> for Vec<u8>
where
    O: Clone,
{
    fn from(sgdata: SgData<O>) -> Self {
        sgdata.to_vec()
    }
}

impl<O> FromIterator<u8> for SgData<O>
where
    O: Clone,
{
    fn from_iter<T>(iter: T) -> Self
    where
        T: IntoIterator<Item = u8>,
    {
        Self {
            data: vec![SgEntry::from_iter(iter)],
            observers: Vec::default(),
        }
    }
}

impl<O> FromIterator<Vec<u8>> for SgData<O>
where
    O: Clone,
{
    fn from_iter<T>(iter: T) -> Self
    where
        T: IntoIterator<Item = Vec<u8>>,
    {
        Self {
            data: iter.into_iter().map(SgEntry::from).collect(),
            observers: Vec::default(),
        }
    }
}

impl<O> FromIterator<Iovec> for SgData<O>
where
    O: Clone,
{
    fn from_iter<T>(iter: T) -> Self
    where
        T: IntoIterator<Item = Iovec>,
    {
        Self {
            data: iter.into_iter().map(SgEntry::from).collect(),
            observers: Vec::default(),
        }
    }
}

impl<O> FromIterator<SgData<O>> for SgData<O>
where
    O: Clone,
{
    fn from_iter<T>(iter: T) -> Self
    where
        T: IntoIterator<Item = Self>,
    {
        let mut iter = iter.into_iter();
        let mut data = iter.nth(0).unwrap_or_else(Self::default);
        data.extend(iter);
        data
    }
}

impl<O> Extend<u8> for SgData<O>
where
    O: Clone,
{
    fn extend<T>(&mut self, iter: T)
    where
        T: IntoIterator<Item = u8>,
    {
        self.data.push(SgEntry::from_iter(iter));
    }
}

impl<'a, O> Extend<&'a [u8]> for SgData<O>
where
    O: Clone,
{
    fn extend<T>(&mut self, iter: T)
    where
        T: IntoIterator<Item = &'a [u8]>,
    {
        self.data
            .extend(iter.into_iter().map(Vec::from).map(SgEntry::from));
    }
}

impl<O> Extend<Iovec> for SgData<O>
where
    O: Clone,
{
    fn extend<T>(&mut self, iter: T)
    where
        T: IntoIterator<Item = Iovec>,
    {
        self.data.extend(iter.into_iter().map(SgEntry::from));
    }
}

impl<'a, O> Extend<&'a Iovec> for SgData<O>
where
    O: Clone,
{
    fn extend<T>(&mut self, iter: T)
    where
        T: IntoIterator<Item = &'a Iovec>,
    {
        self.data
            .extend(iter.into_iter().cloned().map(SgEntry::from));
    }
}

impl<O> Extend<SgData<O>> for SgData<O>
where
    O: Clone,
{
    fn extend<T>(&mut self, iter: T)
    where
        T: IntoIterator<Item = Self>,
    {
        iter.into_iter().for_each(|item| {
            self.data.extend(item.data.into_iter());
            self.observers.extend(item.observers.into_iter());
        });
    }
}

impl<O> Add for SgData<O>
where
    O: Clone,
{
    type Output = Self;

    fn add(mut self, rhs: Self) -> Self::Output {
        self += rhs;
        self
    }
}

impl<O> AddAssign for SgData<O>
where
    O: Clone,
{
    fn add_assign(&mut self, rhs: Self) {
        self.extend(once(rhs));
    }
}

// `SgEntryBytes` struct exists to mimic `SgEntry` during serialization.
// It uses `serde_bytes` to serialize byte slices efficiently and produces only one
// specific `SgEntry::Direct` variant.
struct SgEntryBytes<'a>(Bytes<'a>);

impl<'a> SgEntryBytes<'a> {
    fn new(bytes: &'a [u8]) -> Self {
        Self(Bytes::new(bytes))
    }
}

impl<'a> Serialize for SgEntryBytes<'a> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_newtype_variant("SgEntry", 0, "Direct", &self.0)
    }
}

impl<O> Serialize for SgData<O>
where
    O: Clone,
{
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        // During serialization `SgData` is being "flattened" to a potentially larger number
        // of elements in `self.data` vector. The reason is that both SgList and Zle variants
        // can map into multiple slices, rather than one single slice.
        // When deserializing this will be re-constructed as multiple `SgEntry::Direct` elements,
        // one per underlying slice.

        let length = self
            .data
            .iter()
            .fold(0, |length, entry| length + entry.size());

        let mut seq = serializer.serialize_seq(Some(length))?;

        for entry in &self.data {
            match entry {
                SgEntry::Direct(buf) => {
                    let entry = SgEntryBytes::new(buf);
                    seq.serialize_element(&entry)?;
                }
                SgEntry::SgList(sglist) => {
                    for slice in unsafe { sglist.iter_slices() } {
                        let entry = SgEntryBytes::new(slice);
                        seq.serialize_element(&entry)?;
                    }
                }
                SgEntry::Iovec(iovec) => {
                    let entry = SgEntryBytes::new(iovec);
                    seq.serialize_element(&entry)?;
                }
                SgEntry::Zle(zle) => {
                    for slice in zle.iter() {
                        let entry = SgEntryBytes::new(slice);
                        seq.serialize_element(&entry)?;
                    }
                }
            }
        }

        seq.end()
    }
}

impl<O> slog::Value for SgData<O>
where
    O: Clone,
{
    fn serialize(
        &self,
        _record: &slog::Record<'_>,
        key: slog::Key,
        serializer: &mut dyn slog::Serializer,
    ) -> slog::Result {
        serializer.emit_str(key, "SgData<>")
    }
}

impl<O> PartialEq for SgData<O>
where
    O: Clone,
{
    /// This method tests for `self` and `other` values to be equal, and is used
    /// by `==`.
    fn eq(&self, other: &Self) -> bool {
        self.capacity() == other.capacity()
            && self
                .iter_bytes()
                .zip(other.iter_bytes())
                .all(|(self_byte, other_byte)| self_byte == other_byte)
    }
}

impl<O> Default for SgData<O>
where
    O: Clone,
{
    fn default() -> Self {
        Self {
            data: Vec::default(),
            observers: Vec::default(),
        }
    }
}

impl<O> Clone for SgData<O>
where
    O: Clone,
{
    fn clone(&self) -> Self {
        self.to_vec().into()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use bincode::{deserialize, serialize};
    use libc::{c_void, iovec};
    use std::iter::{once, repeat};
    use std::mem;

    #[derive(Clone, Debug, Default)]
    struct NoneObserver {}
    impl Drop for NoneObserver {
        fn drop(&mut self) {}
    }

    fn _assert_impls() {
        fn assert_send<T: Send>() {}
        fn assert_sync<T: Sync>() {}
        fn assert_clone<T: Clone>() {}
        fn assert_copy<T: Copy>() {}

        assert_send::<SgData<NoneObserver>>();
        assert_sync::<SgData<NoneObserver>>();
        assert_send::<SgEntry>();
        assert_sync::<SgEntry>();
    }

    // NB - never use this code outside of the tests - it leaks memory
    fn vec_into_iovec(mut vec: Vec<u8>) -> iovec {
        let len = vec.len();
        let base = vec.as_mut_ptr();
        mem::forget(vec);
        iovec {
            iov_base: base as *mut c_void,
            iov_len: len,
        }
    }

    // NB - never use this code outside of the tests - it leaks memory
    fn create_sglist(sgvec: Vec<Vec<u8>>) -> SgList {
        let vec = sgvec.into_iter().map(vec_into_iovec).collect::<Vec<_>>();
        let len = vec.len();
        let iov = vec.as_ptr();
        mem::forget(vec);
        SgList::new(iov, len)
    }

    //Drain data into sgvec - add compare values
    unsafe fn drain_data_compare(data_a: &SgData<NoneObserver>, sgvec: Vec<Vec<u8>>) {
        //Drain data into sgvec - add compare values
        let (base, count) = create_sglist(sgvec).into_inner();

        let drained_data = data_a.clone().drain_into(base as *mut libc::iovec, count);

        assert_eq!(data_a, &drained_data);
    }

    // asserts value-equality of to SgData instances.
    fn eq_data<T>(data_a: &SgData<NoneObserver>, data_b: &SgData<NoneObserver>, mask: T)
    where
        T: Into<Option<Range<usize>>>,
    {
        let mask = if let Some(mask) = mask.into() {
            mask
        } else {
            0..usize::max_value()
        };

        let copy_a = data_a.copy_masked(mask.clone());
        let copy_b = data_a.copy_masked(mask.clone());
        assert_eq!(copy_a, copy_b);

        let ser_a = serialize(&copy_a).unwrap();
        let deser_a: SgData<NoneObserver> = deserialize(ser_a.clone().as_slice()).unwrap();
        let ser_b = serialize(&copy_b).unwrap();
        let deser_b: SgData<NoneObserver> = deserialize(ser_b.clone().as_slice()).unwrap();
        assert_eq!(deser_a, deser_b);

        // test converting to a vector deep-copy - i.e cloning.
        assert_eq!(
            data_a.clone_masked(mask.clone()),
            data_b.clone_masked(mask.clone())
        );

        unsafe {
            // test deep-copy - i.e cloning.
            let ser_a = serialize(&data_a.clone_masked(mask.clone())).unwrap();
            let deser_a: SgData<NoneObserver> = deserialize(ser_a.clone().as_slice()).unwrap();
            let ser_b = serialize(&data_b.clone_masked(mask.clone())).unwrap();
            let deser_b: SgData<NoneObserver> = deserialize(ser_b.clone().as_slice()).unwrap();
            assert_eq!(deser_a, deser_b);

            drain_data_compare(
                &data_a,
                partition_vec(vec![42; data_a.capacity()].as_slice()),
            );

            // create a vector of vector where one is of size 1
            drain_data_compare(&data_a, vec![vec![42; data_a.capacity() - 1], vec![44; 1]]);

            // create a vector of vector where one is of size 1
            drain_data_compare(&data_a, vec![vec![44; 1], vec![42; data_a.capacity() - 1]]);

            //Drain data into masked sgvec - add compare values
            drain_data_compare(
                &data_a.clone_masked(mask.clone()),
                partition_vec(vec![43; data_a.copy_masked(mask.clone()).capacity()].as_slice()),
            );

            let data =
                SgData::from_iter(data_b.copy_masked(mask.clone()).iter_mut().map(Iovec::from));
            assert_eq!(data, data_a.clone_masked(mask.clone()));
        }
    }

    // given two value-equal mutable instances of SgData.
    // Test that they are (not)-eqaul after mutation.
    fn compare_data(data_a: &mut SgData<NoneObserver>, data_b: &mut SgData<NoneObserver>) {
        // Generate two sg-datas of different types with equal byte-wise values.
        eq_data(&data_a, &data_b, None);

        let capacity = data_a.capacity();

        // Define an arbitray data representation for sg-data
        let vec_def = gen_def_data(25, 25);

        let assign_a = build_data(vec_def.as_slice());
        let assign_b = build_data(vec_def.as_slice());

        data_a.add_assign(assign_a);

        assert_ne!(data_a, data_b);

        eq_data(
            &data_a,
            &data_b,
            Range {
                start: 0,
                end: capacity,
            },
        );

        data_b.add_assign(assign_b);

        assert_eq!(data_a, data_b);

        let mut vec = data_a.to_vec();
        if !vec.is_empty() {
            vec[0] = if u8::max_value() == vec[0] {
                0
            } else {
                vec[0] + 1
            };

            let data_c = &SgData::from(vec);
            assert_ne!(data_a, data_c);

            eq_data(
                &data_a,
                &data_c,
                Range {
                    start: 1,
                    end: data_a.capacity(),
                },
            );

            assert_ne!(data_b, data_c);
            eq_data(
                &data_b,
                &data_c,
                Range {
                    start: 1,
                    end: data_b.capacity(),
                },
            );
        }

        let data_a: &SgData<NoneObserver> = data_a;
        let sgvec = vec![vec![0_u8; data_a.capacity()]];
        let (base, count) = create_sglist(sgvec).into_inner();

        let drained_data = unsafe { data_a.clone().drain_into(base as *mut libc::iovec, count) };
        assert_eq!(data_a, &drained_data);
    }

    // Given a vector defining a Sg-data - pairs of values X size, create a random SgData instance.
    fn build_data(data_def: &[(u8)]) -> SgData<NoneObserver> {
        use rand::{thread_rng, Rng};
        let mut rng = thread_rng();
        SgData::from_iter(partition_vec(data_def).into_iter().map(|vec| {
            if vec.iter().cloned().all(|val| val == 0) && rng.gen_bool(0.8) {
                SgData::zero(vec.len())
            } else {
                match rng.gen_range(0, 3) {
                    0 => SgData::from(Iovec::from(vec_into_iovec(vec))),
                    1 => SgData::from(vec),
                    _ => SgData::from(create_sglist(partition_vec(vec.as_slice()))),
                }
            }
        }))
    }

    fn gen_def_data(num_of_entries: usize, max_entry_capacity: usize) -> Vec<(u8)> {
        use rand::{distributions::Uniform, random, thread_rng, Rng};
        (0..num_of_entries).fold(Vec::new(), |mut vec, _| {
            let value = if thread_rng().gen_bool(0.2) {
                0
            } else {
                random::<u8>()
            };
            let size = thread_rng().sample::<usize, _>(Uniform::new(1, max_entry_capacity + 1));

            vec.extend(vec![value; size].into_iter());
            vec
        })
    }

    fn partition_vec(vec: &[u8]) -> Vec<Vec<u8>> {
        use rand::{thread_rng, Rng};
        let indices: Vec<usize> = vec
            .iter()
            .enumerate()
            .map(|(index, _)| index)
            .filter(|index| {
                if *index == 0 {
                    true
                } else {
                    thread_rng().gen_bool(0.02)
                }
            })
            .collect();

        indices
            .iter()
            .cloned()
            .zip(indices.iter().cloned().skip(1).chain(once(vec.len())))
            .fold(Vec::new(), |mut gather, (start, end)| {
                gather.push(vec[start..end].to_vec());
                gather
            })
    }

    #[test]
    fn compare_gen() {
        let num_of_entries_iter = [25, 250, 2500].to_vec().into_iter();
        let max_entry_capacity_iter = num_of_entries_iter.clone().rev();

        for (num_of_entries, max_entry_capacity) in num_of_entries_iter.zip(max_entry_capacity_iter)
        {
            let data_def = gen_def_data(num_of_entries, max_entry_capacity);
            println!("data def {:?}", data_def);

            let mut data_a = build_data(data_def.as_slice());
            println!("data a {:?}", data_a);

            let mut data_b = build_data(data_def.as_slice());
            println!("data b {:?}", data_b);

            compare_data(&mut data_a, &mut data_b);
        }
    }

    #[test]
    fn compare_types_zeros() {
        // Define an arbitray data representation for sg-data
        let vec_def: Vec<(u8, usize)> = vec![(0, 44), (0, 511), (0, 44)];

        // define the data0 to be a vector of vectors
        let mut data_a = SgData::from(vec_def.iter().fold(Vec::new(), |mut vec, (val, size)| {
            vec.push(vec![*val; *size]);
            vec
        }));

        let original_capacity = data_a.capacity();

        let vec = vec_def
            .iter()
            .fold(Vec::new(), |mut vec: Vec<u8>, (val, size)| {
                vec.extend(vec![*val; *size].iter());
                vec
            });

        let iovec = Iovec::from(iovec {
            iov_base: vec.as_ptr() as *mut c_void,
            iov_len: vec.len(),
        });

        let mut data_b = SgData::from(iovec);

        compare_data(&mut data_a, &mut data_b);

        let data_zle = SgData::from(Zle::new(data_a.capacity()));

        compare_data(
            &mut data_a.clone_masked(0..original_capacity),
            &mut data_zle.clone_masked(0..original_capacity),
        );

        compare_data(
            &mut data_b.clone_masked(0..original_capacity),
            &mut data_zle.clone_masked(0..original_capacity),
        )
    }

    #[test]
    fn combine() {
        let vec_def: Vec<(u8, usize)> = vec![(75, 44), (144, 80), (77, 44)];

        // define the data0 to be a vector of vectors
        let data0: SgData<NoneObserver> =
            SgData::from(vec_def.iter().fold(Vec::new(), |mut vec, (val, size)| {
                vec.push(vec![*val; *size]);
                vec
            }));

        let mut data_combined = data0.copy_masked(0..usize::max_value());

        let vec1: Vec<u8> = vec_def.iter().fold(Vec::new(), |mut vec, (val, size)| {
            vec.extend(vec![*val; *size].iter());
            vec
        });

        // define the data1 to be a direct vector
        let data1: SgData<NoneObserver> = SgData::from(vec1.clone());

        let data1_elements = data1.copy_masked(0..usize::max_value());
        data_combined.extend(data1_elements.iter());

        let mut vec_copied = data0.to_vec();
        vec_copied.extend(data1.to_vec());
        assert_eq!(data_combined.to_vec(), vec_copied);
    }

    #[test]
    #[allow(clippy::cast_possible_truncation)]
    fn mut_with_capacity() {
        let capacity = 13;
        let num_of_vectors = 17;

        // generated sg-data made out of sg-data's - created with capacity
        let mut data_a = SgData::from_iter(
            repeat(capacity)
                .take(num_of_vectors)
                .map(SgData::<NoneObserver>::with_capacity),
        );

        // edit each of the underlinig elemets - and make sure that vec.len() == data.capacity()
        data_a.iter_mut().enumerate().for_each(|(index, slice)| {
            assert_eq!(slice.len(), capacity);
            slice.iter_mut().for_each(|byte| *byte = index as u8)
        });

        // generate the same data with u8 iterator
        let data_b = SgData::<NoneObserver>::from_iter(
            repeat(capacity)
                .take(num_of_vectors)
                .enumerate()
                .flat_map(|(index, capacity)| repeat(index as u8).take(capacity)),
        );

        // compare expected results.
        assert_eq!(data_b, data_a)
    }

    use test::Bencher;

    #[bench]
    fn zle_serialize(b: &mut Bencher) {
        let data: SgData<NoneObserver> = SgData::zero(8192);

        b.iter(|| serialize(&data).unwrap());
    }

    #[bench]
    fn zle_deserialize(b: &mut Bencher) {
        let data: SgData<NoneObserver> = SgData::zero(8192);
        let buf = serialize(&data).unwrap();

        b.iter(|| deserialize::<SgData<NoneObserver>>(&buf).unwrap());
    }

    #[bench]
    fn direct_serialize(b: &mut Bencher) {
        let data: SgData<NoneObserver> = SgData::from(vec![0x45_u8; 8192]);

        b.iter(|| serialize(&data).unwrap())
    }

    #[bench]
    fn direct_deserialize(b: &mut Bencher) {
        let data: SgData<NoneObserver> = SgData::from(vec![0x45_u8; 8192]);
        let buf = serialize(&data).unwrap();

        b.iter(|| deserialize::<SgData<NoneObserver>>(&buf).unwrap());
    }

    #[bench]
    fn sglist_serialize(b: &mut Bencher) {
        let sgvec = vec![vec![0x45_u8; 4096]; 2];
        let data: SgData<NoneObserver> = SgData::from(create_sglist(sgvec));

        b.iter(|| serialize(&data).unwrap())
    }

    #[bench]
    fn sglist_deserialize(b: &mut Bencher) {
        let sgvec = vec![vec![0x45_u8; 4096]; 2];
        let data: SgData<NoneObserver> = SgData::from(create_sglist(sgvec));
        let buf = serialize(&data).unwrap();

        b.iter(|| deserialize::<SgData<NoneObserver>>(&buf).unwrap());
    }

    #[bench]
    fn iovec_serialize(b: &mut Bencher) {
        let data: SgData<NoneObserver> = vec![vec![0x46; 4096]; 2]
            .into_iter()
            .map(vec_into_iovec)
            .map(Iovec::from)
            .collect();

        b.iter(|| serialize(&data).unwrap())
    }

    #[bench]
    fn iovec_deserialize(b: &mut Bencher) {
        let data: SgData<NoneObserver> = vec![vec![0x46; 4096]; 2]
            .into_iter()
            .map(vec_into_iovec)
            .map(Iovec::from)
            .collect();
        let buf = serialize(&data).unwrap();

        b.iter(|| deserialize::<SgData<NoneObserver>>(&buf).unwrap());
    }
}