StreamState.h

Go to the documentation of this file.

#ifndef MEDUSA_STREAMSTATE_H
#define MEDUSA_STREAMSTATE_H

#include "common.h"
#include "Objects.h"
#include "StreamBuffer.h"
#include "Stats.h"

BOREALIS_NAMESPACE_BEGIN;

class StreamState {

 private:

    Name m_name;

    ptr<StreamDef> m_stream_def;

 public:

    // Stream classification.
    enum Class {
        INJECT,
        OTHER
    };

    int m_class;
    bool is_injected_stream() { return m_class == INJECT; }

    // List of subscriptions
    vector<Subscription> m_fast_subs; // List of clients subscribed to this stream

    // If not a local stream, have I asked to receive events on this stream?
    bool m_subscribed;
    pair<MedusaID,Subscription> m_sub; // Remember our subscription so we can remove it later
    
    // How many local queries are connected to this stream on either end
    int m_nb_queries;

    StreamBuffer m_buffer; // Buffers tuples on this stream
    //bool m_delete;

    // For stats purposes
    double m_tuples;
    double m_bytes;
 
    StreamState(Name stream_name, ptr<StreamDef> stream_def, Class clazz) :
        m_name(stream_name), m_stream_def(stream_def), m_class(clazz), 
        m_subscribed(false), m_nb_queries(0),
        m_buffer(stream_name),
        //m_delete(false), 
        m_tuples(0.0), m_bytes(0.0) {
    }
    
    string as_string() const {
        string out = string("StreamState{") + m_name;
        if (m_stream_def)
            out << "; def=" << *(m_stream_def);

        out << "; class=";
        switch (m_class) {
        case INJECT: out << "INJECT"; break;
        default: out << "OTHER";
        };

        if (m_subscribed) {
            out << "; subscribed";
            out << " to " << m_sub.first;
        }
        out << "; nb_queries is " << m_nb_queries;
        out << "; buffer is " << m_buffer;
        out << "}";
        
        return out;
    }

    string repr() const { return as_string(); }

    ptr<StreamDef> get_stream_def() const { return m_stream_def; }
    void set_definition(ptr<StreamDef> str) { m_stream_def = str; }

    const Name& get_name() const { return m_name; }
    Schema get_schema() const { return m_stream_def->get_schema(); }

    // This will go away once we have a good StatsManager!!!
    // Rates of tuples and bytes on this stream
    struct stream_stats {
        double m_tuple_rate;
        double m_byte_rate;
        stream_stats() : m_tuple_rate(0.0), m_byte_rate(0.0) {}
    };
    stream_stats m_stream_stats;
    void incrStats(double tuples, double bytes) { 
        m_tuples += tuples;
        m_bytes += bytes;
    }
    void endStatPeriod(double delta_time_msecs) {
        m_stream_stats.m_tuple_rate 
            = Stats::average(m_stream_stats.m_tuple_rate,MSECS*m_tuples/delta_time_msecs);
        m_stream_stats.m_byte_rate
            = Stats::average(m_stream_stats.m_byte_rate,MSECS*m_bytes/delta_time_msecs);
        m_tuples = m_bytes = 0.0;
    }
    stream_stats getStreamStats() {
        return m_stream_stats;
    }
    
    double n_tuples() {return m_tuples;};


};

BOREALIS_NAMESPACE_END


#endif

---