/*
 *                         Vortex OpenSplice
 *
 *   This software and documentation are Copyright 2006 to TO_YEAR ADLINK
 *   Technology Limited, its affiliated companies and licensors. All rights
 *   reserved.
 *
 *   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.
 *
 */

#include "dds_dcps.h"
#include "sac_common.h"
#include "sac_objManag.h"
#include "sac_object.h"
#include "sac_entity.h"
#include "sac_subscriber.h"
#include "sac_contentFilteredTopic.h"
#include "sac_dataReader.h"
#include "sac_domainParticipant.h"
#include "u_observable.h"
#include "u_subscriber.h"
#include "v_status.h"
#include "sac_report.h"

#define DDS_SubscriberClaim(_this, subscriber) \
        DDS_Object_claim(DDS_Object(_this), DDS_SUBSCRIBER, (_Object *)subscriber)

#define DDS_SubscriberClaimRead(_this, subscriber) \
        DDS_Object_claim(DDS_Object(_this), DDS_SUBSCRIBER, (_Object *)subscriber)

#define DDS_SubscriberRelease(_this) \
        DDS_Object_release(DDS_Object(_this))

#define DDS_SubscriberCheck(_this, subscriber) \
        DDS_Object_check_and_assign(DDS_Object(_this), DDS_SUBSCRIBER, (_Object *)subscriber)

#define _Subscriber_get_user_entity(_this) \
        u_subscriber(_Entity_get_user_entity(_Entity(_Subscriber(_this))))

#define SUBSCRIBER_STATUS_MASK (DDS_DATA_ON_READERS_STATUS            |\
                                DDS_SAMPLE_REJECTED_STATUS            |\
                                DDS_LIVELINESS_CHANGED_STATUS         |\
                                DDS_REQUESTED_DEADLINE_MISSED_STATUS  |\
                                DDS_REQUESTED_INCOMPATIBLE_QOS_STATUS |\
                                DDS_DATA_AVAILABLE_STATUS             |\
                                DDS_SAMPLE_LOST_STATUS                |\
                                DDS_SUBSCRIPTION_MATCHED_STATUS)

static DDS_ReturnCode_t
_Subscriber_deinit (
    _Object _this)
{
    DDS_ReturnCode_t result = DDS_RETCODE_OK;
    _Subscriber sub;

    sub = _Subscriber(_this);
    if (sub != NULL) {
        if (c_iterLength(sub->readerList) != 0) {
            result = DDS_RETCODE_PRECONDITION_NOT_MET;
            SAC_REPORT(result, "Subscribers has %d DataReaders",
                        c_iterLength(sub->readerList));
        }
    } else {
        result = DDS_RETCODE_BAD_PARAMETER;
        SAC_REPORT(result, "Subscriber = NULL");
    }
    if (result == DDS_RETCODE_OK) {
        DDS_Entity_set_listener_interest(DDS_Entity(sub), 0);
        DDS_Entity_disable_callbacks(DDS_Entity(sub));
        DDS_free(sub->defaultDataReaderQos);
        c_iterFree(sub->readerList);
        result = _Entity_deinit(_this);
    }
    return result;
}

DDS_Subscriber
DDS_SubscriberNew (
    DDS_DomainParticipant participant,
    const DDS_char *name,
    const DDS_SubscriberQos *qos)
{
    DDS_ReturnCode_t result = DDS_RETCODE_OK;
    _DomainParticipant _participant = NULL;
    _Subscriber _this = NULL;
    u_participant uParticipant;
    u_subscriber uSubscriber;
    u_subscriberQos sQos;
    cmn_listenerDispatcher dispatcher;

    assert (qos != NULL && qos != DDS_SUBSCRIBER_QOS_DEFAULT);

    sQos = DDS_SubscriberQos_copyIn(qos);
    if (sQos == NULL) {
        result = DDS_RETCODE_ERROR;
    }
    if (result == DDS_RETCODE_OK) {
        uParticipant = u_participant(_Entity_get_user_entity(participant));
        _participant = _DomainParticipant(participant);
        if (uParticipant != NULL && _participant != NULL) {
            uSubscriber = u_subscriberNew(uParticipant, name, sQos);
            if (uSubscriber != NULL) {
                result = DDS_Object_new(DDS_SUBSCRIBER, _Subscriber_deinit, (_Object *)&_this);
                if (result == DDS_RETCODE_OK) {
                    result = DDS_Entity_init(_this, u_entity(uSubscriber));
                    DDS_Object_set_domain_id(_Object(_this), DDS_Object_get_domain_id(participant));
                }
                if (result == DDS_RETCODE_OK) {
                    dispatcher = DDS_Entity_get_listenerDispatcher(_participant);
                    result = DDS_Entity_set_listenerDispatcher(_this, dispatcher);
                }
                if (result == DDS_RETCODE_OK) {
                    _this->defaultDataReaderQos = DDS_DataReaderQos__alloc();
                    if (_this->defaultDataReaderQos != NULL) {
                        result = DDS_DataReaderQos_init(
                            _this->defaultDataReaderQos, DDS_DATAREADER_QOS_DEFAULT);
                    } else {
                        result = DDS_RETCODE_OUT_OF_RESOURCES;
                    }
                    _this->readerList = NULL;
                    _this->participant = participant;
                    _this->factoryAutoEnable = sQos->entityFactory.v.autoenable_created_entities;
                }
            } else {
                result = DDS_RETCODE_OUT_OF_RESOURCES;
            }
        } else {
            result = DDS_RETCODE_BAD_PARAMETER;
        }
        u_subscriberQosFree(sQos);
    }
    return (DDS_Subscriber)_this;
}

static DDS_ReturnCode_t
_Subscriber_delete_contained_entities (
    _Subscriber _this)
{
    DDS_DataReader reader;
    DDS_ReturnCode_t result, endResult = DDS_RETCODE_OK;
    c_ulong i, nrReaders;

    /* Make sure we attempt to delete each entity only once:
     * entities that are not ready to be deleted should be
     * inserted back into the list, but should not be encountered
     * again during this iteration. So iterate by the number
     * of elements instead of by taking until the list is empty.
     */
    nrReaders = c_iterLength(_this->readerList);
    for (i = 0; i < nrReaders; i++) {
        reader = DDS_DataReader(c_iterTakeFirst(_this->readerList));
        result = DDS_DataReaderFree(reader);
        if (result != DDS_RETCODE_OK) {
            c_iterInsert(_this->readerList, reader);
            endResult = result;
        }
    }
    return endResult;
}

DDS_ReturnCode_t
DDS_SubscriberFree (
    DDS_Subscriber _this)
{
    DDS_ReturnCode_t result = DDS_RETCODE_OK;
    _Subscriber sub;

    if (_this) {
        result = DDS_SubscriberClaim(_this, &sub);
        if (result == DDS_RETCODE_OK) {
            result = _Subscriber_delete_contained_entities(sub);
            DDS_SubscriberRelease(_this);
        }
        if (result == DDS_RETCODE_OK) {
            result = DDS__free(_this);
        }
    }
    return result;
}

/*     DataReader
 *     create_datareader(
 *         in TopicDescription a_topic,
 *         in DataReaderQos qos,
 *         in DataReaderListener a_listener);
 */
DDS_DataReader
DDS_Subscriber_create_datareader (
    DDS_Subscriber _this,
    const DDS_TopicDescription a_topic,
    const DDS_DataReaderQos *qos,
    const struct DDS_DataReaderListener *a_listener,
    const DDS_StatusMask mask)
{
    DDS_ReturnCode_t result = DDS_RETCODE_OK;
    DDS_DataReader reader = NULL;
    DDS_char *topicName;
    DDS_char *name;
    _Subscriber sub;
    size_t length;
    DDS_boolean factoryAutoEnable = FALSE;
    DDS_DataReaderQos readerQos;

    SAC_REPORT_STACK();

    memset(&readerQos, 0, sizeof(DDS_DataReaderQos));
    (void)DDS_DataReaderQos_init(&readerQos, DDS_DATAREADER_QOS_DEFAULT);

    if (a_topic == NULL) {
        result = DDS_RETCODE_BAD_PARAMETER;
        SAC_REPORT(result, "Topic = NULL");
    } else {
        result = DDS_DataReaderQos_is_consistent(qos);
    }

    if (result == DDS_RETCODE_OK) {
        result = DDS_SubscriberClaim(_this, &sub);
    }
    if (result == DDS_RETCODE_OK) {
        if (qos == DDS_DATAREADER_QOS_DEFAULT) {
            qos = sub->defaultDataReaderQos;
        } else if (qos == DDS_DATAREADER_QOS_USE_TOPIC_QOS) {
            result = DDS_DataReaderQos_init(
                &readerQos, sub->defaultDataReaderQos);
            if (result == DDS_RETCODE_OK) {
                result = DDS_Subscriber_copy_from_topicdescription(
                    _this, &readerQos, a_topic);
            }
            if (result == DDS_RETCODE_OK) {
                result = DDS_DataReaderQos_is_consistent(&readerQos);
            }
            qos = &readerQos;
        }

        if (result == DDS_RETCODE_OK) {
            length = strlen("reader <>") + 1;
            topicName = DDS_Topic_get_name(a_topic);
            if (topicName) {
                length += strlen(topicName);
                name = os_malloc(length);
                snprintf (name, length, "reader <%s>", topicName);
                DDS_free(topicName);
            } else {
                name = os_malloc (length);
                os_strncpy (name, "reader", length - 1);
            }

            reader = DDS_DataReaderNew (_this, name, qos, DDS_TopicDescription(a_topic));
            os_free(name);
            if ( reader ) {
                sub->readerList = c_iterInsert (sub->readerList, reader);
            } else {
                result = DDS_RETCODE_ERROR;
            }
            factoryAutoEnable = sub->factoryAutoEnable;
        }
        DDS_SubscriberRelease(_this);
    }
    if (result == DDS_RETCODE_OK) {
        cmn_listenerDispatcher listenerDispatcher;

        listenerDispatcher = DDS_Entity_get_listenerDispatcher(_this);
        result = DDS_Entity_set_listenerDispatcher(reader, listenerDispatcher);
    }
    if (result == DDS_RETCODE_OK) {
        DDS_DataReader_set_listener(reader, a_listener, mask);
        if (factoryAutoEnable && _Entity_is_enabled(_Entity(sub))) {
            result = DDS_Entity_enable(reader);
            if (result != DDS_RETCODE_OK) {
                DDS_ReturnCode_t destroyed;
                destroyed = DDS_Subscriber_delete_datareader (_this, reader);
                assert (destroyed == DDS_RETCODE_OK);
                OS_UNUSED_ARG(destroyed);
                reader = NULL;
            }
        }
    }

    (void)DDS_DataReaderQos_deinit(&readerQos);

    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return reader;
}

/*     ReturnCode_t
 *     delete_datareader(
 *         in DataReader a_datareader);
 */
DDS_ReturnCode_t
DDS_Subscriber_delete_datareader (
    DDS_Subscriber _this,
    const DDS_DataReader reader)
{
    DDS_ReturnCode_t result;
    DDS_DataReader found;
    _Subscriber sub;

    SAC_REPORT_STACK();

    if (reader != NULL) {
        result = DDS_SubscriberClaim(_this, &sub);
    } else {
        result = DDS_RETCODE_BAD_PARAMETER;
        SAC_REPORT(result, "DataReader = NULL");
    }
    if (result == DDS_RETCODE_OK) {
        found = c_iterTake(sub->readerList, reader);
        if (found != reader) {
            /* The following call is expensive so only use it in case of exceptions. */
            if (DDS_Object_get_kind(DDS_Object(reader)) == DDS_DATAREADER) {
                result = DDS_RETCODE_PRECONDITION_NOT_MET;
                SAC_REPORT(result, "DataReader is not from Subscriber");
            } else {
                result = DDS_RETCODE_BAD_PARAMETER;
                SAC_REPORT(result, "DataReader parameter 'reader' is of type %s",
                            DDS_ObjectKind_image(DDS_Object_get_kind(DDS_Object(reader))));
            }
        } else {
            result = DDS__free(reader);
        }
        if (result != DDS_RETCODE_OK) {
           (void)c_iterInsert(sub->readerList, found);
        }
        DDS_SubscriberRelease(_this);
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

struct check_handle_arg {
    DDS_InstanceHandle_t handle;
    DDS_boolean result;
};

static c_bool
reader_check_handle(
    void *object,
    struct check_handle_arg *arg)
{
    assert(object);
    assert(arg);

    if (!arg->result) {
        arg->result = DDS_Entity_check_handle(DDS_Entity(object), arg->handle);
        if ( !arg->result ) {
            arg->result = DDS_DataReader_contains_entity(DDS_DataReader(object), arg->handle);
        }
    }
    return !arg->result;
}

DDS_boolean
DDS_Subscriber_contains_entity (
    DDS_Subscriber _this,
    DDS_InstanceHandle_t  a_handle)
{
    DDS_ReturnCode_t result;
    _Subscriber sub;
    struct check_handle_arg arg;

    SAC_REPORT_STACK();

    arg.handle = a_handle;
    arg.result = FALSE;
    result = DDS_SubscriberClaimRead(_this, &sub);
    if (result == DDS_RETCODE_OK) {
        if (!arg.result) {
            c_iterWalkUntil(sub->readerList, (c_iterAction)reader_check_handle, &arg);
        }
        result = DDS_SubscriberRelease(_this);
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return arg.result;
}

/*     ReturnCode_t
 *     delete_contained_entities();
 */
DDS_ReturnCode_t
DDS_Subscriber_delete_contained_entities (
    DDS_Subscriber _this)
{
    DDS_ReturnCode_t result;
    _Subscriber sub;

    SAC_REPORT_STACK();

    result = DDS_SubscriberClaim(_this, &sub);
    if (result == DDS_RETCODE_OK) {
        result = _Subscriber_delete_contained_entities(sub);
        DDS_SubscriberRelease(_this);
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

struct lookupByTopicArg {
    const DDS_char *topicName;
    DDS_DataReader reader;
};

static c_equality
lookupByTopic (
    c_voidp o,
    c_voidp arg)
{
    DDS_DataReader reader = (DDS_DataReader)o;
    struct lookupByTopicArg *a = (struct lookupByTopicArg *)arg;
    DDS_TopicDescription topic;
    DDS_char *topicName;
    c_equality equality = C_NE;

    topic = DDS_DataReader_get_topicdescription(reader);
    topicName = DDS_TopicDescription_get_name(topic);
    if (strcmp(topicName, a->topicName) == 0) {
        a->reader = reader;
        equality = C_EQ;
    }
    DDS_free(topicName);

    return equality;
}

/*     DataReader
 *     lookup_datareader(
 *         in string topic_name);
 */
DDS_DataReader
DDS_Subscriber_lookup_datareader (
    DDS_Subscriber _this,
    const DDS_char *topic_name)
{
    DDS_ReturnCode_t result;
    DDS_DataReader found = NULL;
    _Subscriber sub;
    struct lookupByTopicArg arg;
    os_char *name = NULL;

    SAC_REPORT_STACK();

    result = DDS_SubscriberClaimRead(_this, &sub);
    if (result == DDS_RETCODE_OK) {
        arg.topicName = topic_name;
        arg.reader = NULL;
        c_iterResolve(sub->readerList, lookupByTopic, &arg);
        found = arg.reader;
        if (found == NULL) {
            name = u_entityName(u_entity(_Subscriber_get_user_entity(sub)));
        }
        DDS_SubscriberRelease(_this);
    }
    if (name) {
        if (strcmp(name, "BuiltinSubscriber") == 0) {
            DDS_DomainParticipant participant;
            DDS_Topic topic;
            DDS_DataReaderQos *rQos = DDS_DataReaderQos__alloc();

            if (rQos) {
                memset(rQos, 0, sizeof(DDS_DataReaderQos));
                result = DDS_Subscriber_get_default_datareader_qos (_this, rQos);
                rQos->durability.kind = DDS_TRANSIENT_DURABILITY_QOS;
                rQos->reliability.kind = DDS_RELIABLE_RELIABILITY_QOS;

                participant = DDS_Subscriber_get_participant(_this);
                topic = DDS_DomainParticipant_lookup_builtin_topic(participant, topic_name);
                found = DDS_Subscriber_create_datareader(_this, topic, rQos, NULL, 0);

                DDS_free(rQos);
            } else {
                result = DDS_RETCODE_OUT_OF_RESOURCES;
                SAC_REPORT(result, "Failed to allocate DDS_DataReaderQos");
            }
        }
        os_free(name);
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return found;
}

/*     ReturnCode_t
 *     get_datareaders(
 *         inout DataReaderSeq readers,
 *         in SampleStateMask sample_states,
 *         in ViewStateMask view_states,
 *         in InstanceStateMask instance_states);
 */
DDS_ReturnCode_t
DDS_Subscriber_get_datareaders (
    DDS_Subscriber _this,
    DDS_DataReaderSeq *readers,
    const DDS_SampleStateMask sample_states,
    const DDS_ViewStateMask view_states,
    const DDS_InstanceStateMask instance_states)
{
    DDS_ReturnCode_t result;
    _Subscriber sub;
    u_dataReader uReader;
    u_result uResult;
    u_sampleMask mask;
    c_iter list = NULL;
    c_ulong length;

    SAC_REPORT_STACK();

    result = DDS_SAMPLE_MASK_IS_VALID(sample_states, view_states, instance_states);
    if (result == DDS_RETCODE_OK) {
        mask = DDS_SAMPLE_MASK(sample_states, view_states, instance_states);
        if (!DDS_sequence_is_valid((_DDS_sequence)readers)) {
            result = DDS_RETCODE_BAD_PARAMETER;
        } else {
            result = DDS_SubscriberClaimRead(_this, &sub);
        }
    } else {
        SAC_REPORT(result, "Invalid mask, sample_states = 0x%x, view_states = 0x%x, instance_states = 0x%x",
                    sample_states, view_states, instance_states);
    }

    if (result == DDS_RETCODE_OK) {
        uResult = u_subscriberGetDataReaders(_Subscriber_get_user_entity(sub), mask, &list);
        if (uResult == U_RESULT_OK) {
            length = c_iterLength(list);
            if (length > readers->_maximum) {
                if (readers->_release == TRUE) {
                    DDS_free(readers->_buffer);
                }
                readers->_buffer = DDS_DataReaderSeq_allocbuf(length);
                readers->_maximum = length;
                readers->_release = TRUE;
            }
            readers->_length = 0;
            while ((uReader = u_dataReader(c_iterTakeFirst(list))) != NULL) {
                readers->_buffer[readers->_length] = u_observableGetUserData(u_observable(uReader));
                readers->_length++;
            }
            c_iterFree(list);
            result = DDS_RETCODE_OK;
        } else {
            result = DDS_ReturnCode_get(uResult);
        }
        DDS_SubscriberRelease(_this);
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

static void
notifyReader(DDS_Object reader, void *arg)
{
    DDS_DataReader dataReader = (DDS_DataReader) reader;
    OS_UNUSED_ARG(arg);
    if (DDS_DataReader_get_status_changes(reader) & DDS_DATA_AVAILABLE_STATUS) {
        struct DDS_DataReaderListener drListener;

        drListener = DDS_DataReader_get_listener(reader);
        if (drListener.on_data_available != NULL) {
            drListener.on_data_available(drListener.listener_data, dataReader);
        }
    }
}

/*     ReturnCode_t
 *     notify_datareaders();
 */
DDS_ReturnCode_t
DDS_Subscriber_notify_datareaders (
    DDS_Subscriber _this)
{
    DDS_ReturnCode_t result;
    _Subscriber s;

    SAC_REPORT_STACK();

    result = DDS_SubscriberClaim(_this, &s);
    if (result == DDS_RETCODE_OK) {
        c_iterWalk(s->readerList, notifyReader, NULL);
        DDS_SubscriberRelease(_this);
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     ReturnCode_t
 *     set_qos(
 *         in SubscriberQos qos);
 */
DDS_ReturnCode_t
DDS_Subscriber_set_qos (
    DDS_Subscriber _this,
    const DDS_SubscriberQos *qos)
{
    DDS_ReturnCode_t result = DDS_RETCODE_OK;
    DDS_SubscriberQos subscriberQos;
    _Subscriber s;
    u_subscriberQos sQos = NULL;
    u_result uResult;
    u_subscriber uSubscriber;

    SAC_REPORT_STACK();

    memset(&subscriberQos, 0, sizeof(DDS_SubscriberQos));
    (void)DDS_SubscriberQos_init(&subscriberQos, DDS_SUBSCRIBER_QOS_DEFAULT);

    result = DDS_SubscriberQos_is_consistent(qos);
    if (result == DDS_RETCODE_OK) {
        result = DDS_SubscriberClaim(_this, &s);
    }
    if (result == DDS_RETCODE_OK) {
        if (qos == DDS_SUBSCRIBER_QOS_DEFAULT) {
            result = DDS_DomainParticipant_get_default_subscriber_qos(
                s->participant, &subscriberQos);
            qos = &subscriberQos;
        }
        if (result == DDS_RETCODE_OK) {
            sQos = DDS_SubscriberQos_copyIn(qos);
            if (sQos == NULL) {
                result = DDS_RETCODE_OUT_OF_RESOURCES;
                SAC_REPORT(result, "Failed to copy DDS_SubscriberQos");
            }
        }
        if (result == DDS_RETCODE_OK) {
            uSubscriber = _Subscriber_get_user_entity(s);
            uResult = u_subscriberSetQos(uSubscriber, sQos);
            result = DDS_ReturnCode_get(uResult);
            if (result == DDS_RETCODE_OK) {
                s->factoryAutoEnable = sQos->entityFactory.v.autoenable_created_entities;
            }
            u_subscriberQosFree(sQos);
        }
        DDS_SubscriberRelease(_this);
    }

    (void)DDS_SubscriberQos_deinit(&subscriberQos);

    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     ReturnCode_t
 *     get_qos(
 *         inout SubscriberQos qos);
 */
DDS_ReturnCode_t
DDS_Subscriber_get_qos (
    DDS_Subscriber _this,
    DDS_SubscriberQos *qos)
{
    DDS_ReturnCode_t result;
    _Subscriber s;
    u_subscriberQos uQos;
    u_result uResult;

    SAC_REPORT_STACK();

    result = DDS_SubscriberCheck(_this, &s);
    if (result == DDS_RETCODE_OK) {
        if (qos == NULL) {
            result = DDS_RETCODE_BAD_PARAMETER;
            SAC_REPORT(result, "SubscriberQos = NULL");
        } else if (qos == DDS_SUBSCRIBER_QOS_DEFAULT) {
            result = DDS_RETCODE_BAD_PARAMETER;
            SAC_REPORT(result, "QoS 'SUBSCRIBER_QOS_DEFAULT' is read-only.");
        }
    }
    if (result == DDS_RETCODE_OK) {
        uResult = u_subscriberGetQos(_Subscriber_get_user_entity(s), &uQos);
        if (uResult == U_RESULT_OK) {
            result = DDS_SubscriberQos_copyOut(uQos, qos);
            u_subscriberQosFree(uQos);
        } else {
            result = DDS_ReturnCode_get(uResult);
        }
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     ReturnCode_t
 *     set_listener(
 *         in SubscriberListener a_listener,
 *         in StatusKindMask mask);
 */
DDS_ReturnCode_t
DDS_Subscriber_set_listener (
    DDS_Subscriber _this,
    const struct DDS_SubscriberListener *a_listener,
    const DDS_StatusMask mask)
{
    DDS_ReturnCode_t result;
    _Subscriber s;

    SAC_REPORT_STACK();

    result = DDS_SubscriberClaim(_this, &s);
    if (result == DDS_RETCODE_OK) {
        if (a_listener != NULL) {
            s->listener = *a_listener;
            result = DDS_Entity_set_listener_interest(DDS_Entity(s), mask);
        } else {
            memset(&s->listener, 0, sizeof(struct DDS_SubscriberListener));
            result = DDS_Entity_set_listener_interest(DDS_Entity(s), mask);
        }
        DDS_SubscriberRelease(_this);
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     SubscriberListener
 *     get_listener();
 */
struct DDS_SubscriberListener
DDS_Subscriber_get_listener (
    DDS_Subscriber _this)
{
    DDS_ReturnCode_t result;
    _Subscriber s;
    struct DDS_SubscriberListener listener;

    SAC_REPORT_STACK();

    result = DDS_SubscriberCheck(_this, &s);
    if (result == DDS_RETCODE_OK) {
        listener = s->listener;
    } else {
        memset(&listener, 0, sizeof(struct DDS_SubscriberListener));
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return listener;
}

/*     ReturnCode_t
 *     set_listener_mask(
 *         in StatusKindMask mask);
 */
DDS_ReturnCode_t
DDS_Subscriber_set_listener_mask (
    _Subscriber _this,
    const DDS_StatusMask mask)
{
    DDS_ReturnCode_t result;

    assert(_this);

    result = DDS_Entity_set_listener_interest(DDS_Entity(_this), mask);

    return result;
}

/*     ReturnCode_t
 *     begin_access();
 */
DDS_ReturnCode_t
DDS_Subscriber_begin_access (
    DDS_Subscriber _this)
{
    DDS_ReturnCode_t result;
    _Subscriber s;
    u_result uResult;

    SAC_REPORT_STACK();

    result = DDS_SubscriberCheck(_this, &s);
    if (result == DDS_RETCODE_OK) {
        uResult = u_subscriberBeginAccess(_Subscriber_get_user_entity(s));
        if (uResult != U_RESULT_OK) {
            result = DDS_ReturnCode_get(uResult);
        }
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     ReturnCode_t
 *     end_access();
 */
DDS_ReturnCode_t
DDS_Subscriber_end_access (
    DDS_Subscriber _this)
{
    DDS_ReturnCode_t result;
    _Subscriber s;
    u_result uResult;

    SAC_REPORT_STACK();

    result = DDS_SubscriberCheck(_this, &s);
    if (result == DDS_RETCODE_OK) {
        uResult = u_subscriberEndAccess(_Subscriber_get_user_entity(s));
        if (uResult != U_RESULT_OK) {
            result = DDS_ReturnCode_get(uResult);
        }
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     DomainParticipant
 *     get_participant();
 */
DDS_DomainParticipant
DDS_Subscriber_get_participant (
    DDS_Subscriber _this)
{
    DDS_ReturnCode_t result;
    _Subscriber s;
    DDS_DomainParticipant participant = NULL;

    SAC_REPORT_STACK();

    result = DDS_SubscriberCheck(_this, &s);
    if (result == DDS_RETCODE_OK) {
        participant = s->participant;
    }
    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return participant;
}

/*     ReturnCode_t
 *     set_default_datareader_qos(
 *         in DataReaderQos qos);
 */
DDS_ReturnCode_t
DDS_Subscriber_set_default_datareader_qos (
    DDS_Subscriber _this,
    const DDS_DataReaderQos *qos)
{
    DDS_ReturnCode_t result = DDS_RETCODE_OK;
    _Subscriber s;
    DDS_DataReaderQos *readerQos = NULL;

    SAC_REPORT_STACK();

    if (qos == DDS_DATAREADER_QOS_USE_TOPIC_QOS) {
        result = DDS_RETCODE_BAD_PARAMETER;
        SAC_REPORT(result, "DataReaderQos = DDS_DATAREADER_QOS_USE_TOPIC_QOS");
    } else {
        result = DDS_DataReaderQos_is_consistent(qos);
    }
    if (result == DDS_RETCODE_OK) {
        readerQos = DDS_DataReaderQos__alloc();
        if (readerQos != NULL) {
            result = DDS_DataReaderQos_init(readerQos, qos);
        } else {
            result = DDS_RETCODE_OUT_OF_RESOURCES;
            SAC_REPORT(result, "Failed to copy DDS_DataReaderQos");
        }
    }
    if (result == DDS_RETCODE_OK) {
        result = DDS_SubscriberClaim(_this, &s);
    }
    if (result == DDS_RETCODE_OK) {
        DDS_free(s->defaultDataReaderQos);
        s->defaultDataReaderQos = readerQos;
        DDS_SubscriberRelease(_this);
    } else {
        DDS_free(readerQos);
    }

    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     ReturnCode_t
 *     get_default_datareader_qos(
 *         inout DataReaderQos qos);
 */
DDS_ReturnCode_t
DDS_Subscriber_get_default_datareader_qos (
    DDS_Subscriber _this,
    DDS_DataReaderQos *qos)
{
    DDS_ReturnCode_t result;
    _Subscriber s;

    SAC_REPORT_STACK();

    result = DDS_SubscriberClaimRead(_this, &s);
    if (result == DDS_RETCODE_OK) {
        result = DDS_DataReaderQos_init(qos, s->defaultDataReaderQos);
        DDS_SubscriberRelease(_this);
    }

    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

/*     ReturnCode_t
 *     copy_from_topic_qos(
 *         inout DataReaderQos a_datareader_qos,
 *         in TopicQos a_topic_qos);
 */
DDS_ReturnCode_t
DDS_Subscriber_copy_from_topic_qos (
    DDS_Subscriber _this,
    DDS_DataReaderQos *rQos,
    const DDS_TopicQos *tQos)
{
    DDS_ReturnCode_t result;
    DDS_TopicQos *topicQos = NULL;
    _Subscriber s;

    SAC_REPORT_STACK();

    result = DDS_SubscriberCheck(_this, &s);
    if (result == DDS_RETCODE_OK) {
        if (rQos == NULL) {
            result = DDS_RETCODE_BAD_PARAMETER;
            SAC_REPORT(result, "a_datareader_qos 'null' is invalid.");
        } else if (rQos == DDS_DATAREADER_QOS_DEFAULT) {
            result = DDS_RETCODE_BAD_PARAMETER;
            SAC_REPORT(result, "QoS 'DATAREADER_QOS_DEFAULT' is read-only.");
        } else if (rQos == DDS_DATAREADER_QOS_USE_TOPIC_QOS) {
            result = DDS_RETCODE_BAD_PARAMETER;
            SAC_REPORT(result, "QoS 'DATAREADER_QOS_USE_TOPIC_QOS' is read-only.");
        } else if (tQos == NULL) {
            result = DDS_RETCODE_BAD_PARAMETER;
            SAC_REPORT(result, "a_topic_qos 'null' is invalid.");
        } else if (tQos == DDS_TOPIC_QOS_DEFAULT) {
            topicQos = DDS_TopicQos__alloc();
            if (topicQos == NULL) {
                result = DDS_RETCODE_ERROR;
                SAC_REPORT(result, "Could not copy DataReaderQos.");
            } else {
                result = DDS_DomainParticipant_get_default_topic_qos(
                    s->participant, topicQos);
                if (result == DDS_RETCODE_OK) {
                    tQos = topicQos;
                }
            }
        }

        if (result == DDS_RETCODE_OK) {
            rQos->durability         = tQos->durability;
            rQos->deadline           = tQos->deadline;
            rQos->latency_budget     = tQos->latency_budget;
            rQos->liveliness         = tQos->liveliness;
            rQos->reliability        = tQos->reliability;
            rQos->destination_order  = tQos->destination_order;
            rQos->history            = tQos->history;
            rQos->ownership          = tQos->ownership;
            rQos->resource_limits    = tQos->resource_limits;
        }
    }

    if (topicQos != NULL) {
        (void)DDS__free(topicQos);
    }

    SAC_REPORT_FLUSH(_this, result != DDS_RETCODE_OK);
    return result;
}

DDS_ReturnCode_t
DDS_Subscriber_copy_from_topicdescription (
    DDS_Subscriber _this,
    DDS_DataReaderQos *qos,
    const DDS_TopicDescription description)
{
    DDS_ReturnCode_t result;
    DDS_ObjectKind objectKind;
    DDS_Topic topic;
    DDS_TopicQos topicQos;

    assert (_this != NULL);
    assert (qos != NULL);
    assert (description != NULL);

    /* initialize to safely deinitialize on exit */
    memset(&topicQos, 0, sizeof(DDS_TopicQos));
    (void)DDS_TopicQos_init(&topicQos, DDS_TOPIC_QOS_DEFAULT);

    objectKind = DDS_Object_get_kind(DDS_Object(description));
    switch (objectKind) {
        case DDS_CONTENTFILTEREDTOPIC:
            topic = DDS_ContentFilteredTopic_get_related_topic(description);
            break;
        default:
            assert (objectKind == DDS_TOPIC);
            topic = DDS_Topic(description);
            break;
    }

    assert (topic != NULL);

    result = DDS_Topic_get_qos(topic, &topicQos);
    if (result == DDS_RETCODE_OK) {
        result = DDS_Subscriber_copy_from_topic_qos(_this, qos, &topicQos);
    }

    (void)DDS_TopicQos_deinit(&topicQos);

    return result;
}

DDS_ReturnCode_t
DDS_Subscriber_notify_listener (
    DDS_Subscriber _this,
    v_listenerEvent event)
{
    DDS_ReturnCode_t result;
    u_eventMask triggerMask;
    DDS_Entity entity;
    struct DDS_SubscriberListener cb;

    cb = _Subscriber(_this)->listener;
    triggerMask = event->kind;
    result = DDS_RETCODE_OK;

    entity = u_observableGetUserData(u_observable(event->source));

    if (triggerMask & V_EVENT_ON_DATA_ON_READERS) {
        if (cb.on_data_on_readers != NULL) {
            result = DDS_Entity_reset_on_data_on_readers_status(DDS_Entity(entity));
            if (result == DDS_RETCODE_OK) {
                cb.on_data_on_readers(cb.listener_data, entity);
            }
        }
    } else {
        if (triggerMask & V_EVENT_DATA_AVAILABLE) {
            if (cb.on_data_available != NULL) {
                result = DDS_Entity_reset_dataAvailable_status(DDS_Entity(entity));
                if (result == DDS_RETCODE_OK) {
                    cb.on_data_available(cb.listener_data, entity);
                }
            }
        }
    }
    if ((triggerMask & V_EVENT_SAMPLE_REJECTED) &&
            (cb.on_sample_rejected != NULL))
    {
        DDS_SampleRejectedStatus status;
        DDS_SampleRejectedStatus_init(&status, &((v_readerStatus)event->eventData)->sampleRejected);
        cb.on_sample_rejected(cb.listener_data, entity, &status);
    }
    if ((triggerMask & V_EVENT_LIVELINESS_CHANGED) &&
            (cb.on_liveliness_changed != NULL))
    {
        DDS_LivelinessChangedStatus status;
        DDS_LivelinessChangedStatus_init(&status, &((v_readerStatus)event->eventData)->livelinessChanged);
        cb.on_liveliness_changed(cb.listener_data, entity, &status);
    }
    if ((triggerMask & V_EVENT_REQUESTED_DEADLINE_MISSED) &&
            (cb.on_requested_deadline_missed != NULL))
    {
        DDS_RequestedDeadlineMissedStatus status;
        DDS_RequestedDeadlineMissedStatus_init(&status, &((v_readerStatus)event->eventData)->deadlineMissed);
        cb.on_requested_deadline_missed(cb.listener_data, entity, &status);
    }
    if ((triggerMask & V_EVENT_REQUESTED_INCOMPATIBLE_QOS) &&
            (cb.on_requested_incompatible_qos != NULL))
    {
        DDS_RequestedIncompatibleQosStatus status;
        DDS_RequestedIncompatibleQosStatus_init(&status, &((v_readerStatus)event->eventData)->incompatibleQos);
        cb.on_requested_incompatible_qos(cb.listener_data, entity, &status);
    }
    if ((triggerMask & V_EVENT_SAMPLE_LOST) &&
            (cb.on_sample_lost != NULL))
    {
        DDS_SampleLostStatus status;
        DDS_SampleLostStatus_init(&status, &((v_readerStatus)event->eventData)->sampleLost);
        cb.on_sample_lost(cb.listener_data, entity, &status);
    }
    if ((triggerMask & V_EVENT_SUBSCRIPTION_MATCHED) &&
            (cb.on_subscription_matched != NULL))
    {
        DDS_SubscriptionMatchedStatus status;
        DDS_SubscriptionMatchedStatus_init(&status, &((v_readerStatus)event->eventData)->subscriptionMatch);
        cb.on_subscription_matched(cb.listener_data, entity, &status);
    }

    return result;
}
