/* SPDX-License-Identifier: MPL-2.0 */

#ifndef __ZMQ_YPIPE_CONFLATE_HPP_INCLUDED__

#define __ZMQ_YPIPE_CONFLATE_HPP_INCLUDED__

#include "platform.hpp"

#include "dbuffer.hpp"

#include "ypipe_base.hpp"

namespace zmq

{

//  Adapter for dbuffer, to plug it in instead of a queue for the sake

//  of implementing the conflate socket option, which, if set, makes

//  the receiving side to discard all incoming messages but the last one.

//

//  reader_awake flag is needed here to mimic ypipe delicate behaviour

//  around the reader being asleep (see 'c' pointer being NULL in ypipe.hpp)

template <typename T> class ypipe_conflate_t ZMQ_FINAL : public ypipe_base_t<T>

{

  public:

    //  Initialises the pipe.

    ypipe_conflate_t () : reader_awake (false) {}

    //  Following function (write) deliberately copies uninitialised data

    //  when used with zmq_msg. Initialising the VSM body for

    //  non-VSM messages won't be good for performance.

#ifdef ZMQ_HAVE_OPENVMS

#pragma message save

#pragma message disable(UNINIT)

#endif

    void write (const T &value_, bool incomplete_)

    {

        (void) incomplete_;

        dbuffer.write (value_);

    }

#ifdef ZMQ_HAVE_OPENVMS

#pragma message restore

#endif

    // There are no incomplete items for conflate ypipe

    bool unwrite (T *) { return false; }

    //  Flush is no-op for conflate ypipe. Reader asleep behaviour

    //  is as of the usual ypipe.

    //  Returns false if the reader thread is sleeping. In that case,

    //  caller is obliged to wake the reader up before using the pipe again.

    bool flush () { return reader_awake; }

    //  Check whether item is available for reading.

    bool check_read ()

    {

        const bool res = dbuffer.check_read ();

        if (!res)

            reader_awake = false;

        return res;

    }

    //  Reads an item from the pipe. Returns false if there is no value.

    //  available.

    bool read (T *value_)

    {

        if (!check_read ())

            return false;

        return dbuffer.read (value_);

    }

    //  Applies the function fn to the first element in the pipe

    //  and returns the value returned by the fn.

    //  The pipe mustn't be empty or the function crashes.

    bool probe (bool (*fn_) (const T &)) { return dbuffer.probe (fn_); }

  protected:

    dbuffer_t<T> dbuffer;

    bool reader_awake;

    ZMQ_NON_COPYABLE_NOR_MOVABLE (ypipe_conflate_t)

};

}

#endif