src/stages/transaction.hh

/* -*- mode: c++; c-basic-offset: 4; -*- */
#ifndef SENDS_TRANSACTION_HH
#define SENDS_TRANSACTION_HH

//#include "tbb/concurrent_queue.h"
#include "mt_queue.hh"
#include "socket_api.hh"
#include "user_data.hh"
#include "chan_list.hh"
#include "segdb.hh"
#include <vector>
#include <iosfwd>

namespace sends {

    class transaction;
    class buffer;
    //typedef tbb::concurrent_queue<transaction*> tran_queue;
    typedef thread::mt_queue<transaction*> tran_queue;

    class transaction {
    public:

        typedef chan_list::chan_index chan_index;

        typedef unsigned long gps_type;

        class chan_stat {
            chan_index _chinx;
            gps_type   _fillpt;
        public:
            chan_stat(chan_index chn);

            chan_stat(void);

            gps_type fill(void) const;

            chan_index index(void) const;

            bool ready(gps_type stop) const;

            void set_fill(gps_type t);
        };

        typedef std::vector<chan_stat> cinx_vect;

        typedef seg_db::seg_index seg_index;

        typedef std::vector<seg_index> seg_vect;

        enum command_code {
            kNoop,
            kGetStatus,
            kGetData,
            kGetServerStatus,
            kGetChannels,
            kQuit,
            kCommandCount
        };

    public:
        transaction(tran_queue& retQ);

        ~transaction(void);

        long block_seq(void);

        void add_channel(chan_index inx);

        void clear(void);

        void clear_channels(void);

        command_code command(void) const;

        std::ostream& dump(std::ostream& out) const;

        gps_type getChanFill(int i) const;

        chan_index getChanIndex(int i) const;

        bool getChanReady(int i) const;

        static command_code getCommandCode(const std::string& str);

        static const char*  getCommandString(command_code code);

        int getNChan(void) const;

        long get_stage(void) const;

        gps_type getStartGPS(void) const;

        gps_type getStopGPS(void) const;

        gps_type getStrideEnd(void) const;

        long ident(void) const;

        bool next_stride(void);

        int put_response(int resp);

        socket_api& ref_socket(void);

        user_data& ref_user(void);

        int  read(buffer& buf);

        void error_reply(const std::string& err);

        void reply(const buffer& reply);

        void release(void);

        void release_segments(void);

        void release_segments(gps_type before);

        void reserve_segments(void);

        void setCommand(command_code cmd);

        void setCommand(const std::string& cmd);

        void set_id(long tranid);

        void set_socket(socket_api* sock);

        void set_stage(long s);

        void set_user(const user_data& user);

        void setTime(gps_type start, gps_type stop, gps_type stride=0);

        gps_type stride(void) const;
    private:
        long trans_id;
        tran_queue* home;
        socket_api* trans_sock;
        user_data trans_user;
        command_code trans_code;
        gps_type trans_start;
        gps_type trans_stop;
        gps_type trans_stride;
        cinx_vect trans_channels;
        seg_vect trans_segments;
        long trans_blockid;
        long stage;
    };


    //==================================  transaction::chan_stat inline methods
    inline 
    transaction::chan_stat::chan_stat(chan_index chn)
        : _chinx(chn), _fillpt(0)
    {}

    inline 
    transaction::chan_stat::chan_stat(void) {
    }

    inline transaction::gps_type
    transaction::chan_stat::fill(void) const {
        return _fillpt;
    }

    inline transaction::chan_index 
    transaction::chan_stat::index(void) const {
        return _chinx;
    }

    inline bool 
    transaction::chan_stat::ready(gps_type t) const {
        return _fillpt >= t;
    }

    inline void 
    transaction::chan_stat::set_fill(gps_type t) {
        _fillpt = t;
    }

    //==================================  transction inline methods
    inline long 
    transaction::block_seq(void) {
        return trans_blockid++;
    }

    inline transaction::command_code 
    transaction::command(void) const {
        return trans_code;
    }

    inline transaction::gps_type
    transaction::getChanFill(int i) const {
        gps_type t = trans_channels[i].fill();
        if (t < trans_start) t = trans_start;
        return t;
    }

    inline transaction::chan_index
    transaction::getChanIndex(int i) const {
        return trans_channels[i].index();
    }

    inline bool
    transaction::getChanReady(int i) const {
        return trans_channels[i].ready(trans_start + trans_stride);
    }

    inline int
    transaction::getNChan(void) const {
        return trans_channels.size();
    }

    inline long
    transaction::get_stage(void) const {
        return stage;
    }

    inline transaction::gps_type 
    transaction::getStartGPS(void) const {
        return trans_start;
    }

    inline transaction::gps_type  
    transaction::getStopGPS(void) const {
        return trans_stop;
    }

    inline transaction::gps_type  
    transaction::getStrideEnd(void) const {
        return trans_start + trans_stride;
    }

    inline long
    transaction::ident(void) const {
        return trans_id;
    }

    inline socket_api&
    transaction::ref_socket(void) {
        return *trans_sock;
    }

    inline user_data& 
    transaction::ref_user(void) {
        return trans_user;
    }
    inline transaction::gps_type 
    transaction::stride(void) const {
        return trans_stride;
    }

}

#endif // !defined(SENDS_TRANSACTION_HH)

---