// Copyright 2020 Proyectos y Sistemas de Mantenimiento SL (eProsima).

//

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

//     http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

/**

 * @file DataSharingNotification.hpp

 */

#ifndef RTPS_HISTORY_DATASHARINGNOTIFICATION_HPP

#define RTPS_HISTORY_DATASHARINGNOTIFICATION_HPP

#include <fastdds/dds/log/Log.hpp>

#include <rtps/history/PoolConfig.h>

#include <utils/shared_memory/SharedMemSegment.hpp>

#include <utils/shared_memory/SharedDir.hpp>

#include <fastdds/rtps/common/Guid.hpp>

#include <memory>

#include <vector>

#include <mutex>

#include <atomic>

namespace eprosima {

namespace fastdds {

namespace rtps {

class DataSharingNotification

{

    friend class DataSharingListener;

    friend class DataSharingNotifier;

public:

    typedef fastdds::rtps::SharedSegmentBase Segment;

    DataSharingNotification() = default;

    virtual ~DataSharingNotification() = default;

    /**

     * Notifies of new data

     */

    inline void notify()

    {

        try

        {

            std::unique_lock<Segment::mutex> lock(notification_->notification_mutex);

            notification_->new_data.store(true);

            lock.unlock();

            notification_->notification_cv.notify_all();

        }

        catch (const boost::interprocess::interprocess_exception& /*e*/)

        {

            // Timeout when locking

        }

    }

    /**

     * Returns the GUID of the reader listening to the notifications

     */

    inline const GUID_t&  reader() const

    {

        return segment_id_;

    }

    static std::shared_ptr<DataSharingNotification> create_notification(

            const GUID_t& reader_guid,

            const std::string& shared_dir = std::string());

    static std::shared_ptr<DataSharingNotification> open_notification(

            const GUID_t& reader_guid,

            const std::string& shared_dir = std::string());

    void destroy();

    static std::string get_default_directory()

    {

        std::string dir;

        fastdds::rtps::SharedDir::get_default_shared_dir(dir);

        return dir;

    }

    constexpr static const char* domain_name()

    {

        return "fast_datasharing";

    }

protected:

#pragma warning(push)

#pragma warning(disable:4324)

    struct alignas (8) Notification

    {

        //! CV to wait for new notifications

        Segment::condition_variable notification_cv;

        //! synchronization mutex

        Segment::mutex notification_mutex;

        //! New data available

        std::atomic<bool> new_data;

    };

#pragma warning(pop)

    static std::string generate_segment_name(

            const std::string& shared_dir,

            const GUID_t& reader_guid)

    {

        std::stringstream ss;

        if (!shared_dir.empty())

        {

            ss << shared_dir << "/";

        }

        ss << DataSharingNotification::domain_name() << "_" << reader_guid.guidPrefix << "_" << reader_guid.entityId;

        return ss.str();

    }

    bool create_and_init_notification(

            const GUID_t& reader_guid,

            const std::string& shared_dir = std::string());

    bool open_and_init_notification(

            const GUID_t& reader_guid,

            const std::string& shared_dir = std::string());

    template <typename T>

    bool create_and_init_shared_segment_notification(

            const GUID_t& reader_guid,

            const std::string& shared_dir)

    {

        segment_id_ = reader_guid;

        segment_name_ = generate_segment_name(shared_dir, reader_guid);

        std::unique_ptr<T> local_segment;

        try

        {

            uint32_t per_allocation_extra_size = T::compute_per_allocation_extra_size(

                alignof(Notification), DataSharingNotification::domain_name());

            uint32_t segment_size = static_cast<uint32_t>(sizeof(Notification)) + per_allocation_extra_size;

            //Open the segment

            T::remove(segment_name_);

            local_segment.reset(

                new T(boost::interprocess::create_only,

                segment_name_,

                segment_size + T::EXTRA_SEGMENT_SIZE));

        }

        catch (const std::exception& e)

        {

            EPROSIMA_LOG_ERROR(HISTORY_DATASHARING_LISTENER, "Failed to create segment " << segment_name_

                                                                                         << ": " << e.what());

            return false;

        }

        try

        {

            // Alloc and initialize the Node

            notification_ = local_segment->get().template construct<Notification>("notification_node")();

            notification_->new_data.store(false);

        }

        catch (std::exception& e)

        {

            T::remove(segment_name_);

            EPROSIMA_LOG_ERROR(HISTORY_DATASHARING_LISTENER, "Failed to create listener queue " << segment_name_

                                                                                                << ": " << e.what());

            return false;

        }

        segment_ = std::move(local_segment);

        owned_ = true;

        return true;

    }

Unexecuted instantiation: bool eprosima::fastdds::rtps::DataSharingNotification::create_and_init_shared_segment_notification<eprosima::fastdds::rtps::SharedSegment<boost::interprocess::basic_managed_shared_memory<char, boost::interprocess::rbtree_best_fit<boost::interprocess::mutex_family, boost::interprocess::offset_ptr<void, unsigned int, unsigned long, 0ul>, 0ul>, boost::interprocess::iset_index>, boost::interprocess::shared_memory_object> >(eprosima::fastdds::rtps::GUID_t const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&)

Unexecuted instantiation: bool eprosima::fastdds::rtps::DataSharingNotification::create_and_init_shared_segment_notification<eprosima::fastdds::rtps::SharedSegment<boost::interprocess::basic_managed_mapped_file<char, boost::interprocess::rbtree_best_fit<boost::interprocess::mutex_family, boost::interprocess::offset_ptr<void, unsigned int, unsigned long, 0ul>, 0ul>, boost::interprocess::iset_index>, boost::interprocess::file_mapping> >(eprosima::fastdds::rtps::GUID_t const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&)

    template <typename T>

    bool open_and_init_shared_segment_notification(

            const GUID_t& reader_guid,

            const std::string& shared_dir)

    {

        segment_id_ = reader_guid;

        segment_name_ = generate_segment_name(shared_dir, reader_guid);

        //Open the segment

        std::unique_ptr<T> local_segment;

        try

        {

            local_segment = std::unique_ptr<T>(

                new T(boost::interprocess::open_only,

                segment_name_.c_str()));

        }

        catch (const std::exception& e)

        {

            EPROSIMA_LOG_ERROR(HISTORY_DATASHARING_LISTENER, "Failed to open segment " << segment_name_

                                                                                       << ": " << e.what());

            return false;

        }

        // Initialize values from the segment

        notification_ = (local_segment->get().template find<Notification>(

                    "notification_node")).first;

        if (!notification_)

        {

            local_segment.reset();

            EPROSIMA_LOG_ERROR(HISTORY_DATASHARING_LISTENER, "Failed to open listener queue " << segment_name_);

            return false;

        }

        segment_ = std::move(local_segment);

        return true;

    }

Unexecuted instantiation: bool eprosima::fastdds::rtps::DataSharingNotification::open_and_init_shared_segment_notification<eprosima::fastdds::rtps::SharedSegment<boost::interprocess::basic_managed_shared_memory<char, boost::interprocess::rbtree_best_fit<boost::interprocess::mutex_family, boost::interprocess::offset_ptr<void, unsigned int, unsigned long, 0ul>, 0ul>, boost::interprocess::iset_index>, boost::interprocess::shared_memory_object> >(eprosima::fastdds::rtps::GUID_t const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&)

Unexecuted instantiation: bool eprosima::fastdds::rtps::DataSharingNotification::open_and_init_shared_segment_notification<eprosima::fastdds::rtps::SharedSegment<boost::interprocess::basic_managed_mapped_file<char, boost::interprocess::rbtree_best_fit<boost::interprocess::mutex_family, boost::interprocess::offset_ptr<void, unsigned int, unsigned long, 0ul>, 0ul>, boost::interprocess::iset_index>, boost::interprocess::file_mapping> >(eprosima::fastdds::rtps::GUID_t const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&)

    GUID_t segment_id_;         //< The ID of the segment is the GUID of the reader

    std::string segment_name_;  //< Segment name

    std::unique_ptr<Segment> segment_;  //< Shared memory segment

    Notification* notification_;        //< The notification data

    bool owned_ = false;                //< Whether the shared segment is owned by this instance

};

}  // namespace rtps

}  // namespace fastdds

}  // namespace eprosima

#endif  // RTPS_DATASHARING_DATASHARINGNOTIFICATION_HPP