c_bool
determineFirstLeaseToExpire(
    c_object o,
    c_voidp arg)
{
    c_time headExpTime;
    c_time leaseExpTime;
    v_leaseAction leaseAction = v_leaseAction(o);
    v_leaseManager lm = v_leaseManager(arg);

    if (lm->firstLeaseToExpire == NULL)
    {
        lm->firstLeaseToExpire = c_keep(leaseAction);
    } else
    {
        headExpTime = v_leaseExpiryTime(lm->firstLeaseToExpire->lease);
        leaseExpTime = v_leaseExpiryTime(leaseAction->lease);
        if (c_timeCompare(headExpTime, leaseExpTime) == C_GT)
        {
            c_free(lm->firstLeaseToExpire);
            lm->firstLeaseToExpire = c_keep(leaseAction);
        }
    }
    return TRUE;
}

v_cache
v_cacheNew (
    v_kernel kernel,
    c_type type,
    v_cacheKind kind)
{
    c_base base = NULL;
    c_type cache_t;
    v_cache cache = NULL;

    assert(C_TYPECHECK(cache,v_cache));
    assert(C_TYPECHECK(type,c_type));

    if (type) {
        base = c_getBase(type);
        if (base) {
            cache_t = c_keep(v_kernelType(kernel,K_CACHE));
            cache = c_new(cache_t);
            c_free(cache_t);
            if (cache) {
                cache->kind = kind;
                cache->itemType = c_keep(type);
                v_cacheNodeInit(v_cacheNode(cache));
            }
        }
    }
    return cache;
}

/**************************************************************
 * constructor/destructor
 **************************************************************/
void
v_readerInit(
    v_reader r,
    const c_char *name,
    v_subscriber s,
    v_readerQos qos,
    v_statistics rs,
    c_bool enable)
{
    v_kernel kernel;

    assert(r != NULL);
    assert(s != NULL);
    assert(C_TYPECHECK(r,v_reader));
    assert(C_TYPECHECK(s,v_subscriber));
    assert(C_TYPECHECK(qos, v_readerQos));
    /* We demand the qos to be allocated in the kernel, by v_readerQosNew().
     * This way we are sure that the qos is consistent!
     */

    kernel = v_objectKernel(r);
    v_collectionInit(v_collection(r),name,rs,enable);

    r->subscriber = s;
    r->qos = c_keep(qos);
    r->subQos = c_keep(s->qos); /* reference is readonly */
    r->entrySet.entries = c_setNew(v_kernelType(kernel,K_ENTRY));
    c_mutexInit(&r->entrySet.mutex, SHARED_MUTEX);

    r->historicalDataRequest  = NULL;
    r->historicalDataComplete = FALSE;
    c_condInit(&r->historicalDataCondition, &(v_observer(r)->mutex), SHARED_COND);

}

d_storeResult
d_groupInfoBackup(
    d_groupInfo _this,
    const d_store store,
    d_groupInfo* backup)
{
    c_base base;
    c_type groupInfoType, instanceType;
    c_string keyExpr;
    d_storeResult result;

    assert(_this);
    assert(backup);

    base = c_getBase(_this->kernel);
    groupInfoType = c_resolve(base,"durabilityModule2::d_groupInfo");
    *backup = d_groupInfo(c_new(groupInfoType));
    c_free(groupInfoType);

    if (*backup) {
        (*backup)->kernel = _this->kernel; /* Unmanaged pointer */
        (*backup)->topic = c_keep(_this->topic);
        (*backup)->partition = c_keep(_this->partition);
        (*backup)->quality = _this->quality;
        (*backup)->completeness = _this->completeness;

        instanceType = d_topicInfoGetInstanceType(_this->topic);
        keyExpr = d_topicInfoGetInstanceKeyExpr(_this->topic);

        (*backup)->instances = _this->instances; /*Here's the trick */
        _this->instances = c_tableNew(instanceType, keyExpr);

        c_free(keyExpr);
        c_free(instanceType);

        if(_this->instances){
            result = D_STORE_RESULT_OK;
        } else {
            _this->instances = (*backup)->instances;

            (*backup)->instances = NULL;
            c_free(*backup);
            *backup = NULL;

            result = D_STORE_RESULT_OUT_OF_RESOURCES;
        }
    } else {
        assert(FALSE);
        result = D_STORE_RESULT_OUT_OF_RESOURCES;
    }
    return result;
}

/**************************************************************
 * Protected functions
 **************************************************************/
void
v_lifespanAdminInsert(
    v_lifespanAdmin admin,
    v_lifespanSample sample)
{
    v_lifespanSample placeHolder;
    c_equality eq;

    assert(C_TYPECHECK(admin,v_lifespanAdmin));
    assert(C_TYPECHECK(sample,v_lifespanSample));
    assert(admin->sampleCount >= 0);

    CHECK_ADMIN(admin, sample);
    eq = c_timeCompare(sample->expiryTime, C_TIME_INFINITE);
    if (eq == C_EQ) {
        return; /* no insert, since sample never expires! */
    }

    if (admin->head == NULL) {
        assert(admin->tail == NULL);
        admin->head = c_keep(sample);
        admin->tail = c_keep(sample);
    } else {
        placeHolder = admin->tail;
        eq = c_timeCompare(placeHolder->expiryTime, sample->expiryTime);
        while ((placeHolder->prev != NULL) && (eq != C_LT) /* >= */) {
            placeHolder = placeHolder->prev;
            if (placeHolder != NULL) {
                eq = c_timeCompare(placeHolder->expiryTime, sample->expiryTime);
            }
        }
        if (eq != C_LT) { /* insert before placeholder */
            assert(placeHolder == admin->head);
            sample->next = admin->head; /* transfer ref count */
            admin->head->prev = sample;
            admin->head = c_keep(sample);
        } else {
            if (placeHolder->next != NULL) {                
                placeHolder->next->prev = sample;
            } else {
                assert(placeHolder == admin->tail);
                c_free(admin->tail);
                admin->tail = c_keep(sample);
	    }
            sample->next = placeHolder->next; /* transfer refcount */
            placeHolder->next = c_keep(sample);
            sample->prev = placeHolder;
        }
    }
    admin->sampleCount++;
    CHECK_ADMIN(admin, sample);
}

c_bool
collectExpired(
    c_object o,
    c_voidp arg)
{
    v_leaseAction leaseAction = v_leaseAction(o);
    struct collectExpiredArg *a = (struct collectExpiredArg *)arg;
    c_time headExpTime;
    c_time leaseExpTime;
    c_bool setHead;
    c_equality cmp;

    setHead = TRUE;
    leaseExpTime = v_leaseExpiryTime(leaseAction->lease);
    /*
     * A lease is expired if the expiry time is greater than or equal
     * to the current time!
     */
    cmp = c_timeCompare(a->now, leaseExpTime);
    if ((cmp ==  C_GT) || (cmp == C_EQ)) {
        a->expiredLeases = c_iterInsert(a->expiredLeases, c_keep(leaseAction));
        /* An expired lease can still become the next expirytime,
         * if it should be repeated
         */
        if (leaseAction->repeat)
        {
            v_leaseRenew(leaseAction->lease, NULL);
        } else
        {
            setHead = FALSE;
        }
    }
    if (setHead) {
        if (a->firstLeaseToExpire == NULL) {
            a->firstLeaseToExpire = c_keep(leaseAction);
        } else {
            headExpTime = v_leaseExpiryTime(a->firstLeaseToExpire->lease);
            leaseExpTime = v_leaseExpiryTime(leaseAction->lease);

            if (c_timeCompare(headExpTime, leaseExpTime) == C_GT) {
                c_free(a->firstLeaseToExpire);
                a->firstLeaseToExpire = c_keep(leaseAction);
            }
        }
    }

    return TRUE;
}

static v_actionResult
readerActionView (
    c_object o,
    c_voidp copyArg)
{
    readerViewActionArg    *info    = (readerViewActionArg *) copyArg;
    v_readerSampleSeq  *samples = info->samples;
    v_readerSample      sample;
    v_actionResult      result  = V_PROCEED;
    gapi_unsigned_long  i;

    if ( o ) {
        sample = v_readerSample(o);
        if (gapi_matchesReaderMask(o, &info->datareaderview->reader_mask)) {
            if ( !sampleSeqContains(samples, sample) ) {
                i = samples->_length;

                if ( v_readerSampleSeq_setLength(samples, i+1) ) {
                    samples->_buffer[i] = c_keep(sample);
                    c_keep(sample->instance);
                } else {
                    info->result = GAPI_RETCODE_OUT_OF_RESOURCES;
                    v_actionResultClear(result, V_PROCEED);
                }

                if ( samples->_length >= info->max ) {
                    v_actionResultClear(result, V_PROCEED);
                }
            }
        } else {
            v_actionResultSet(result, V_SKIP);
        }
    } else {
        if ( samples->_length > 0 ) {
            determineSampleInfoView(info);
            for ( i = 0UL; i < samples->_length; i++ ) {
                c_free(samples->_buffer[i]->instance);
                c_free(samples->_buffer[i]);
            }
        } else {
            info->readerCopy(NULL, info->readerInfo);
            info->result = GAPI_RETCODE_NO_DATA;
        }
        v_actionResultClear(result, V_PROCEED);
    }

    return result;
}

static c_bool
takeOne(
    c_object o,
    c_voidp arg)
{
    v_readerSample s;
    v_readerSample *sample;
    c_bool result;

    s      = (v_readerSample)o;
    sample = (v_readerSample *)arg;

    if (s != NULL) {
        result = TRUE;
        if (v_stateTest(s->sampleState, L_VALIDDATA)) {
            *sample = c_keep(s);
            result = FALSE;
        } else {
            *sample = NULL;
        }
    } else { /* last sample */
        result = FALSE;
    }

    return result;
}

void
v_networkReaderFree(
    v_networkReader reader)
{
    c_bool sendTo, more;
    v_message message;
    c_ulong sequenceNumber, priority;
    v_gid sender, receiver;
    os_timeE sendBefore;
    v_networkReaderEntry entry;
    c_ulong i;

    c_keep(reader);

    /* call inherited free,
     * which will remove the entry from the associated groups
     */
    v_readerFree(v_reader(reader));

    /* remove the messages still pressent in the network queues */
    for (i = 0; i < reader->nofQueues; i++) {
        while (v_networkQueueTakeFirst (
                    v_networkQueue(reader->queues[i]), &message, &entry, &sequenceNumber,
                    &sender, &sendTo, &receiver, &sendBefore, &priority, &more)) {
            c_free(message);
            c_free(entry);
        }
    }

    c_free(reader);
}

c_metaObject
c_scopeInsert(
    c_scope scope,
    c_metaObject object)
{
    c_binding binding, found;

    binding = c_bindingNew(scope, object);
    found = c_binding(c_avlTreeInsert(c_avlTree(scope),
                                      (void *)binding,
                                      c_bindingCompare,
                                      NULL));
    if (found == binding) {
        if (!scope->headInsOrder) {
            scope->headInsOrder = binding;
        }
        if (scope->tailInsOrder) {
           scope->tailInsOrder->nextInsOrder = binding;
        }
        scope->tailInsOrder = binding;
    } else {
        if (c_isFinal(found->object) == FALSE) {
            c_metaCopy(object,found->object);
        }
        c_bindingFree(binding,MM(scope));
    }

    c_keep(found->object);

    /** Note that if inserted (found == binding) the object reference count is increased by 2.
        one for being inserted and one for being returned.
     */
    return found->object;
}

c_metaObject
c_scopeLookup(
    c_scope scope,
    const c_char *name,
    c_long metaFilter)
{
    c_binding binding;
    c_metaObject o;
    struct c_resolveArg resolve;

    if (scope == NULL) {
        return NULL;
    }

    resolve.name = (char *)name;
    resolve.metaFilter = metaFilter;
    binding = c_avlTreeFind(c_avlTree(scope),
                            NULL,
                            c_bindingCompare,
                            &resolve);
    if (binding != NULL) {
	if (CQ_KIND_IN_MASK (binding->object, metaFilter)) {
            o = c_keep(binding->object);
        } else {
            o = NULL;
        }
    } else {
        o = NULL;
    }
    return o;
}

void
v_topicAdapterInit(
    v_topicAdapter adapter,
    v_topic        topic,
    v_participant  p,
    const c_char  *name)
{
    v_eventMask mask = V_EVENT_ALL_DATA_DISPOSED | V_EVENT_INCONSISTENT_TOPIC;

    assert(adapter != NULL);
    assert(p != NULL);
    assert(C_TYPECHECK(adapter, v_topicAdapter));
    assert(C_TYPECHECK(p,v_participant));
    assert(C_TYPECHECK(topic,v_topic));

    v_entityInit(v_entity(adapter), name);
    adapter->topic = c_keep(topic);
    (void)v_entityEnable(v_entity(adapter));

    (void)v_observerSetEvent(v_observer(adapter), mask);

    OSPL_ADD_OBSERVER(topic, adapter, mask, NULL);
    v_participantAdd(p, v_object(adapter));
    v_topic(adapter)->owner = p;
}

v_writerSample
_v_writerSampleNew(
    v_writer writer,
    v_message message)
{
    v_writerSample sample;

    assert(writer != NULL);
    assert(C_TYPECHECK(writer,v_writer));
    assert(message);
    assert(C_TYPECHECK(message,v_message));

#ifdef _EXTENT_
    sample = v_writerSample(c_extentCreate(writer->sampleExtent));
#else
    sample = v_writerSample(c_new(writer->sampleField->type));
#endif

    if (sample) {
        v_writerSampleTemplate(sample)->message = c_keep(message);
        sample->next = NULL;
        sample->prev = NULL;
        sample->sequenceNumber = 1;
        sample->sentBefore = FALSE;
        v_writerSampleClear(sample);
    } else {
        OS_REPORT(OS_ERROR,
                  "v_writerSampleNew",0,
                  "Failed to allocate sample.");
    }

    assert(C_TYPECHECK(sample,v_writerSample));

    return sample;
}

/* This method registers a database object to be managed by the user-layer.
 * Once a process has registered an object it can free its reference.
 * The user-layer will keep the registered object alive until it is deregistered
 * using the u_userFree method.
 * The user-layer will free all references of registered objects on process
 * termination via the installed exit handler (u_userExit).
 */
c_object
u_userKeep(
    c_object o)
{
    u_user u;
    u_domainAdmin ka;
    c_long i;

    if (o) {
        u = u__userLock();
        if (u) {
            /* the user-layer object exists so now find the domain that holds
             * the given object.
             */
            for (i=1; i <= u->domainCount; i++) {
                ka = &u->domainList[i];
                if (ka->domain) {
                    /* A valid domain admin exists, now check if the objects
                     * address is in the domains address range.
                     */
                    if (((c_address)o >= ka->lowerBound) &&
                        ((c_address)o <= ka->upperBound))
                    {
                        c_keep(o);
                        ka->keepList = c_iterInsert(ka->keepList,o);
                        i = u->domainCount + 1; /* jump out of the loop */
                    }
                }
            }
            u__userUnlock();
        }
    }
    return o;
}

v_partition
v_partitionNew(
    v_kernel kernel,
    const c_char *name,
    v_partitionQos qos)
{
    v_partition partition, found;

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

    assert(name != NULL);
    assert(v_partitionExpressionIsAbsolute(name));

    partition = v_partition(v_objectNew(kernel,K_DOMAIN));
    v_entityInit(v_entity(partition),name, NULL, TRUE);

    found = v_addPartition(kernel,partition);

    if (found != partition) {
        v_partitionFree(partition);
        c_free(partition); /* v_partitionFree has removed all dependancies, now delete local reference */
        partition = c_keep(found); /* this one will be returned, so a keep is required */
    }
    return partition;
}

c_baseObject
c_scopeResolve(
    c_scope scope,
    const char *name,
    c_long metaFilter)
{
    c_metaObject o = NULL;
    struct c_walkArg warg;
    struct c_findArg farg;

    if (scope == NULL) {
        return NULL;
    }
    if (metaFilter & CQ_CASEINSENSITIVE) {

        warg.arg = &farg;
        warg.compare = c_metaNameCompare;
        warg.object = NULL;
	farg.name = name;
	farg.metaFilter = metaFilter;
	farg.object = NULL;
        if (c_avlTreeWalk(c_avlTree(scope),walkCompare,&warg,C_POSTFIX) == FALSE) {
            if (warg.object) {
                o = c_keep(warg.object);
            }
        } else {
            o = NULL;
        }
    } else {
        o = c_scopeLookup(scope, name, metaFilter);
    }
    return c_baseObject(o);
}

c_bool
d_readerListenerCopy(
    c_object object,
    c_voidp copyArg)
{
    d_readerListener listener;
    d_message        message;
    v_message        readerMessage;
    c_bool result = FALSE;

    if(object != NULL) {
        if (v_stateTest(v_readerSampleState(object), L_VALIDDATA)) {
            listener  = d_readerListener(copyArg);
            readerMessage = v_message(v_dataReaderSampleTemplate(object)->message);
            message   = C_DISPLACE(readerMessage, listener->fieldOffset);
            listener->lastInsertTime = v_dataReaderSample(object)->insertTime;
            listener->lastSourceTime = readerMessage->writeTime;
            listener->value = c_keep(object);

            if(listener->value){
                listener->message = message;
            }
            listener->processMessage = TRUE;
        } else {
            /* Ignore data that is not valid but continue with the read action */
            result = TRUE;
        }
    }
    return result;
}

v_deadLineInstanceList
v_deadLineInstanceListNew(
    c_base base,
    v_leaseManager leaseManager,
    v_duration leaseDuration,
    v_leaseActionId actionId,
    v_public o)
{
    v_deadLineInstanceList list;
    c_type type;

    assert(C_TYPECHECK(leaseManager,v_leaseManager));
    assert(C_TYPECHECK(o,v_public));

    type = c_resolve(base, "kernelModule::v_deadLineInstanceList");
    assert(type);
    list = c_new(type);
    c_free(type);
    if (list) {
        v_objectKind(list) = K_DEADLINEINSTANCE;
        v_instanceInit(v_instance(list));
        list->leaseManager = c_keep(leaseManager);
        list->leaseDuration = leaseDuration;
        list->deadlineLease = NULL;
        list->actionObject = o; /* no keep, since actionObject is onwer of v_deadLineInstanceList */
        list->actionId = actionId;
    } else {
        OS_REPORT(OS_ERROR,
                  "v_deadLineInstanceListNew",0,
                  "Failed to allocate v_deadLineInstanceList.");
    }

    return list;
}

v_cache
v_groupCacheNew (
    v_kernel kernel,
    v_cacheKind kind)
{
    c_base base;
    c_type type;
    v_cache cache;

    assert(C_TYPECHECK(kernel,v_kernel));

    base = c_getBase(kernel);
    type = c_keep(v_kernelType(kernel,K_GROUPCACHEITEM));
    cache = v_cacheNew(kernel,type,kind);
    c_free(type);

    if (!cache) {
        OS_REPORT(OS_ERROR,
                  "v_groupCacheNew",0,
                  "Failed to allocate group cache.");
    }

    assert(C_TYPECHECK(cache, v_cache));

    return cache;
}

void
v_dataViewInstanceWipe(
    v_dataViewInstance instance)
{
    v_dataViewSample sample,prev,firstSample;

    if (instance == NULL) {
        return;
    }

    assert (C_TYPECHECK (instance, v_dataViewInstance));

    CHECK_INSTANCE(instance);

    if (instance->sampleCount != 0) {
        sample = v_dataViewInstanceTemplate(instance)->sample;
        firstSample = c_keep(sample);
        while (sample != NULL) {
            v_dataViewSampleListRemove(v_dataViewSampleList(sample));
            prev = sample->prev;
            sample->prev = NULL;
            c_free(sample);
            sample = prev;
        }
        instance->sampleCount = 0;
        v_dataViewInstanceTemplate(instance)->sample = firstSample;
    }

    CHECK_ZERO_INSTANCE(instance);
}