include/salticidae/buffer.h


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

#ifndef _SALTICIDAE_BUFFER_H
#define _SALTICIDAE_BUFFER_H

#include <list>

namespace salticidae {

class SegBuffer {
    public:
    struct buffer_entry_t {
        bytearray_t data;
        bytearray_t::iterator offset;
        buffer_entry_t(): offset(data.begin()) {}
        buffer_entry_t(bytearray_t &&_data):
            data(std::move(_data)), offset(data.begin()) {}

        buffer_entry_t(buffer_entry_t &&other) {
            size_t _offset = other.offset - other.data.begin();
            data = std::move(other.data);
            offset = data.begin() + _offset;
        }

        buffer_entry_t &operator=(buffer_entry_t &&other) {
            size_t _offset = other.offset - other.data.begin();
            data = std::move(other.data);
            offset = data.begin() + _offset;
            return *this;
        }

        buffer_entry_t(const buffer_entry_t &other): data(other.data) {
            offset = data.begin() + (other.offset - other.data.begin());
        }

        size_t length() const { return data.end() - offset; }
    };

    private:
    std::list<buffer_entry_t> buffer;
    size_t _size;

    public:
    SegBuffer(): _size(0) {}
    ~SegBuffer() { clear(); }

    void swap(SegBuffer &other) {
        std::swap(buffer, other.buffer);
        std::swap(_size, other._size);
    }

    SegBuffer(const SegBuffer &other):
        buffer(other.buffer), _size(other._size) {}

    SegBuffer(SegBuffer &&other):
        buffer(std::move(other.buffer)), _size(other._size) {
        other._size = 0;
    }

    SegBuffer &operator=(SegBuffer &&other) {
        if (this != &other)
        {
            SegBuffer tmp(std::move(other));
            tmp.swap(*this);
        }
        return *this;
    }
 
    SegBuffer &operator=(const SegBuffer &other) {
        if (this != &other)
        {
            SegBuffer tmp(other);
            tmp.swap(*this);
        }
        return *this;
    }

    void rewind(bytearray_t &&data) {
        _size += data.size();
        buffer.push_front(buffer_entry_t(std::move(data)));
    }
  
    void push(bytearray_t &&data) {
        _size += data.size();
        buffer.push_back(buffer_entry_t(std::move(data)));
    }

    bytearray_t move_pop() {
        auto res = std::move(buffer.front().data);
        buffer.pop_front();
        _size -= res.size();
        return std::move(res);
    }
    
    bytearray_t pop(size_t len) {
        bytearray_t res;
        auto i = buffer.begin();
        while (len && i != buffer.end())
        {
            size_t copy_len = std::min(i->length(), len);
            res.insert(res.end(), i->offset, i->offset + copy_len);
            i->offset += copy_len;
            len -= copy_len;
            if (i->offset == i->data.end())
                i++;
        }
        buffer.erase(buffer.begin(), i);
        _size -= res.size();
        return std::move(res);
    }
    
    size_t size() const { return _size; }
    bool empty() const { return buffer.empty(); }
    
    void clear() {
        buffer.clear();
        _size = 0;
    }
};

struct MPSCWriteBuffer {
    using buffer_entry_t = SegBuffer::buffer_entry_t;
    using queue_t = MPSCQueueEventDriven<buffer_entry_t>;
    queue_t buffer;

    MPSCWriteBuffer() {}

    MPSCWriteBuffer(const SegBuffer &other) = delete;
    MPSCWriteBuffer(SegBuffer &&other) = delete;

    void rewind(bytearray_t &&data) {
        buffer.rewind(buffer_entry_t(std::move(data)));
    }
  
    void push(bytearray_t &&data) {
        buffer_entry_t d(std::move(data));
        // TODO: better bounded buffer impl
        while (!buffer.try_enqueue(d)) {}
    }

    bytearray_t move_pop() {
        buffer_entry_t res;
        buffer.try_dequeue(res);
        return std::move(res.data);
    }
    
    queue_t &get_queue() { return buffer; }
};

}

#endif