filter.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_FILTER_HPP__
#define SHORE_PROCESSING_FILTER_HPP__

#include <stdexcept>
#include <vector>

#include <boost/function.hpp>


#include "pipeline.hpp"
#include "shore/processing/feed.hpp"
#include "shore/processing/buffer_chain.hpp"


namespace shore {

template<typename T,typename U=T>
class thru
:public pipe_facade<thru<T,U>,T,U>
{
 private:

        friend class pipeline_core_access;

        void append(const T &obj)
        {
                this->emit(obj);
        }
};

template<typename T,typename U=T>
struct conversion
{
        T & operator()(T & dst,const U & src) const
        {
                dst=src;
        }
};

template<typename T,typename U=T,typename Converter=conversion<T,U> >
class conversion_pipe
:public pipe_facade<conversion_pipe<T,U>,T,U>,
 public buffer_chain<U>
{
 private:

        friend class pipeline_core_access;

        Converter m_conv;

        void next()
        {
                if(buffer_chain<U>::buffer_chain_pop())
                        emit(buffer_chain<U>::buffer_chain_front());
        }

        void append(const T & obj)
        {
                m_conv(buffer_chain<U>::buffer_chain_push(),obj);

                if(buffer_chain<U>::buffer_chain_prepare())
                        emit(buffer_chain<U>::buffer_chain_front());
        }

        void flush()
        {
                if(buffer_chain<U>::buffer_chain_flush(false))
                        emit(buffer_chain<U>::buffer_chain_front());
        }
};

template<typename T>
class filter
{
 private:

        typedef boost::function<bool (const T&)> functor_t;

        const T* m_current;

        std::vector<functor_t> m_ops;

 public:

        typedef T current_type;
        typedef T append_type;

        filter():m_current(0) {}

        void add_op(functor_t func)
        {
                m_ops.push_back(func);
        }

        bool has_data() const
        {
                return m_current!=0;
        }

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

        void next()
        {
                m_current=0;
        }

        void append(const T& obj)
        {
                if(m_current!=0)
                        throw std::logic_error("filter::append: filter is full");

                for(size_t i=0;i<m_ops.size();++i)
                {
                        if(!(m_ops[i](obj)))
                                return;
                }
                m_current=&obj;
        }

        void flush() {}

        bool empty() const
        {
                return m_ops.empty();
        }
};

template<typename T>
class splitter
:public pipe_facade<splitter<T>,T,T>
{
 private:

        friend class shore::pipeline_core_access;

        typedef boost::function<bool (const T&)> functor_t;

        std::vector<functor_t> m_ops;

        feed<T> m_fail;


        void append(const T &obj)
        {
                for(size_t i=0;i<m_ops.size();++i)
                {
                        if(!(m_ops[i](obj)))
                        {
                                m_fail.append(obj);
                                return;
                        }
                }
                this->emit(obj);
        }

        void flush()
        {
                m_fail.flush();
        }

 public:

        feed<T> &failures()
        {
                return m_fail;
        }

        void add_op(functor_t func)
        {
                m_ops.push_back(func);
        }

        bool empty() const
        {
                return m_ops.empty();
        }
};

} //namespace shore

#endif // SHORE_PROCESSING_FILTER_HPP__