merge.hpp

Go to the documentation of this file.

/*
 * Copyright 2008,2009,2010,2011,2012 Stephan Ossowski, Korbinian Schneeberger,
 * Felix Ott, Joerg Hagmann, Alf Scotland, Sebastian Bender
 * 
 * This file is part of SHORE.
 * 
 * SHORE is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * SHORE is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with SHORE.  If not, see <http://www.gnu.org/licenses/>.
 */


#ifndef SHORE_PROCESSING_MERGE_HPP__
#define SHORE_PROCESSING_MERGE_HPP__

#include <set>
#include <vector>

#include "pipeline.hpp"

namespace shore {

template<typename T,typename Cmp>
class merge
{
 public:

        class channel;

 private:

        struct ccmp
        {
                Cmp cmp;

                ccmp(Cmp c):cmp(c) {}

                bool operator()(channel*const c1,channel*const c2)
                {
                        return cmp(c1->current(),c2->current());
                }
        };

        std::vector<sink<channel,T>*> m_channels;
        std::multiset<channel*,ccmp> m_actives;

 public:

        class channel
        {
         private:

                friend class shore::pipeline_core_access;
                friend class merge;
                friend class sink<channel,T>;

                int m_channelnumber;

                merge* m_parent;
                sink<channel,T>* m_sink;

                bool m_inmap;
                typename std::multiset<channel*,ccmp>::iterator m_where;

                const T *m_data;

                channel(merge*const parent)
                :m_channelnumber(parent->m_channels.size()),
                 m_parent(parent),m_sink(0),m_inmap(false),
                 m_where(m_parent->m_actives.begin()),
                 m_data(0)
                {}

                channel(const channel&);

                void set_sink(sink<channel,T>*const s)
                {
                        m_sink=s;
                }

                int channelnumber() const
                {
                        return m_channelnumber;
                }

                const T& current() const
                {
                        return *m_data;
                }

                bool has_data()
                {
                        if(!m_inmap)
                                m_sink->sigthaw().emit();
                        return m_inmap;
                }

                void next()
                {
                        if(m_inmap)
                        {
                                m_parent->m_actives.erase(m_where);
                                m_inmap=false;
                        }
                        m_sink->sigthaw().emit();
                }

                void append(const T& v)
                {
                        if(m_inmap)
                                m_parent->m_actives.erase(m_where);
                        m_data=&v;
                        m_where=m_parent->m_actives.insert(this);
                        m_inmap=true;
                        m_sink->sigfreeze().emit();
                }

                void flush()
                {}
        };

        typedef T current_type;

        merge(Cmp c=Cmp())
        :m_actives(ccmp(c))
        {}

        ~merge()
        {
                while(!m_channels.empty())
                {
                        delete m_channels.back();
                        m_channels.pop_back();
                }
        }

        sink<channel,T>& make_channel()
        {
                m_channels.push_back(new sink<channel,T>(this));
                (*m_channels.back())->set_sink(m_channels.back());
                return *m_channels.back();
        }

        const int current_id() const
        {   
                return (*m_actives.begin())->channelnumber();
        }

        const current_type& current() const
        {   
                return (*m_actives.begin())->current();
        }

        bool has_data() const
        {
                if(m_actives.empty())
                {
                        for(size_t i=0;i<m_channels.size();++i)
                        {
                                (*m_channels[i])->has_data();
                        }
                }
                return !m_actives.empty();
        }

        void next()
        {
                (*m_actives.begin())->next();
        }
};

} //namespace shore

#endif // SHORE_PROCESSING_MERGE_HPP__