dataexchange.h

Go to the documentation of this file.

/********************************************************************************
 *  FARSA                                                                       *
 *  Copyright (C) 2007-2012                                                     *
 *  Gianluca Massera <emmegian@yahoo.it>                                        *
 *  Stefano Nolfi <stefano.nolfi@istc.cnr.it>                                   *
 *  Tomassino Ferrauto <tomassino.ferrauto@istc.cnr.it>                         *
 *                                                                              *
 *  This program 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 2 of the License, or           *
 *  (at your option) any later version.                                         *
 *                                                                              *
 *  This program 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 this program; if not, write to the Free Software                 *
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA  *
 ********************************************************************************/

#ifndef DATAEXCHANGE_H
#define DATAEXCHANGE_H

#include <QEvent>
#include <QObject>
#include <QSet>
#include <QMutex>
#include <QMutexLocker>
#include <QWaitCondition>
#include <QSharedData>
#include <QExplicitlySharedDataPointer>
#include <memory>
#include "baseexception.h"
#include "utilitiesexceptions.h"

namespace farsa {

namespace __DataExchange_internal {
    class QueueHolderBase;
    template <class DataType_t>
    class QueueHolder;
}
template <class DataType_t>
class DatumToUpload;
template <class DataType_t>
class DataDownloader;
template <class DataType_t>
class Downloader;

template <class DataType_t>
class FARSA_UTIL_TEMPLATE DataUploader
{
public:
    typedef DataType_t DataType;

    typedef farsa::DatumToUpload<DataType> DatumToUpload;

    enum FullQueueBehavior {
        OverrideOlder, 
        BlockUploader, 
        IncreaseQueueSize, 
        SignalUploader 
    };

public:
    DataUploader(unsigned int queueSize, FullQueueBehavior b);

    ~DataUploader();

    unsigned int getQueueSize() const
    {
        return m_queueSize;
    }

    FullQueueBehavior getFullQueueBehavior() const
    {
        return m_fullQueueBehavior;
    }

    bool downloaderPresent() const;

    unsigned int getAvailableSpace() const;

    unsigned int getNumDataInQueue() const;

    DataType* createDatum();

    void uploadDatum();

    bool datumCreatedNotUploaded() const;

    void checkAssociationBeforeUpload(bool v)
    {
        m_checkAssociationBeforeUpload = v;
    }

    bool associationBeforeUploadChecked() const
    {
        return m_checkAssociationBeforeUpload;
    }

private:
    const unsigned int m_queueSize;

    const FullQueueBehavior m_fullQueueBehavior;

    QExplicitlySharedDataPointer<__DataExchange_internal::QueueHolder<DataType> > m_queue;

    bool m_checkAssociationBeforeUpload;

    DataUploader(const DataUploader<DataType>& other);

    DataUploader& operator=(const DataUploader<DataType>& other);

    friend class GlobalUploaderDownloader;
};

template <class DataType_t>
class FARSA_UTIL_TEMPLATE DatumToUpload
{
public:

    typedef DataType_t DataType;

public:
    DatumToUpload(DataUploader<DataType>& uploader) :
        m_uploader(uploader),
        m_datum(m_uploader.createDatum()),
        m_datumUploaded(false)
    {
    }

    ~DatumToUpload()
    {
        uploadDatum();
    }

    void uploadDatum()
    {
        if (!m_datumUploaded) {
            m_uploader.uploadDatum();
            m_datum = NULL;
            m_datumUploaded = true;
        }
    }

    const DataType* operator->() const
    {
        return m_datum;
    }

    DataType* operator->()
    {
        return m_datum;
    }

    bool operator==(const DataType* other) const
    {
        return (m_datum == other);
    }

    operator bool() const
    {
        return (m_datum != NULL);
    }

private:
    DataUploader<DataType>& m_uploader;

    DataType* m_datum;

    bool m_datumUploaded;

    DatumToUpload(const DatumToUpload<DataType>& other);

    DatumToUpload& operator=(const DatumToUpload<DataType>& other);
};

template<class DataType_t>
class FARSA_UTIL_TEMPLATE NewDatumEvent : public QEvent
{
public:
    static const QEvent::Type newDatumEventType = static_cast<QEvent::Type>(QEvent::User + 1);

    typedef DataType_t DataType;

public:
    NewDatumEvent(DataDownloader<DataType>* downloader) :
        QEvent(newDatumEventType),
        m_downloader(downloader)
    {
    }

    virtual ~NewDatumEvent()
    {
    }

    const DataDownloader<DataType>* getDownloader() const
    {
        return m_downloader;
    }

    DataDownloader<DataType>* getDownloader()
    {
        return m_downloader;
    }

private:
    DataDownloader<DataType>* const m_downloader;
};

template <class DataType_t>
class FARSA_UTIL_TEMPLATE NewDatumNotifiable
{
public:
    typedef DataType_t DataType;

public:
    virtual ~NewDatumNotifiable()
    {
    }

    virtual void newDatumAvailable(DataDownloader<DataType>* downloader) = 0;
};

template <class DataType_t>
class FARSA_UTIL_TEMPLATE DataDownloader
{
public:
    typedef DataType_t DataType;

    enum NewDatumAvailableBehavior {
        NoNotification, 
        NoNotificationBlocking, 
        QtEvent, 
        Callback 
    };

public:
    DataDownloader(NewDatumAvailableBehavior b);

    DataDownloader(QObject* o);

    DataDownloader(NewDatumNotifiable<DataType>* o);

    ~DataDownloader();

    NewDatumAvailableBehavior getNewDatumAvailableBehavior() const
    {
        return m_newDatumAvailableBehavior;
    }

    bool uploaderPresent() const;

    unsigned int getNumAvailableData() const;

    const DataType* downloadDatum();

private:
    void sendNotification();

    const NewDatumAvailableBehavior m_newDatumAvailableBehavior;

    QObject* const m_qoject;

    NewDatumNotifiable<DataType>* const m_newDatumNotifiable;

    QExplicitlySharedDataPointer<__DataExchange_internal::QueueHolder<DataType> > m_queue;

    mutable QMutex m_mutex;

    friend class DataUploader<DataType>;

    friend class GlobalUploaderDownloader;

    DataDownloader(const DataDownloader<DataType>& other);

    DataDownloader& operator=(const DataDownloader<DataType>& other);
};

template <class UploadedData_t, class DownloadedData_t>
class FARSA_UTIL_TEMPLATE DataUploaderDownloader : public DataUploader<UploadedData_t>, public DataDownloader<DownloadedData_t>
{
public:
    typedef UploadedData_t UploadedData;

    typedef DownloadedData_t DownloadedData;

    typedef typename DataUploader<UploadedData>::FullQueueBehavior FullQueueBehavior;

    typedef typename DataDownloader<DownloadedData>::NewDatumAvailableBehavior NewDatumAvailableBehavior;

public:
    DataUploaderDownloader(unsigned int uploadQueueSize, FullQueueBehavior fullQueueBehavior, NewDatumAvailableBehavior newDatumAvailableBehavior) :
        DataUploader<UploadedData>(uploadQueueSize, fullQueueBehavior),
        DataDownloader<DownloadedData>(newDatumAvailableBehavior)
    {
    }

    DataUploaderDownloader(unsigned int uploadQueueSize, FullQueueBehavior fullQueueBehavior, QObject* o) :
        DataUploader<UploadedData>(uploadQueueSize, fullQueueBehavior),
        DataDownloader<DownloadedData>(o)
    {
    }

    DataUploaderDownloader(unsigned int uploadQueueSize, FullQueueBehavior fullQueueBehavior, NewDatumNotifiable<DownloadedData>* o) :
        DataUploader<UploadedData>(uploadQueueSize, fullQueueBehavior),
        DataDownloader<DownloadedData>(o)
    {
    }

    ~DataUploaderDownloader()
    {
        // Nothing to do here
    }
};

class FARSA_UTIL_API GlobalUploaderDownloader
{
public:
    template <class DataType>
    static void associate(DataUploader<DataType>* uploader, DataDownloader<DataType>* downloader);

    template <class DataType1, class DataType2>
    static void associate(DataUploaderDownloader<DataType1, DataType2>* first, DataUploaderDownloader<DataType2, DataType1>* second);

    template <class DataType>
    static void detach(DataUploader<DataType>* uploader);

    template <class DataType>
    static void detach(DataDownloader<DataType>* downloader);

    template <class DataType1, class DataType2>
    static void detach(DataUploaderDownloader<DataType1, DataType2>* uploaderDownloader);

    static void stopAllDataExchanges();

private:
    GlobalUploaderDownloader();

    static GlobalUploaderDownloader& getInstance();

    template <class DataType>
    void internalAssociate(DataUploader<DataType>* uploader, DataDownloader<DataType>* downloader);

    template <class DataType1, class DataType2>
    void internalAssociate(DataUploaderDownloader<DataType1, DataType2>* first, DataUploaderDownloader<DataType2, DataType1>* second);

    template <class DataType>
    void internalDetach(DataUploader<DataType>* uploader);

    template <class DataType>
    void internalDetach(DataDownloader<DataType>* downloader);

    template <class DataType1, class DataType2>
    void internalDetach(DataUploaderDownloader<DataType1, DataType2>* uploaderDownloader);

    void internalStopAllDataExchanges();

    void addQueueHolder(__DataExchange_internal::QueueHolderBase* queueHolder);

    void removeQueueHolder(__DataExchange_internal::QueueHolderBase* queueHolder);

    template <class DataType>
    void uploaderDestroyed(DataUploader<DataType>* uploader);

    template <class DataType>
    void downloaderDestroyed(DataDownloader<DataType>* downloader);

    QSet<__DataExchange_internal::QueueHolderBase*> m_queueHolders;

    QMutex m_mutex;

    template <class DataType_t>
    friend class DataUploader;

    template <class DataType_t>
    friend class DataDownloader;

    template <class DataType_t>
    friend class __DataExchange_internal::QueueHolder;

private:
    GlobalUploaderDownloader(const GlobalUploaderDownloader&);

    GlobalUploaderDownloader& operator=(const GlobalUploaderDownloader&);
};

}  // end namespace farsa

// Implementation of all template members
#include <QCoreApplication>
#include <QMutex>
#include <QLinkedList>
#include <QWaitCondition>
#include <QMutexLocker>
#include <QtGlobal>

namespace farsa {

namespace __DataExchange_internal {
    class QueueHolderBase : public QSharedData
    {
    public:
        QueueHolderBase()
            : mutex()
            , waitCondition()
            , dataExchangeStopped(false)
        {
        }

    public:
        QMutex mutex;

        QWaitCondition waitCondition;

        bool dataExchangeStopped;
    };

    template <class DataType_t>
    class QueueHolder : public QueueHolderBase
    {
    public:
        typedef DataType_t DataType;

    public:
        QueueHolder(unsigned int queueSize, DataUploader<DataType>* u)
            : QueueHolderBase()
            , queue()
            , availableSpace(queueSize)
            , numDataInQueue(0)
            , queueFullLastDatumCreation(false)
            , datumCreatedNotUploaded(false)
            , nextUploaderDatum(NULL)
            , currentDownloaderDatum(NULL)
            , nextUploadIt()
            , nextDownloadIt()
            , uploader(u)
            , downloader(NULL)
        {
            // Allocating all memory. We use auto_ptr to ensure exception safety
            std::auto_ptr<DataType> uploaderDatum(new DataType());
            std::auto_ptr<DataType> downloaderDatum(new DataType());

            // Explicitly using a try-catch block to be exception-safe
            try {
                for (unsigned int i = 0; i < queueSize; i++) {
                    queue.push_back(new DataType());
                }
            } catch (...) {
                // If an exception is thrown, deleting all objects allocated so far
                foreach(DataType* d, queue) {
                    delete d;
                }

                // Propagating exception
                throw;
            }

            // Now initializing the iterators for the uploader and the downloader
            nextUploadIt = queue.begin();
            nextDownloadIt = queue.begin();

            // Now releasing the auto_ptrs
            nextUploaderDatum = uploaderDatum.release();
            currentDownloaderDatum = downloaderDatum.release();

            // Adding ourself to the list of queue holders
            GlobalUploaderDownloader::getInstance().addQueueHolder(this);
        }

        ~QueueHolder()
        {
            // Removing ourself from the list of queue holders
            GlobalUploaderDownloader::getInstance().removeQueueHolder(this);

            foreach(DataType* d, queue) {
                delete d;
            }
            queue.clear();

            delete nextUploaderDatum;
            delete currentDownloaderDatum;
        }

    public:
        QLinkedList<DataType*> queue;

        unsigned int availableSpace;

        unsigned int numDataInQueue;

        bool queueFullLastDatumCreation;

        bool datumCreatedNotUploaded;

        DataType* nextUploaderDatum;

        DataType* currentDownloaderDatum;

        typename QLinkedList<DataType*>::iterator nextUploadIt;

        typename QLinkedList<DataType*>::iterator nextDownloadIt;

        DataUploader<DataType>* uploader;

        DataDownloader<DataType>* downloader;
    };
}

template <class DataType_t>
DataUploader<DataType_t>::DataUploader(unsigned int queueSize, FullQueueBehavior b) :
    m_queueSize((queueSize == 0) ? 1 : queueSize),
    m_fullQueueBehavior(b),
    m_queue(new __DataExchange_internal::QueueHolder<DataType>(m_queueSize, this)),
    m_checkAssociationBeforeUpload(true)
{
}

template <class DataType_t>
DataUploader<DataType_t>::~DataUploader()
{
    // Calling the function of GlobalUploaderDownloader, it will take care of removing the
    // association, if present
    GlobalUploaderDownloader::getInstance().uploaderDestroyed(this);

    // The queue is deleted if we are the only one referencing it (we have used
    // QExplicitlySharedDataPointer)
}

template <class DataType_t>
bool DataUploader<DataType_t>::downloaderPresent() const
{
    QMutexLocker locker(&(m_queue->mutex));

    return (m_queue->downloader != NULL);
}

template <class DataType_t>
unsigned int DataUploader<DataType_t>::getAvailableSpace() const
{
    QMutexLocker locker(&(m_queue->mutex));

    if (m_fullQueueBehavior == IncreaseQueueSize) {
        return (m_queue->availableSpace == 0) ? 1 : m_queue->availableSpace;
    } else {
        return m_queue->availableSpace;
    }
}

template <class DataType_t>
unsigned int DataUploader<DataType_t>::getNumDataInQueue() const
{
    QMutexLocker locker(&(m_queue->mutex));

    return m_queue->numDataInQueue;
}

template <class DataType_t>
DataType_t* DataUploader<DataType_t>::createDatum()
{
    QMutexLocker locker(&(m_queue->mutex));

    // If the datum has already been created, returning the same datum again
    if (m_queue->datumCreatedNotUploaded) {
        return m_queue->nextUploaderDatum;
    }

    // Checking whether data exchange has been stopped
    if (m_queue->dataExchangeStopped) {
        return NULL;
    }

    // Checking if we are associated with a downloader if we have to
    if (m_checkAssociationBeforeUpload && (m_queue->downloader == NULL)) {
        throw UploaderDownloaderAssociationNotPresentException(UploaderDownloaderAssociationNotPresentException::DownloaderNotPresent);
    }

    // Resetting the flag signalling whether the queue was full
    m_queue->queueFullLastDatumCreation = false;

    // Checking if the queue is full
    if (m_queue->availableSpace == 0) {
        // Checking that uploader and downloader iterators point to the same location (this must
        // always happend if we get here)
        Q_ASSERT(m_queue->nextUploadIt == m_queue->nextDownloadIt);

        // The queue is full, setting the flag
        m_queue->queueFullLastDatumCreation = true;

        // What to do depends on the FullQueueBehavior
        switch (m_fullQueueBehavior) {
            case OverrideOlder:
                // The next datum would go where the downloader is going to take the element, so
                // we have to move the downloader forward
                ++m_queue->nextDownloadIt;
                if (m_queue->nextDownloadIt == m_queue->queue.end()) {
                    m_queue->nextDownloadIt = m_queue->queue.begin();
                }
                break;
            case BlockUploader:
                // Waiting on the wait condition, we will be woke up when the downloader downloads
                // an element. In that case the nextDownloadIt has been moved forward, so we can safely
                // use nextUploadIt
                m_queue->waitCondition.wait(&m_queue->mutex);

                // If we were woken up because data exchange has been stopped, simply returning NULL
                if (m_queue->dataExchangeStopped) {
                    return NULL;
                }
                break;
            case IncreaseQueueSize: {
                    // We have to add a datum to the queue at the current location. After the current
                    // datum has been uploaded, this element will be the nextUploaderDatum
                    std::auto_ptr<DataType> d(new DataType());
                    m_queue->nextUploadIt = m_queue->queue.insert(m_queue->nextDownloadIt, d.get());
                    d.release();
                } break;
            case SignalUploader:
                // Returning NULL to tell the uploader that there is no space
                return NULL;
                break;
        }
    }

    m_queue->datumCreatedNotUploaded = true;

    // Returning the datum to modify
    return m_queue->nextUploaderDatum;
}

template <class DataType_t>
void DataUploader<DataType_t>::uploadDatum()
{
    QMutexLocker locker(&(m_queue->mutex));

    // If the datum hasn't been created, doing nothing
    if (!m_queue->datumCreatedNotUploaded) {
        return;
    }

    // Checking whether data exchange has been stopped
    if (m_queue->dataExchangeStopped) {
        return;
    }

    // Putting the new datum in the queue, extracting the next element and moving the uploader iterator forward
    DataType* tmp = *m_queue->nextUploadIt;
    *m_queue->nextUploadIt = m_queue->nextUploaderDatum;
    m_queue->nextUploaderDatum = tmp;
    ++m_queue->nextUploadIt;
    if (m_queue->nextUploadIt == m_queue->queue.end()) {
        m_queue->nextUploadIt = m_queue->queue.begin();
    }

    // To update the variables with available space and data in queue, we have to check if the queue was full
    // and then check which is the FullQueueBehavior, to undestand what has been done in createDatum()
    if (m_queue->queueFullLastDatumCreation) {
        switch (m_fullQueueBehavior) {
            case OverrideOlder:
                // Nothing has changed, the queue is still full
                break;
            case BlockUploader:
                // The downloader has downloaded one datum before we could add the new one, so we have to update the counters
                --m_queue->availableSpace;
                ++m_queue->numDataInQueue;
                break;
            case IncreaseQueueSize:
                // The queue has grown, but still there is no space left. We only have to increase numDataInQueue
                ++m_queue->numDataInQueue;
                break;
            case SignalUploader:
                // Nothing has been done (we should never get here)
                Q_ASSERT(false);
                break;
        }
    } else {
        --m_queue->availableSpace;
        ++m_queue->numDataInQueue;
    }

    m_queue->datumCreatedNotUploaded = false;

    // Waking up the downloader, in case it was sleeping. This doesn't conflict with sendNotification() because if the
    // downloader is sleeping it is not expecting any notification
    m_queue->waitCondition.wakeAll();

    // Now we have to notify the downloader
    if (m_queue->downloader != NULL) {
        // If the downloader expects a callback to be called, we have to release the lock, otherwise a deadlock
        // is possible if the downloader tries to get the datum from inside the callback
        if (m_queue->downloader->m_newDatumAvailableBehavior == DataDownloader<DataType>::Callback) {
            locker.unlock();
        }

        // Notifying the downloader
        m_queue->downloader->sendNotification();
    }
}

template <class DataType_t>
bool DataUploader<DataType_t>::datumCreatedNotUploaded() const
{
    QMutexLocker locker(&(m_queue->mutex));

    return m_queue->datumCreatedNotUploaded;
}

template <class DataType_t>
DataDownloader<DataType_t>::DataDownloader(NewDatumAvailableBehavior b) :
    m_newDatumAvailableBehavior(b),
    m_qoject(NULL),
    m_newDatumNotifiable(NULL),
    m_queue(),
    m_mutex()
{
    if ((m_newDatumAvailableBehavior != NoNotification) && (m_newDatumAvailableBehavior != NoNotificationBlocking)) {
        if (m_newDatumAvailableBehavior == QtEvent) {
            throw InvalidNewDatumAvailableBehaviorException("when the NewDatumAvailableBehavior is \"QtEvent\" you must specify the QObject that receives the event");
        } else if (m_newDatumAvailableBehavior == Callback) {
            throw InvalidNewDatumAvailableBehaviorException("when the NewDatumAvailableBehavior is \"Callback\" you must specify the NewDatumNotifiable object whose callback is called");
        }
    }
}

template <class DataType_t>
DataDownloader<DataType_t>::DataDownloader(QObject* o) :
    m_newDatumAvailableBehavior(QtEvent),
    m_qoject(o),
    m_newDatumNotifiable(NULL),
    m_queue(),
    m_mutex()
{
    if (m_qoject == NULL) {
        throw InvalidNewDatumAvailableBehaviorException("when the NewDatumAvailableBehavior is \"QtEvent\" you must specify a valid (i.e. not NULL) QObject");
    }
}

template <class DataType_t>
DataDownloader<DataType_t>::DataDownloader(NewDatumNotifiable<DataType>* o) :
    m_newDatumAvailableBehavior(Callback),
    m_qoject(NULL),
    m_newDatumNotifiable(o),
    m_queue(),
    m_mutex()
{
    if (m_newDatumNotifiable == NULL) {
        throw InvalidNewDatumAvailableBehaviorException("when the NewDatumAvailableBehavior is \"Callback\" you must specify a valid (i.e. not NULL) NewDatumNotifiable");
    }
}

template <class DataType_t>
DataDownloader<DataType_t>::~DataDownloader()
{
    // Calling the function of GlobalUploaderDownloader, it will take care of removing the
    // association, if present
    GlobalUploaderDownloader::getInstance().downloaderDestroyed(this);

    // The queue is deleted if we are the only one referencing it (we have used
    // QExplicitlySharedDataPointer)
}

template <class DataType_t>
bool DataDownloader<DataType_t>::uploaderPresent() const
{
    QMutexLocker internalLocker(&m_mutex);

    if (!m_queue) {
        return NULL;
    }

    QMutexLocker locker(&m_queue->mutex);

    return (m_queue->uploader != NULL);
}

template <class DataType_t>
unsigned int DataDownloader<DataType_t>::getNumAvailableData() const
{
    QMutexLocker internalLocker(&m_mutex);

    if (!m_queue) {
        throw UploaderDownloaderAssociationNotPresentException(UploaderDownloaderAssociationNotPresentException::UploaderNotPresent);
    }

    QMutexLocker locker(&m_queue->mutex);

    return m_queue->numDataInQueue;
}

template <class DataType_t>
const DataType_t* DataDownloader<DataType_t>::downloadDatum()
{
    QMutexLocker internalLocker(&m_mutex);

    // Checking if the queue is there, We allow to download even if no uploader is present to
    // be able to remove the last data from the queue
    if (!m_queue) {
        throw UploaderDownloaderAssociationNotPresentException(UploaderDownloaderAssociationNotPresentException::UploaderNotPresent);
    }

    QMutexLocker locker(&(m_queue->mutex));

    // Checking whether data exchange has been stopped
    if (m_queue->dataExchangeStopped) {
        return NULL;
    }

    // Checking if the queue is empty
    if (m_queue->numDataInQueue == 0) {
        // Checking that uploader and downloader iterators point to the same location (this must
        // always happend if we get here)
        Q_ASSERT(m_queue->nextUploadIt == m_queue->nextDownloadIt);

        // What to do depends on the NewDatumAvailableBehavior
        switch (m_newDatumAvailableBehavior) {
            case NoNotificationBlocking:
                // Waiting on the wait condition, we will be woke up when the uploader has uploaded something
                m_queue->waitCondition.wait(&m_queue->mutex);

                // If we were woken up because data exchange has been stopped, simply returning NULL
                if (m_queue->dataExchangeStopped) {
                    return NULL;
                }
                break;
            default:
                // In all the other cases we return NULL to tell that no datum is available
                return NULL;
                break;
        }
    }

    // We can return the current datum. We have to put back in the queue the datum that was downloaded before and move
    // the iterator forward
    DataType* oldDatum = m_queue->currentDownloaderDatum;
    m_queue->currentDownloaderDatum = *m_queue->nextDownloadIt;
    *m_queue->nextDownloadIt = oldDatum;
    ++m_queue->nextDownloadIt;
    if (m_queue->nextDownloadIt == m_queue->queue.end()) {
        m_queue->nextDownloadIt = m_queue->queue.begin();
    }

    // Incrementing the available space and decrementing the number of data in the queue
    ++m_queue->availableSpace;
    --m_queue->numDataInQueue;

    // Waking up the uploader, in case it was sleeping
    m_queue->waitCondition.wakeAll();

    return m_queue->currentDownloaderDatum;
}

template <class DataType_t>
void DataDownloader<DataType_t>::sendNotification()
{
    // What we do here depends on the NewDatumAvailableBehavior
    switch (m_newDatumAvailableBehavior) {
        case NoNotification:
        case NoNotificationBlocking:
            // Nothing to do
            break;
        case QtEvent:
            // Posting a QT event
            Q_ASSERT(m_qoject != NULL);
            QCoreApplication::postEvent(m_qoject, new NewDatumEvent<DataType>(this));
            break;
        case Callback:
            // Calling the callback
            Q_ASSERT(m_newDatumNotifiable != NULL);
            m_newDatumNotifiable->newDatumAvailable(this);
            break;
    }
}

template <class DataType>
void GlobalUploaderDownloader::associate(DataUploader<DataType>* uploader, DataDownloader<DataType>* downloader)
{
        getInstance().internalAssociate(uploader, downloader);
}

template <class DataType1, class DataType2>
void GlobalUploaderDownloader::associate(DataUploaderDownloader<DataType1, DataType2>* first, DataUploaderDownloader<DataType2, DataType1>* second)
{
        getInstance().internalAssociate(first, second);
}

template <class DataType>
void GlobalUploaderDownloader::detach(DataUploader<DataType>* uploader)
{
        getInstance().internalDetach(uploader);
}

template <class DataType>
void GlobalUploaderDownloader::detach(DataDownloader<DataType>* downloader)
{
        getInstance().internalDetach(downloader);
}

template <class DataType1, class DataType2>
void GlobalUploaderDownloader::detach(DataUploaderDownloader<DataType1, DataType2>* uploaderDownloader)
{
        getInstance().internalDetach(uploaderDownloader);
}

template <class DataType>
void GlobalUploaderDownloader::internalAssociate(DataUploader<DataType>* uploader, DataDownloader<DataType>* downloader)
{
    // Locking order is the same in all functions: our mutex, downloader mutex, queue mutex
    QMutexLocker locker(&m_mutex);

    // Locking the downloader mutex
    QMutexLocker downloaderMutexLocker(&(downloader->m_mutex));

    // Now also locking the queue mutex (we take it from the uploader)
    QMutexLocker queueMutexLocker(&(uploader->m_queue->mutex));

    // First of all checking that neither the uploader nor the downloader is already associated
    if (uploader->m_queue->downloader != NULL) {
        throw UploaderDownloaderAssociationNotUniqueException(UploaderDownloaderAssociationNotUniqueException::UploaderAlreadyAssociated);
    } else if (downloader->m_queue && (downloader->m_queue->uploader != NULL)) {
        throw UploaderDownloaderAssociationNotUniqueException(UploaderDownloaderAssociationNotUniqueException::DownloaderAlreadyAssociated);
    }

    // No need to check whether a datum has been created but not uploaded as both ends are not associated!

    // Creating the association. The old queue in the downloader is deleted by QExplicitlySharedDataPointer
    // if present
    uploader->m_queue->downloader = downloader;
    uploader->m_queue->uploader = uploader;
    downloader->m_queue = uploader->m_queue;

    // If there are data available in the queue, we must notify the downloader
    if (uploader->m_queue->numDataInQueue != 0) {
        downloader->sendNotification();
    }
}

template <class DataType1, class DataType2>
void GlobalUploaderDownloader::internalAssociate(DataUploaderDownloader<DataType1, DataType2>* first, DataUploaderDownloader<DataType2, DataType1>* second)
{
    // Locking order is the same in all functions: our mutex, downloader mutex, queue mutex
    QMutexLocker locker(&m_mutex);

    // Splitting downloaders and uploaders
    DataUploader<DataType1>* const firstUploader = first;
    DataDownloader<DataType2>* const firstDownloader = first;
    DataUploader<DataType2>* const secondUploader = second;
    DataDownloader<DataType1>* const secondDownloader = second;

    // Taking all locks
    QMutexLocker firstDownloaderMutexLocker(&(firstDownloader->m_mutex));
    QMutexLocker firstQueueMutexLocker(&(firstUploader->m_queue->mutex));
    QMutexLocker secondDownloaderMutexLocker(&(secondDownloader->m_mutex));
    QMutexLocker secondQueueMutexLocker(&(secondUploader->m_queue->mutex));

    // First of all checking that neither uploaders nor downloaders are already associated
    if (firstUploader->m_queue->downloader != NULL) {
        throw UploaderDownloaderAssociationNotUniqueException(UploaderDownloaderAssociationNotUniqueException::UploaderAlreadyAssociated);
    } else if (firstDownloader->m_queue && (firstDownloader->m_queue->uploader != NULL)) {
        throw UploaderDownloaderAssociationNotUniqueException(UploaderDownloaderAssociationNotUniqueException::DownloaderAlreadyAssociated);
    } else if (secondUploader->m_queue->downloader != NULL) {
        throw UploaderDownloaderAssociationNotUniqueException(UploaderDownloaderAssociationNotUniqueException::UploaderAlreadyAssociated);
    } else if (secondDownloader->m_queue && (secondDownloader->m_queue->uploader != NULL)) {
        throw UploaderDownloaderAssociationNotUniqueException(UploaderDownloaderAssociationNotUniqueException::DownloaderAlreadyAssociated);
    }

    // No need to check whether a datum has been created but not uploaded as both ends are not associated!

    // Creating the association. The old queue in downloaders is deleted by QExplicitlySharedDataPointer
    // if present
    firstUploader->m_queue->downloader = secondDownloader;
    firstUploader->m_queue->uploader = firstUploader;
    secondDownloader->m_queue = firstUploader->m_queue;
    secondUploader->m_queue->downloader = firstDownloader;
    secondUploader->m_queue->uploader = secondUploader;
    firstDownloader->m_queue = secondUploader->m_queue;

    // If there are data available in the queue, we must notify the downloader
    if (firstUploader->m_queue->numDataInQueue != 0) {
        secondDownloader->sendNotification();
    }
    if (secondUploader->m_queue->numDataInQueue != 0) {
        firstDownloader->sendNotification();
    }
}

template <class DataType>
void GlobalUploaderDownloader::internalDetach(DataUploader<DataType>* uploader)
{
    // Locking order is the same in all functions: our mutex, downloader mutex, queue mutex
    // Here we must take the downloader from the queue and we do this without locking. This
    // is fine because the pointer to the downloader in the queue is protected by the mutex
    // in GlobalUploaderDownloader (which is locked first)
    QMutexLocker locker(&m_mutex);

    // If no association is present, returning directly
    DataDownloader<DataType>* const downloader = uploader->m_queue->downloader;
    if (downloader == NULL) {
        return;
    }

    // Locking the downloader mutex
    QMutexLocker downloaderMutexLocker(&(downloader->m_mutex));

    // Now also locking the queue mutex (we take it from the uploader)
    QMutexLocker queueMutexLocker(&(uploader->m_queue->mutex));

    // Removing the association
    uploader->m_queue->downloader = NULL;

    // Unlocking the lock on the queue because here it could be destroyed and if it
    // isn't, the lock is not necessary
    queueMutexLocker.unlock();
    downloader->m_queue.reset();
}

template <class DataType>
void GlobalUploaderDownloader::internalDetach(DataDownloader<DataType>* downloader)
{
    // Locking order is the same in all functions: our mutex, downloader mutex, queue mutex
    QMutexLocker locker(&m_mutex);

    // Locking the downloader mutex
    QMutexLocker downloaderMutexLocker(&(downloader->m_mutex));

    // If no association is present, returning directly
    if (!downloader->m_queue) {
        return;
    }

    // Now also locking the queue mutex (we take it from the downloader)
    QMutexLocker queueMutexLocker(&(downloader->m_queue->mutex));

    // Removing the association
    downloader->m_queue->downloader = NULL;

    // Unlocking the lock on the queue because here it could be destroyed and if it
    // isn't, the lock is not necessary
    queueMutexLocker.unlock();
    downloader->m_queue.reset();
}

template <class DataType1, class DataType2>
void GlobalUploaderDownloader::internalDetach(DataUploaderDownloader<DataType1, DataType2>* uploaderDownloader)
{
    // Locking order is the same in all functions: our mutex, downloader mutex, queue mutex
    // Here we must take the downloader associated with the uploader in uploaderDownloader from
    // the queue and we do this without locking. This is fine because the pointer to the downloader
    // in the queue is protected by the mutex in GlobalUploaderDownloader (which is locked first)
    QMutexLocker locker(&m_mutex);

    // Splitting downloader and uploader
    DataUploader<DataType1>* const uploader = uploaderDownloader;
    DataDownloader<DataType2>* const downloader = uploaderDownloader;

    // Taking all locks. To be able to use RAII with the mutex of the downloader associated with
    // the uploader of uploaderDownloader and with the queue associated with the downloader of
    // uploaderDownloader (which may not exist), we use an std::auto_ptr
    std::auto_ptr<QMutexLocker> otherDownloaderMutexLocker;
    DataDownloader<DataType1>* const otherDownloader = uploader->m_queue->downloader;
    if (otherDownloader != NULL) {
        otherDownloaderMutexLocker.reset(new QMutexLocker(&(otherDownloader->m_mutex)));
    }
    QMutexLocker queueMutexLocker(&(uploader->m_queue->mutex));
    QMutexLocker downloaderMutexLocker(&(downloader->m_mutex));
    std::auto_ptr<QMutexLocker> otherQueueMutexLocker;
    if (downloader->m_queue) {
        otherQueueMutexLocker.reset(new QMutexLocker(&(downloader->m_queue->mutex)));
    }

    // Removing all associations
    uploader->m_queue->downloader = NULL;
    if (otherDownloader != NULL) {
        // Unlocking the lock on the queue because here it could be destroyed and if it
        // isn't, the lock is not necessary
        otherDownloaderMutexLocker->unlock();
        otherDownloader->m_queue.reset();
    }
    if (downloader->m_queue) {
        downloader->m_queue->downloader = NULL;

        // Unlocking the lock on the queue because here it could be destroyed and if it
        // isn't, the lock is not necessary
        otherQueueMutexLocker->unlock();
        downloader->m_queue.reset();
    }
}

template <class DataType>
void GlobalUploaderDownloader::uploaderDestroyed(DataUploader<DataType>* uploader)
{
    // Locking order is the same in all functions: our mutex, downloader mutex, queue mutex
    // Here we must take the downloader from the queue and we do this without locking. This
    // is fine because the pointer to the downloader in the queue is protected by the mutex
    // in GlobalUploaderDownloader (which is locked first)
    QMutexLocker locker(&m_mutex);

    // If no association is present, returning directly
    DataDownloader<DataType>* const downloader = uploader->m_queue->downloader;
    if (downloader == NULL) {
        return;
    }

    // Locking the downloader mutex
    QMutexLocker downloaderMutexLocker(&(downloader->m_mutex));

    // Now also locking the queue mutex (we take it from the uploader)
    QMutexLocker queueMutexLocker(&(uploader->m_queue->mutex));

    // Removing the association
    uploader->m_queue->uploader = NULL;

    // Unlocking the lock on the queue because here it could be destroyed and if it
    // isn't, the lock is not necessary
    queueMutexLocker.unlock();
    uploader->m_queue.reset();
}

template <class DataType>
void GlobalUploaderDownloader::downloaderDestroyed(DataDownloader<DataType>* downloader)
{
    // Here we can simply call detach on the downloader
    internalDetach(downloader);
}

} // end namespace farsa

#endif