/**************************************************************
 * constructor/destructor
 **************************************************************/
v_service
v_serviceNew(
    v_serviceManager manager,
    const c_char *name,
    const c_char *extStateName,
    v_participantQos qos,
    v_statistics stats)
{
    v_kernel k;
    v_service s;
    v_participantQos q;

    assert(C_TYPECHECK(manager, v_serviceManager));
    /* Do not C_TYPECHECK qos parameter, since it might be allocated on heap! */
    assert(name != NULL);

    k = v_objectKernel(manager);
    /* do no use cast method on qos parameter,
     * it is most likely allocated on heap! */
    q = v_participantQosNew(k, (v_participantQos)qos);
    if (q == NULL) {
        OS_REPORT(OS_ERROR, "v_serviceNew", 0,
                  "Service not created: inconsistent qos");
        s = NULL;
    } else {
        s = v_service(v_objectNew(k, K_SERVICE));
        v_serviceInit(s, manager, name, extStateName, q, stats);
        c_free(q);
        /* always add, even when s->state==NULL, since v_participantFree always
           removes the participant.*/
        v_addParticipant(k, v_participant(s));
        if (s->state == NULL) {
            v_serviceFree(s);
            s = NULL;
        }
    }

    return s;
}

void
v_cfElementInit (
    v_cfElement element,
    v_configuration config,
    const c_char *tagName)
{
    c_type attrType;
    c_type nodeType;
    const c_char *keyList;
    
    assert(C_TYPECHECK(element, v_cfElement));
    assert(tagName != NULL);

    v_cfNodeInit(v_cfNode(element), config, V_CFELEMENT, tagName);

    attrType = c_resolve(c_getBase(element), "kernelModule::v_cfAttribute");
    nodeType = c_resolve(c_getBase(element), "kernelModule::v_cfNode");
    keyList = "name";

    element->attributes = c_tableNew(attrType, keyList);
    element->children = c_setNew(nodeType);
}

/* Precondition: protect the sample yourself */
v_writerSampleStatus
_v_writerSampleGetStatus(
    v_writerSample sample)
{
    v_writerSampleStatus result;

    assert(sample);
    assert(C_TYPECHECK(sample,v_writerSample));

    if ((c_long)sample->resend == TRUE) {
        /* Someone has rejected the sample, resend it */
        result = V_SAMPLE_RESEND;
    } else {
        if ((c_long)sample->decayCount > 0) {
            result = V_SAMPLE_KEEP;
        } else {
            result = V_SAMPLE_RELEASE;
        }
    }

    return result;
}

void
v_listenerFlush(
    v_listener _this,
    v_eventMask events,
    c_voidp userData)
{
    v_listenerEvent event, *prev;

    if (_this == NULL) {
        return;
    }
    assert(C_TYPECHECK(_this,v_listener));
    c_mutexLock(&_this->mutex);
    /* wakeup blocking threads to evaluate new condition. */
    /* remove all events */
    prev = &_this->eventList;
    event = _this->eventList;
    while (event != NULL) {
        if (event->userData == userData) {
            event->kind &= ~events;
        }
        if (event->kind == 0) {
            if (event == _this->lastEvent) {
                _this->lastEvent = c_keep(_this->lastEvent->next);
                v_listenerEventDeinit(event);
                c_free(event);
            }
            *prev = event->next;
            v_listenerEventDeinit(event);
            c_free(event);
            event = *prev;
        } else {
            prev = &event->next;
            event = event->next;
        }
    }
    c_condBroadcast(&_this->cv);
    c_mutexUnlock(&_this->mutex);
}

/**************************************************************
 * constructor/destructor
 **************************************************************/
v_leaseManager
v_leaseManagerNew(
    v_kernel k)
{
    v_leaseManager _this;

    assert(C_TYPECHECK(k, v_kernel));

    _this = v_leaseManager(v_objectNew(k, K_LEASEMANAGER));
    if(_this)
    {
        v_leaseManagerInit(_this);
    } else
    {
        OS_REPORT(OS_ERROR, "v_leaseManager", 0,
            "Failed to create a v_leaseManager object. "
            "Most likely not enough shared memory available "
            "to complete the operation.");
    }

    return _this;
}

/**
 * This method will invalidate a handle and mark the resources as ready for reuse.
 * Note that the info and handle musr correspond and that info is locked.
 */
static void
v_handleInvalidate (
    v_handle handle,
    v_handleInfo *info)
{
    v_handleServer server;
    c_object entity;

    server = v_handleServer((c_object)handle.server);
    assert(C_TYPECHECK(server,v_handleServer));
    if (server) {
        c_mutexLock(&server->mutex);
        info->nextFree = server->firstFree;
        server->firstFree = handle.index;
        info->serial = (info->serial + 1) % MAXSERIAL;
        entity = info->object;
        info->object = NULL;
        c_mutexUnlock(&server->mutex);
        c_mutexUnlock(&info->mutex);
        v_publicDispose(v_public(entity));
    }
}

v_writeResult
v_instanceWrite(
    v_instance instance,
    v_message message)
{
    c_char *metaName;

    assert(C_TYPECHECK(instance, v_instance));

    switch (v_objectKind(instance)) {
    case K_DATAREADERINSTANCE:
        return v_dataReaderInstanceWrite(v_dataReaderInstance(instance),message);
    default:
        metaName = c_metaName(c_metaObject(c_getType(instance)));
        OS_REPORT_1(OS_ERROR,
                    "v_instanceWrite",0,
                    "Unknown instance type <%s>",
                    metaName);
        c_free(metaName);
        return V_WRITE_PRE_NOT_MET;
    }
}

void
v_listenerFree(
   v_listener _this)
{
    v_participant p;
    v_listenerEvent event;
    os_duration delay;

    assert(_this != NULL);
    assert(C_TYPECHECK(_this,v_listener));
    p = v_participant(_this->participant);
    assert(p != NULL);

    c_mutexLock(&_this->mutex);

    /* wakeup blocking threads to evaluate new condition. */
    /* remove all events */
    while (_this->eventList != NULL) {
        event = _this->eventList;
        _this->eventList = event->next;
        v_listenerEventDeinit(event);
        c_free(event);
    }
    _this->eventList = NULL;
    c_free(_this->lastEvent);
    _this->lastEvent = NULL;
    _this->terminate = TRUE;
    c_condBroadcast(&_this->cv);
    c_mutexUnlock(&_this->mutex);

    delay = OS_DURATION_INIT(0, 1000);
    while (_this->waitCount > 0 && !p->processIsZombie) {
        ospl_os_sleep(delay);
    }

    v_participantRemove(p, v_object(_this));
    _this->participant = NULL;
    v_publicFree(v_public(_this));
}

v_topicAdapter
v_topicAdapterNewFromTopicInfo(
    v_participant p,
    const struct v_topicInfo *info,
    c_bool announce)
{
    v_topicAdapter adapter = NULL;
    v_topicImpl topic;
    v_kernel kernel;

    assert(p != NULL);
    assert(C_TYPECHECK(p,v_participant));

    kernel = v_objectKernel(p);

    topic = v_topicImplNewFromTopicInfo(kernel, info, announce);
    if (topic) {
        adapter = v_topicAdapterWrap(p, v_topic(topic));
    }

    return adapter;
}

v_writeResult
v_networkReaderEntryWrite(
    v_networkReaderEntry entry,
    v_message message,
    v_networkId writingNetworkId)
{
    v_writeResult result = V_WRITE_SUCCESS;
    c_bool writeSucceeded;
    static v_gid zeroAddressee = {0,0,0};

    assert(C_TYPECHECK(entry, v_networkReaderEntry));
    assert(message != NULL);

    /* First check if there is any remote interest at all */

    if (v_networkReader(v_entry(entry)->reader)->remoteActivity) {
        /* Only forward messages that come from this kernel */
        if (writingNetworkId == V_NETWORKID_LOCAL || entry->isRouting) {
            /* OK, message is from this kernel or this is a routing entry. Now
             * attach the correct fields if needed */

            /* TODO: For networking services that support routing perhaps
             * messages shouldn't be forwarded to 'self' (e.g., echo cancellation
             * may be needed). For R&R this modus is fine. */
            writeSucceeded = v_networkReaderWrite(
                                  v_networkReader(v_entry(entry)->reader),
                                  message, entry, 0, message->writerGID,
                                  FALSE /* no p2p */, zeroAddressee);
            if (writeSucceeded) {
                result = V_WRITE_SUCCESS;
            } else {
                result = V_WRITE_REJECTED;
            }
        }
    }

    return result;
}

/**************************************************************
 * constructor/destructor
 **************************************************************/
v_cfNode
v_cfNodeNew(
    v_configuration config,
    v_cfKind kind)
{
    v_cfNode node;
    c_type type;
    
    assert(C_TYPECHECK(config, v_configuration));
    
    switch (kind) {
    case V_CFELEMENT:
        type = c_resolve(c_getBase(config), "kernelModule::v_cfElement");
    break;
    case V_CFATTRIBUTE:
        type = c_resolve(c_getBase(config), "kernelModule::v_cfAttribute");
    break;
    case V_CFDATA:
        type = c_resolve(c_getBase(config), "kernelModule::v_cfData");
    break;
    case V_CFNODE:
    default:
        OS_REPORT_1(OS_ERROR,"v_cfNodeNew failed",0,"Illegal kind (%d) specified",kind);
        assert(FALSE); 
        type = NULL;
    break;
    }

    if (type != NULL) {
        node = c_new(type);
    } else {
        node = NULL;
    }
    /* init is done by specific class itself, this is just a 
       convenience function! */

    return node;
}

v_result
v_deliveryWaitListNotify (
    v_deliveryWaitList _this,
    v_deliveryInfoTemplate msg)
{
    c_ulong size, i, count;
    v_gid *list;

    assert(C_TYPECHECK(_this,v_deliveryWaitList));
    assert(msg);

    list = (v_gid *)_this->readerGID;
    if(msg->userData.sequenceNumber == _this->sequenceNumber)
    {
        count = 0;
        
        size = c_arraySize(_this->readerGID);
        for (i=0; i<size; i++) {
            if (v_gidEqual(list[i],msg->userData.readerGID)) {
                /* Set the found readerGID to zero,
                 * iThe waitlist can be unblocked when
                 * all expected systemIds are zero.
                 * In that case count will be 0.
                 */
                v_gidSetNil(list[i]);
            }
            count += v_gidSystemId(list[i]);
        }
        if (count == 0) {
            c_free(_this->readerGID);
            _this->readerGID = NULL;
            c_mutexLock(&_this->mutex);
            c_condSignal(&_this->cv);
            c_mutexUnlock(&_this->mutex);
        }
    }
    return V_RESULT_OK;
}

c_iter
v_resolveTopics(
    v_kernel kernel,
    const c_char *name)
{
    c_iter list;
    c_collection q;
    q_expr expr;
    c_value params[1];

    assert(kernel != NULL);
    assert(C_TYPECHECK(kernel,v_kernel));

    expr = (q_expr)q_parse("name like %0");
    params[0] = c_stringValue((char *)name);
    q = c_queryNew(kernel->topics,expr,params);
    q_dispose(expr);
    c_lockRead(&kernel->lock);
    list = c_select(q,0);
    c_lockUnlock(&kernel->lock);
    c_free(q);
    return list;
}

void
v_dataViewInstanceRemove(
    v_dataViewInstance instance)
{
    v_dataView dataView;
    v_dataViewInstance found;

    assert(C_TYPECHECK(instance,v_dataViewInstance));

    if (instance->sampleCount == 0) {
        CHECK_ZERO_INSTANCE(instance);
        if (v_objectKind (instance) == K_DATAVIEWINSTANCE) {
            dataView = v_dataView(v_instanceEntity(instance));
            found = c_remove(dataView->instances,instance,NULL,NULL);
            assert(found == instance);
            OS_UNUSED_ARG(found);
            v_publicFree(v_public(instance));
            c_free(instance);
        }
    } else {
        CHECK_INSTANCE(instance);
    }
}

void
writerDeadlineMissed(
    v_leaseAction leaseAction,
    c_time now)
{
    v_object w;
    v_handleResult r;

    assert(leaseAction != NULL);
    assert(C_TYPECHECK(leaseAction, v_leaseAction));

    r = v_handleClaim(leaseAction->actionObject, &w);
    if (r == V_HANDLE_OK)
    {
        v_writerCheckDeadlineMissed(v_writer(w), now);
        r = v_handleRelease(leaseAction->actionObject);
        if(r != V_HANDLE_OK)
        {
            OS_REPORT_1(OS_WARNING, "v_leaseManager", 0,
                "Handle release failed with result code %d ", r);
        }
    }
}

c_ulong
v__observerWait(
    v_observer o)
{
    os_result result = os_resultSuccess;
    c_ulong flags;

    assert(o != NULL);
    assert(C_TYPECHECK(o,v_observer));

    if (o->eventFlags == 0) {
        o->waitCount++;
        result = c_condWait(&o->cv,&o->mutex);
        o->waitCount--;
    }
    flags = o->eventFlags;
    /* Reset events but remember destruction event.
     * To avoid any further use of this observer in case of destruction.
     */
    o->eventFlags &= V_EVENT_OBJECT_DESTROYED;

    return flags;
}

void
v_groupStreamDeinit(
    v_groupStream stream)
{
    c_iter groups;
    v_group group;

    assert(C_TYPECHECK(stream, v_groupStream));

    v_readerDeinit(v_reader(stream));

    groups = ospl_c_select(stream->groups, 0);
    group = v_group(c_iterTakeFirst(groups));

    while(group){
        v_groupRemoveStream(group, stream);
        c_free(group);
        group = v_group(c_iterTakeFirst(groups));
    }
    c_iterFree(groups);
    c_free(stream->groups);
    stream->groups = NULL;
}

/**************************************************************
 * Protected functions
 **************************************************************/
v_partitionAdmin
v_partitionAdminNew(
    v_kernel kernel)
{
    v_partitionAdmin da;
    c_base base;

    assert(C_TYPECHECK(kernel,v_kernel));

    base = c_getBase(kernel);
    da = v_partitionAdmin(v_objectNew(kernel, K_DOMAINADMIN));
    if (da != NULL) {
        da->partitions         = c_tableNew(v_kernelType(kernel, K_DOMAIN),"name");
        da->partitionInterests = c_tableNew(v_kernelType(kernel, K_DOMAININTEREST), "expression");
        c_mutexInit(&da->mutex,SHARED_MUTEX);

        if ((da->partitions == NULL) || (da->partitionInterests == NULL)) {
            c_free(da);
            da = NULL;
        }
    }
    return da;
}

v_result
v_deliveryWaitListFree(
    v_deliveryWaitList _this)
{
    v_deliveryWaitList found;
    v_result result;

    assert(C_TYPECHECK(_this,v_deliveryWaitList));

    if (_this) {
        /* lookup or create a writer specific admin.
         */
        found = c_remove(v_deliveryGuard(_this->guard)->waitlists, _this, NULL, NULL);
        assert(found == _this);
        assert(c_refCount(found) == 2);
        c_free(found);
        c_free(_this);
        result = V_RESULT_OK;
    } else {
        result = V_RESULT_ILL_PARAM;
    }
    return result;
}

v_rnr
v_rnrNew(
    v_kernel kernel,
    const c_char *name,
    const c_char *extStateName,
    v_participantQos qos,
    c_bool enable)
{
    v_rnr _this;
    v_participantQos q;

    assert(C_TYPECHECK(kernel, v_kernel));
    assert(name != NULL);

    q = v_participantQosNew(kernel, qos);
    if (q == NULL) {
        OS_REPORT(OS_ERROR, "v_rnrNew", V_RESULT_ILL_PARAM,
                  "Record and Replay service not created: inconsistent qos");
        _this = NULL;
    } else {
        _this = v_rnr(v_objectNew(kernel, K_RNR));
        _this->statistics = v_rnrStatisticsNew (kernel, name);
        v_serviceInit(v_service(_this), name, extStateName, V_SERVICETYPE_RECORD_REPLAY, q, enable);
        c_free(q);
        /* always add, even when s->state==NULL, since v_participantFree always
         * removes the participant.
         */
        v_addParticipant(kernel, v_participant(_this));
        if (v_service(_this)->state == NULL) {
            v_serviceFree(v_service(_this));
            _this = NULL;
        } else {
            OSPL_ADD_OBSERVER(kernel, _this, V_EVENT_NEW_GROUP, NULL);
        }
    }
    return _this;
}