OutputPort.hpp

Go to the documentation of this file.

/***************************************************************************
  tag: Peter Soetens  Thu Oct 22 11:59:08 CEST 2009  OutputPort.hpp

                        OutputPort.hpp -  description
                           -------------------
    begin                : Thu October 22 2009
    copyright            : (C) 2009 Sylvain Joyeux
    email                : sylvain.joyeux@m4x.org

 ***************************************************************************
 *   This library 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;                 *
 *   version 2 of the License.                                             *
 *                                                                         *
 *   As a special exception, you may use this file as part of a free       *
 *   software library without restriction.  Specifically, if other files   *
 *   instantiate templates or use macros or inline functions from this     *
 *   file, or you compile this file and link it with other files to        *
 *   produce an executable, this file does not by itself cause the         *
 *   resulting executable to be covered by the GNU General Public          *
 *   License.  This exception does not however invalidate any other        *
 *   reasons why the executable file might be covered by the GNU General   *
 *   Public License.                                                       *
 *                                                                         *
 *   This library 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     *
 *   Lesser General Public License for more details.                       *
 *                                                                         *
 *   You should have received a copy of the GNU General Public             *
 *   License along with this library; if not, write to the Free Software   *
 *   Foundation, Inc., 59 Temple Place,                                    *
 *   Suite 330, Boston, MA  02111-1307  USA                                *
 *                                                                         *
 ***************************************************************************/


#ifndef ORO_OUTPUT_PORT_HPP
#define ORO_OUTPUT_PORT_HPP

#include "base/OutputPortInterface.hpp"
#include "base/DataObject.hpp"
#include "internal/DataObjectDataSource.hpp"
#include "internal/Channels.hpp"
#include "internal/ConnFactory.hpp"
#include "Service.hpp"
#include "OperationCaller.hpp"

#include "InputPort.hpp"

namespace RTT
{
    template<typename T>
    class OutputPort : public base::OutputPortInterface
    {
    private:
        friend class internal::ConnInputEndpoint<T>;
        typename internal::ConnInputEndpoint<T>::shared_ptr endpoint;

        virtual bool connectionAdded( base::ChannelElementBase::shared_ptr channel_input, ConnPolicy const& policy ) {
            // Initialize the new channel with last written data if requested
            // (and available)

            // This this the input channel element of the whole connection
            typename base::ChannelElement<T>::shared_ptr channel_el_input = channel_input.get()->narrow<T>();

            if (has_initial_sample)
            {
                T const& initial_sample = sample->Get();
                if ( channel_el_input->data_sample(initial_sample, /* reset = */ false) != NotConnected ) {
                    if ( has_last_written_value && policy.init )
                        return ( channel_el_input->write(initial_sample) != NotConnected );
                    return true;
                } else {
                    Logger::In in("OutputPort");
                    log(Error) << "Failed to pass data sample to data channel. Aborting connection."<<endlog();
                    return false;
                }
            }

            // even if we're not written, test the connection with a default sample.
            return ( channel_el_input->data_sample( T(), /* reset = */ false ) != NotConnected );
        }

        bool has_last_written_value;
        // data_sample or by calling write() with keeps_next_written_value or
        // keeps_last_written_value to true
        bool has_initial_sample;
        // This is used to initialize connections with a known sample
        bool keeps_next_written_value;
        // This is used to allow the use of the 'init' connection policy option
        bool keeps_last_written_value;
        typename base::DataObjectInterface<T>::shared_ptr sample;

        OutputPort( OutputPort const& orig );
        OutputPort& operator=(OutputPort const& orig);

    public:
        OutputPort(std::string const& name = "unnamed", bool keep_last_written_value = true)
            : base::OutputPortInterface(name)
            , endpoint(new internal::ConnInputEndpoint<T>(this))
            , has_last_written_value(false)
            , has_initial_sample(false)
            , keeps_next_written_value(false)
            , keeps_last_written_value(false)
            , sample( new base::DataObject<T>() )
        {
            if (keep_last_written_value)
                keepLastWrittenValue(true);
        }

        virtual ~OutputPort() { disconnect(); }

        void keepNextWrittenValue(bool keep)
        {
            keeps_next_written_value = keep;
        }

        void keepLastWrittenValue(bool keep)
        {
            keeps_last_written_value = keep;
        }

        bool keepsLastWrittenValue() const { return keeps_last_written_value; }

        T getLastWrittenValue() const
        {
            return sample->Get();
        }

        bool getLastWrittenValue(T& sample) const
        {
            if (has_last_written_value)
            {
                this->sample->Get(sample);
                return true;
            }
            return false;
        }

        virtual base::DataSourceBase::shared_ptr getDataSource() const
        {
            // we create this on the fly.
            return new internal::DataObjectDataSource<T>( sample );
        }

        void setDataSample(const T& sample)
        {
            this->sample->Set(sample);
            has_initial_sample = true;
            has_last_written_value = false;

            if (connected()) {
                WriteStatus result = getEndpoint()->getWriteEndpoint()->data_sample(sample, /* reset = */ true);
                if (result == NotConnected) {
                    log(Error) << "A channel of port " << getName() << " has been invalidated during setDataSample(), it will be removed" << endlog();
                }
            }
        }

        void clear()
        {
            has_last_written_value = false;
            getEndpoint()->getWriteEndpoint()->clear(); // only affects shared pull connections, where getInputEndpoint() would return the port's buffer object

            // eventually clear shared connection
            internal::SharedConnectionBase::shared_ptr shared_connection = cmanager.getSharedConnection();
            if (shared_connection) {
                shared_connection->clear();
            }
        }

        WriteStatus write(const T& sample)
        {
            if (keeps_last_written_value || keeps_next_written_value)
            {
                keeps_next_written_value = false;
                has_initial_sample = true;
                this->sample->Set(sample);
            }
            has_last_written_value = keeps_last_written_value;

            WriteStatus result = NotConnected;
            if (connected()) {
                result = getEndpoint()->getWriteEndpoint()->write(sample);
                if (result == NotConnected) {
                    log(Error) << "A channel of port " << getName() << " has been invalidated during write(), it will be removed" << endlog();
                }
            }

            return result;
        }

        WriteStatus write(base::DataSourceBase::shared_ptr source)
        {
            typename internal::AssignableDataSource<T>::shared_ptr ds =
                boost::dynamic_pointer_cast< internal::AssignableDataSource<T> >(source);
            if (ds)
                return write(ds->rvalue());
            else
            {
                typename internal::DataSource<T>::shared_ptr ds =
                    boost::dynamic_pointer_cast< internal::DataSource<T> >(source);
                if (ds)
                    return write(ds->get());
                else
                    log(Error) << "trying to write from an incompatible data source" << endlog();
            }
            return WriteFailure;
        }

        virtual const types::TypeInfo* getTypeInfo() const
        { return internal::DataSourceTypeInfo<T>::getTypeInfo(); }

        virtual base::PortInterface* clone() const
        { return new OutputPort<T>(this->getName()); }

        virtual base::PortInterface* antiClone() const
        { return new InputPort<T>(this->getName()); }

        using base::OutputPortInterface::createConnection;

        virtual bool createConnection(base::InputPortInterface& input_port, ConnPolicy const& policy)
        {
            return internal::ConnFactory::createConnection(*this, input_port, policy);
        }

        virtual bool createStream(ConnPolicy const& policy)
        {
            return internal::ConnFactory::createStream(*this, policy);
        }

#ifndef ORO_DISABLE_PORT_DATA_SCRIPTING

        virtual Service* createPortObject()
        {
            Service* object = base::OutputPortInterface::createPortObject();
            // Force resolution on the overloaded write method
            typedef WriteStatus (OutputPort<T>::*WriteSample)(T const&);
            WriteSample write_m = &OutputPort::write;
            typedef T (OutputPort<T>::*LastSample)() const;
            LastSample last_m = &OutputPort::getLastWrittenValue;
            object->addSynchronousOperation("write", write_m, this).doc("Writes a sample on the port.").arg("sample", "");
            object->addSynchronousOperation("last", last_m, this).doc("Returns last written value to this port.");
            return object;
        }
#endif

        virtual internal::ConnInputEndpoint<T>* getEndpoint() const
        {
            assert(endpoint);
            return endpoint.get();
        }

        virtual typename base::ChannelElement<T>::shared_ptr getSharedBuffer() const
        {
            return getEndpoint()->getSharedBuffer();
        }
    };

}

#endif