1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
//! Module with logic for arranging items in-order on multiple streams.
//!
//! _"Order is the process of putting something in a particular order."_
//!
//! # How ordering works.
//! Imagine we have this sequence: `1,5,4,2,3` and we want the user to eventually see: `1,2,3,4,5`.
//!
//! Let's define some variables:
//!
//! ## Variable Setup
//! **hashmap**
//!
//! | Key     | Value |
//! | :-------------: | :-------------:    |
//! |       _       |        _         |
//!
//! `expected_index = 1;`
//!
//! ## Ordering
//! **Receive '1'**
//!
//! - Packet 1 is equals to our expected index we can return it immediately.
//! - Increase `expected_index` to '2'
//!
//! **Receive '5'**
//!
//! Packet '5' is not equal to our expected index so we need to store it until we received **all** packets up to 5 before returning.
//!
//! | Key     | Value |
//! | :-------------: | :-------------:    |
//! |       5       |        packet         |
//!
//! **Receive '4'**
//!
//! Packet '4' is not equal to our expected index so we need to store it until we received **all** packets up to 4 before returning.
//!
//! | Key     | Value |
//! | :-------------: | :-------------:    |
//! |       5       |        packet         |
//! |       4       |        packet         |
//!
//! **Receive '3'**
//!
//! Packet '3' is not equal to our expected index so we need to store it until we received **all** packets up to 3 before returning.
//!
//! | Key     | Value |
//! | :-------------: | :-------------:    |
//! |       5       |        packet         |
//! |       4       |        packet         |
//! |       4       |        packet         |
//!
//! **Receive '2'**
//!
//! - Packet 2 is equals to our expected index we can return it immediately.
//! - Increase `expected_index` to '3'
//!
//! Now we received our `expected_index` we can check if we have the next `expected_index` in storage.
//!
//! This could be done with an iterator which returns packets as long there are packets in our storage matching the `expected_index`.
//!
//! ```no-run
//! let stream = OrderingStream::new();
//!
//! let iter = stream.iter_mut();
//!
//! while let Some(packet) = iter.next() {
//!    // packets from iterator are in order.
//! }
//! ```
//!
//! # Remarks
//! - See [super-module](../index.html) description for more details.

use super::{Arranging, ArrangingSystem};
use crate::packet::SequenceNumber;
use std::collections::HashMap;

/// An ordering system that can arrange items in order on different streams.
///
/// Checkout [`OrderingStream`](./struct.OrderingStream.html), or module description for more details.
///
/// # Remarks
/// - See [super-module](../index.html) for more information about streams.
pub struct OrderingSystem<T> {
    // '[HashMap]' with streams on which items can be ordered.
    streams: HashMap<u8, OrderingStream<T>>,
}

impl<T> OrderingSystem<T> {
    /// Constructs a new [`OrderingSystem`](./struct.OrderingSystem.html).
    pub fn new() -> OrderingSystem<T> {
        OrderingSystem {
            streams: HashMap::with_capacity(32),
        }
    }
}

impl<'a, T> ArrangingSystem for OrderingSystem<T> {
    type Stream = OrderingStream<T>;

    /// Returns the number of ordering streams currently active.
    fn stream_count(&self) -> usize {
        self.streams.len()
    }

    /// Try to get an [`OrderingStream`](./struct.OrderingStream.html) by `stream_id`.
    /// When the stream does not exist, it will be inserted by the given `stream_id` and returned.
    fn get_or_create_stream(&mut self, stream_id: u8) -> &mut Self::Stream {
        self.streams
            .entry(stream_id)
            .or_insert_with(|| OrderingStream::new(stream_id))
    }
}

/// A stream on which items will be arranged in-order.
///
/// # Algorithm
///
/// With every ordering operation an `incoming_index` is given. We also keep a local record of the `expected_index`.
///
/// There are three scenarios that are important to us.
/// 1. `incoming_index` == `expected_index`.
/// This package meets the expected order, so we can return it immediately.
/// 2. `incoming_index` > `expected_index`.
/// This package is newer than we expect, so we have to hold it temporarily until we have received all previous packages.
/// 3. `incoming_index`< `expected_index`
/// This can only happen in cases where we have a duplicated package. Again we don't give anything back.
/// # Remarks
/// - See [super-module](../index.html) for more information about streams.
pub struct OrderingStream<T> {
    // the id of this stream.
    _stream_id: u8,
    // the storage for items that are waiting for older items to arrive.
    // the items will be stored by key and value where the key is the incoming index and the value is the item value.
    storage: HashMap<usize, T>,
    // the next expected item index.
    expected_index: usize,
    // unique identifier which should be used for ordering on a different stream e.g. the remote endpoint.
    unique_item_identifier: u16,
}

impl<T> OrderingStream<T> {
    /// Constructs a new, empty [`OrderingStream<T>`](./struct.OrderingStream.html).
    ///
    /// The default stream will have a capacity of 32 items.
    pub fn new(stream_id: u8) -> OrderingStream<T> {
        OrderingStream::with_capacity(1024, stream_id)
    }

    /// Constructs a new, empty [`OrderingStream`] with the specified capacity.
    ///
    /// The stream will be able to hold exactly capacity elements without
    /// reallocating. If capacity is 0, the vector will not allocate.
    ///
    /// It is important to note that although the returned stream has the capacity specified,
    /// the stream will have a zero length.
    ///
    /// [`OrderingStream`]: ./struct.OrderingStream.html
    pub fn with_capacity(size: usize, stream_id: u8) -> OrderingStream<T> {
        OrderingStream {
            storage: HashMap::with_capacity(size),
            expected_index: 1,
            _stream_id: stream_id,
            unique_item_identifier: 0,
        }
    }

    /// Returns the identifier of this stream.
    #[cfg(test)]
    pub fn stream_id(&self) -> u8 {
        self._stream_id
    }

    /// Returns the next expected index.
    #[cfg(test)]
    pub fn expected_index(&self) -> usize {
        self.expected_index
    }

    /// Returns the unique identifier which should be used for ordering on the other stream e.g. the remote endpoint.
    pub fn new_item_identifier(&mut self) -> SequenceNumber {
        self.unique_item_identifier = self.unique_item_identifier.wrapping_add(1);
        self.unique_item_identifier
    }

    /// Returns an iterator of stored items.
    ///
    /// # Algorithm for returning items from an Iterator.
    ///
    /// 1. See if there is an item matching our `expected_index`
    /// 2. If there is return the `Some(item)`
    ///    - Increase the `expected_index`
    ///    - Start at '1'
    /// 3. If there isn't return `None`
    ///
    /// # Example
    ///
    /// ```ignore
    /// let stream = OrderingStream::new();
    ///
    /// let iter = stream.iter_mut();
    ///
    /// while let Some(item) = iter.next() {
    ///    // Items from iterator are in order.
    /// }
    /// ```
    ///
    /// # Remarks
    /// - Iterator mutates the `expected_index`.
    /// - You can't use this iterator for iterating trough all cached values.
    pub fn iter_mut(&mut self) -> IterMut<T> {
        IterMut {
            items: &mut self.storage,
            expected_index: &mut self.expected_index,
        }
    }
}

impl<T> Arranging for OrderingStream<T> {
    type ArrangingItem = T;

    /// Will order the given item based on the ordering algorithm.
    ///
    /// With every ordering operation an `incoming_index` is given. We also keep a local record of the `expected_index`.
    ///
    /// # Algorithm
    ///
    /// There are three scenarios that are important to us.
    /// 1. `incoming_index` == `expected_index`.
    /// This package meets the expected order, so we can return it immediately.
    /// 2. `incoming_index` > `expected_index`.
    /// This package is newer than we expect, so we have to hold it temporarily until we have received all previous packages.
    /// 3. `incoming_index` < `expected_index`
    /// This can only happen in cases where we have a duplicated package. Again we don't give anything back.
    ///
    /// # Remark
    /// - When we receive an item there is a possibility that a gab is filled and one or more items will could be returned.
    ///   You should use the `iter_mut` instead for reading the items in order.
    ///   However the item given to `arrange` will be returned directly when it matches the `expected_index`.
    fn arrange(
        &mut self,
        incoming_offset: usize,
        item: Self::ArrangingItem,
    ) -> Option<Self::ArrangingItem> {
        if incoming_offset == self.expected_index {
            self.expected_index += 1;
            Some(item)
        } else if incoming_offset > self.expected_index {
            self.storage.insert(incoming_offset, item);
            None
        } else {
            // only occurs when we get a duplicated incoming_offset.
            None
        }
    }
}

/// Mutable Iterator for [`OrderingStream<T>`](./struct.OrderingStream.html).
///
/// # Algorithm for returning items from Iterator.
///
/// 1. See if there is an item matching our `expected_index`
/// 2. If there is return the `Some(item)`
///    - Increase the `expected_index`
///    - Start at '1'
/// 3. If there isn't return `None`
///
/// # Remarks
///
/// - Iterator mutates the `expected_index`.
/// - You can't use this iterator for iterating trough all cached values.
pub struct IterMut<'a, T> {
    items: &'a mut HashMap<usize, T>,
    expected_index: &'a mut usize,
}

impl<'a, T> Iterator for IterMut<'a, T> {
    type Item = T;

    /// Returns `Some` when there is an item in our cache matching the `expected_index`.
    /// Returns `None` if there are no times matching our `expected` index.
    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
        match self.items.remove(&self.expected_index) {
            None => None,
            Some(e) => {
                *self.expected_index += 1;
                Some(e)
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{Arranging, ArrangingSystem, OrderingSystem};

    #[derive(Debug, PartialEq, Clone)]
    struct Packet {
        pub sequence: usize,
        pub ordering_stream: u8,
    }

    impl Packet {
        fn new(sequence: usize, ordering_stream: u8) -> Packet {
            Packet {
                sequence,
                ordering_stream,
            }
        }
    }

    #[test]
    fn create_stream() {
        let mut system: OrderingSystem<Packet> = OrderingSystem::new();
        let stream = system.get_or_create_stream(1);

        assert_eq!(stream.expected_index(), 1);
        assert_eq!(stream.stream_id(), 1);
    }

    #[test]
    fn create_existing_stream() {
        let mut system: OrderingSystem<Packet> = OrderingSystem::new();

        system.get_or_create_stream(1);
        let stream = system.get_or_create_stream(1);

        assert_eq!(stream.stream_id(), 1);
    }

    #[test]
    fn can_iterate() {
        let mut system: OrderingSystem<Packet> = OrderingSystem::new();

        system.get_or_create_stream(1);
        let stream = system.get_or_create_stream(1);

        let stub_packet1 = Packet::new(1, 1);
        let stub_packet2 = Packet::new(2, 1);
        let stub_packet3 = Packet::new(3, 1);
        let stub_packet4 = Packet::new(4, 1);
        let stub_packet5 = Packet::new(5, 1);

        {
            assert_eq!(
                stream.arrange(1, stub_packet1.clone()).unwrap(),
                stub_packet1
            );

            stream.arrange(4, stub_packet4.clone()).is_none();
            stream.arrange(5, stub_packet5.clone()).is_none();
            stream.arrange(3, stub_packet3.clone()).is_none();
        }
        {
            let mut iterator = stream.iter_mut();

            // since we are awaiting for packet '2' our iterator should not return yet.
            assert_eq!(iterator.next(), None);
        }
        {
            assert_eq!(
                stream.arrange(2, stub_packet2.clone()).unwrap(),
                stub_packet2
            );
        }
        {
            // since we processed packet 2 by now we should be able to iterate and get back: 3,4,5;
            let mut iterator = stream.iter_mut();

            assert_eq!(iterator.next().unwrap(), stub_packet3);
            assert_eq!(iterator.next().unwrap(), stub_packet4);
            assert_eq!(iterator.next().unwrap(), stub_packet5);
        }
    }

    /// Asserts that the given collection, on the left, should result - after it is ordered - into the given collection, on the right.
    macro_rules! assert_order {
        ( [$( $x:expr ),*] , [$( $y:expr),*] , $stream_id:expr) => {
        {
            // initialize vector of given range on the left.
            let mut before: Vec<usize> = Vec::new();
            $(
                before.push($x);
            )*

            // initialize vector of given range on the right.
            let mut after: Vec<usize> = Vec::new();
            $(
                after.push($y);
            )*

            // generate test packets
            let mut packets = Vec::new();
            for (_, v) in before.iter().enumerate() {
                packets.push(Packet::new(*v, $stream_id));
            }

            // create system to handle the ordering of our packets.
            let mut ordering_system = OrderingSystem::<Packet>::new();

            // get stream '1' to order the packets on.
            let stream = ordering_system.get_or_create_stream(1);

            // order packets
            let mut ordered_packets = Vec::new();

            for packet in packets.into_iter() {
                match stream.arrange(packet.sequence, packet.clone()) {
                    Some(packet) => {
                        ordered_packets.push(packet.sequence);
                         // empty the remaining ordered packets into an vector so that we can check if they are ordered.
                        let mut iter = stream.iter_mut();

                        while let Some(packet) = iter.next() {
                            ordered_packets.push(packet.sequence);
                        }
                    }
                    None => {}
                };
            }

             // assert if the expected range of the given numbers equals to the processed range which is in sequence.
             assert_eq!(after, ordered_packets);
            }
        };
    }

    #[test]
    fn expect_right_order() {
        // we order on stream 1
        assert_order!([1, 3, 5, 4, 2], [1, 2, 3, 4, 5], 1);
        assert_order!([1, 5, 4, 3, 2], [1, 2, 3, 4, 5], 1);
        assert_order!([5, 3, 4, 2, 1], [1, 2, 3, 4, 5], 1);
        assert_order!([4, 3, 2, 1, 5], [1, 2, 3, 4, 5], 1);
        assert_order!([2, 1, 4, 3, 5], [1, 2, 3, 4, 5], 1);
        assert_order!([5, 2, 1, 4, 3], [1, 2, 3, 4, 5], 1);
        assert_order!([3, 2, 4, 1, 5], [1, 2, 3, 4, 5], 1);
        assert_order!([2, 1, 4, 3, 5], [1, 2, 3, 4, 5], 1);
    }

    #[test]
    fn order_on_multiple_streams() {
        // we order on streams [1...8]
        assert_order!([1, 3, 5, 4, 2], [1, 2, 3, 4, 5], 1);
        assert_order!([1, 5, 4, 3, 2], [1, 2, 3, 4, 5], 2);
        assert_order!([5, 3, 4, 2, 1], [1, 2, 3, 4, 5], 3);
        assert_order!([4, 3, 2, 1, 5], [1, 2, 3, 4, 5], 4);
        assert_order!([2, 1, 4, 3, 5], [1, 2, 3, 4, 5], 5);
        assert_order!([5, 2, 1, 4, 3], [1, 2, 3, 4, 5], 6);
        assert_order!([3, 2, 4, 1, 5], [1, 2, 3, 4, 5], 7);
        assert_order!([2, 1, 4, 3, 5], [1, 2, 3, 4, 5], 8);
    }
}