InputPort.h

Go to the documentation of this file.

/* ============================================================
 *
 * This file is a part of CCA project
 *
 * Copyright (C) 2011 by Arne Nordmann <anordman at cor-lab dot uni-bielefeld dot de>
 *
 * This file may be licensed under the terms of the
 * GNU Lesser General Public License Version 3 (the ``LGPL''),
 * or (at your option) any later version.
 *
 * Software distributed under the License is distributed
 * on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
 * express or implied. See the LGPL for the specific language
 * governing rights and limitations.
 *
 * You should have received a copy of the LGPL along with this
 * program. If not, go to http://www.gnu.org/licenses/lgpl.html
 * or write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 * The development of this software was supported by:
 *   CoR-Lab, Research Institute for Cognition and Robotics
 *     Bielefeld University
 *
 * ============================================================ */

#pragma once

#include <boost/shared_ptr.hpp>
#include <boost/bind.hpp>

#include <rsb/Listener.h>
#include <rsb/Handler.h>
#include <rsb/Event.h>

#include "cca/Port.h"
#include "cca/buffer/SingleItemBuffer.h"
#include "cca/buffer/SynchronizedQueue.h"

namespace cca {

class InputPortBase;
typedef ::boost::shared_ptr<InputPortBase> InputPortPtr;

class InputPortBase: public Port {

public:

    InputPortBase(bool optional = false,
            const std::string &datatype = "unknown");

    virtual void configureSpecifics();

    virtual void setHandler(::rsb::HandlerPtr handler);

    virtual void purge() = 0;

    virtual bool empty() const = 0;

    virtual bool newItem() const = 0;

    virtual unsigned int size() const = 0;

    virtual ~InputPortBase();

protected:

    class InputHandler: public rsb::Handler {
    public:
        InputHandler(const std::string &datatype) :
                rsb::Handler(), callback(), handler(), mydatatype("unknown") {
            mydatatype = datatype;
        }

        void setNextHandler(rsb::HandlerPtr h) {
            this->handler = h;
        }

        void setInputCallback(const boost::function1<void, rsb::EventPtr> &c) {
            callback = c;
        }

        void handle(rsb::EventPtr event) {
            this->callback(event);

            rsb::HandlerPtr h = this->handler;
            if (h) {
                h->handle(event);
            }
        }

        void printContents(std::ostream& stream) const {
            stream << "type = " << mydatatype;
        }

    protected:
        boost::function1<void, rsb::EventPtr> callback;
        rsb::HandlerPtr handler;
        std::string mydatatype;
    };

    boost::shared_ptr<InputHandler> handler;

    InputPortBase(InputPortBase &port);

    void operator=(const InputPortBase &port);

    ::rsb::ListenerPtr listener;
    rsc::logging::LoggerPtr logger;
};

template<typename DATATYPE>
class InputPort: public InputPortBase, public Buffering<DATATYPE> {

public:

    typedef ::boost::shared_ptr<DATATYPE> DataPtr;

    typedef ::boost::shared_ptr<InputPort<DATATYPE> > Ptr;

    typedef typename Buffer<DATATYPE>::Ptr BufferPtr;

    static Ptr create() {
        return Ptr(new InputPort<DATATYPE>());
    }

    virtual void configureSpecifics() {
        RSCTRACE(logger, "InputPort::configureSpecifics()");
        // Buffering
        if (config->getQueueSize() > 1) {
            this->buffer = typename Buffer<DATATYPE>::Ptr(
                    new cca::SynchronizedQueue<DATATYPE>(config->getQueueSize(),
                            config->keepLatest()));
        } else {
            this->buffer = typename Buffer<DATATYPE>::Ptr(
                    new cca::SingleItemBuffer<DATATYPE>(config->keepLatest()));
        }

        handler->setInputCallback(
                boost::bind(&InputPort<DATATYPE>::addEvent, this, _1));

        InputPortBase::configureSpecifics();
    }

    InputPort(bool optional = false) :
            InputPortBase(optional, rsc::runtime::typeName<DATATYPE>()),
                    buffer(),
                    logger(rsc::logging::Logger::getLogger("cca.inputport")) {
    }

    virtual DataPtr get() throw () {
        if (this->configured) {
            return buffer->get();
        }
        return DataPtr();
    }

    virtual void addEvent(rsb::EventPtr item) {
        if (this->configured) {
            buffer->add(boost::static_pointer_cast<DATATYPE>(item->getData()));
        }
    }

    virtual void add(DataPtr item) {
        RSCTRACE(logger, "InputPort::add()");
        if (this->configured) {
            buffer->add(item);
        }
    }

    virtual bool empty() const {
        if (this->configured) {
            return buffer->empty();
        }
        return true;
    }

    virtual bool newItem() const {
        if (this->configured) {
            return buffer->newItem();
        }
        return false;
    }

    virtual unsigned int size() const {
        if (this->configured) {
            return buffer->size();
        }
        return 0;
    }

    virtual void purge() {
        if (this->configured) {
            buffer->purge();
        }
    }

    virtual std::string print() const {
        ::std::ostringstream outstream(::std::ostringstream::out);
        outstream << "<" << rsc::runtime::typeName<DATATYPE>() << ">";
        if (configured) {
            outstream << " " << getConfig()->print();
            outstream << "(" << getScopePtr()->toString() << ")";
        } else {
            outstream << " <unconfigured>";
        }
        if (optional) {
            outstream << " [optional]";
        }
        return outstream.str();
    }

    virtual std::string bla() const {
        ::std::ostringstream outstream(::std::ostringstream::out);
        outstream << "<" << rsc::runtime::typeName<DATATYPE>() << ">";
        return outstream.str();
    }

    virtual ~InputPort() {
    }

protected:

    InputPort(InputPort &port);

    void operator=(const InputPort &port);

    BufferPtr buffer;
    rsc::logging::LoggerPtr logger;

};

}