StreamBuffer.h

Go to the documentation of this file.

#ifndef BOREALIS_STREAMBUFFER_H
#define BOREALIS_STREAMBUFFER_H

#include "common.h"
#include "StreamEvent.h"

BOREALIS_NAMESPACE_BEGIN

class StreamBuffer {

 private:

    Name m_name;

    // Circular buffer of tuples produced or received on that stream
    bool m_init; // Buffer has been initialized... default false
    ptr<dynbuf> m_bin_tuples; // Buffer of tuples
    int m_tuple_size; // Size of tuples in this buffer
    char* m_begin; // Points to what is currently the oldest tuples
    char* m_end;  // Points to what is currently the end of the most recent tuple
    char* m_next; // Points to the position for the next buffer_as_next_event.
    bool m_next_equals_begin;
    bool m_empty; // If we buffer but trim fast enough, buffer is empty
    Mutex m_buffer_lock; // Thread safety

    char* find(string tuple);

    ptr<StreamEvent> buffer_as_event(char* begin, char* end);

 public:

    StreamBuffer(Name name);
    ~StreamBuffer();
    string as_string() const;
    string repr() const;

    void init_buffer(int tuple_size);

    void reset_next();

    void buffer(const char* new_tuples, int length, int tuple_size);

    bool is_duplicate(string tuple) { 
        char* location;
        locking(m_buffer_lock) {
            location = find(tuple);
        }
        return location != NULL; 
    }

    char* first_new(char* buffer, int buffer_size); 

    ptr<StreamEvent> all_tuples_since(string first_tuple = string());

    ptr<StreamEvent> all_tuples_before(string last_tuple = string());

    ptr<StreamEvent> next_tuples();

    string last_tuple();

    void trim(string tuple);
   
};


BOREALIS_NAMESPACE_END

#endif

---