自学教程：C++ v_objectKernel函数代码示例

voidv_participantDeleteHistoricalData(    v_participant participant,    const c_char* partitionExpr,    const c_char* topicExpr){    c_iter matchingGroups;    v_group group;    c_time t;    c_value params[2];    C_STRUCT(v_event) event;    C_STRUCT(v_historyDeleteEventData) hde;    assert(participant != NULL);    assert(C_TYPECHECK(participant, v_participant));    assert(partitionExpr);    assert(topicExpr);    if(partitionExpr && topicExpr){        params[0]  = c_stringValue((c_string)partitionExpr);        params[1]  = c_stringValue((c_string)topicExpr);        c_lockRead(&participant->lock);        t = v_timeGet();        matchingGroups = v_groupSetSelect(                                v_objectKernel(participant)->groupSet,                                "partition.name like %0 AND topic.name like %1",                                params);        c_lockUnlock(&participant->lock);        group = v_group(c_iterTakeFirst(matchingGroups));        while(group){            v_groupDeleteHistoricalData(group, t);            c_free(group);            group = v_group(c_iterTakeFirst(matchingGroups));        }        c_iterFree(matchingGroups);        hde.partitionExpression = (c_char *)partitionExpr;        hde.topicExpression = (c_char *)topicExpr;        hde.deleteTime = t;        event.kind = V_EVENT_HISTORY_DELETE;        event.source = v_publicHandle(v_public(participant));        event.userData = &hde;        v_observableNotify(v_observable(v_objectKernel(participant)),&event);    }    return;}

c_boolv_waitsetAttach (    v_waitset _this,    v_observable o,    c_voidp userData){    c_bool result;    v_proxy proxy;    findProxyArgument arg;    assert(_this != NULL);    assert(C_TYPECHECK(_this,v_waitset));    arg.observable = v_publicHandle(v_public(o));    arg.proxy = NULL;    v_waitsetLock(_this);    c_setWalk(_this->observables, findProxy,&arg);    if (arg.proxy == NULL) { /* no proxy to the observer exists */        proxy = v_proxyNew(v_objectKernel(_this),                           arg.observable, userData);        c_insert(_this->observables,proxy);        c_free(proxy);    }    v_waitsetUnlock(_this);    result = v_observableAddObserver(o,v_observer(_this), userData);    /* wakeup blocking threads to evaluate new condition. */    if (v_observerWaitCount(_this)) {        v_waitsetTrigger(_this, NULL);    }    return result;}

v_cmsoapv_cmsoapNew(    v_serviceManager manager,    const c_char *name,    const c_char *extStateName,    v_participantQos qos){    v_kernel k;    v_cmsoap s;    v_participantQos q;    assert(C_TYPECHECK(manager, v_serviceManager));    assert(name != NULL);    k = v_objectKernel(manager);    q = v_participantQosNew(k, qos);     if (q == NULL) {        OS_REPORT(OS_ERROR, "v_cmsoapNew", 0,                  "CMSoap service not created: inconsistent qos");        s = NULL;    } else {        s = v_cmsoap(v_objectNew(k, K_CMSOAP));        v_serviceInit(v_service(s), manager, name, extStateName, q, v_statistics(v_cmsoapStatisticsNew(k)));        c_free(q);        /* always add, even when s->state==NULL, since v_participantFree always           removes the participant.*/        v_addParticipant(k, v_participant(s));        if (v_service(s)->state == NULL) {            v_serviceFree(v_service(s));            s = NULL;        }    }    return s;}

v_dataViewInstancev_dataViewInstanceNew(    v_dataView dataView,    v_dataViewSample viewSample){    v_dataViewInstance instance;    assert(dataView);    assert(viewSample);    assert(C_TYPECHECK(dataView,v_dataView));    assert(C_TYPECHECK(viewSample,v_dataViewSample));    instance = v_dataViewInstance(c_new(dataView->instanceType));    if (instance) {        v_object(instance)->kernel = v_objectKernel(dataView);        v_objectKind(instance) = K_DATAVIEWINSTANCE;        v_instanceInit(v_instance(instance), v_entity(dataView));        viewSample->next = viewSample;        v_dataViewInstanceTemplate(instance)->sample = viewSample;        instance->sampleCount = 1;        v_stateSet(v_instanceState(instance),L_NEW);        v_stateClear(v_readerSample(viewSample)->sampleState,L_READ);        assert(C_TYPECHECK(instance,v_dataViewInstance));        CHECK_INSTANCE(instance);    } else {        OS_REPORT(OS_FATAL, OS_FUNCTION, V_RESULT_INTERNAL_ERROR,            "Failed to allocate v_dataViewInstance");        assert(FALSE);    }    return instance;}

voidcmx_participantDomainsAction(    v_public p,    c_voidp args){    cmx_walkEntityArg arg;    c_iter partitions;    v_entity partition;    c_bool proceed;    c_char* xmlEntity;    arg = cmx_walkEntityArg(args);    partitions = v_resolvePartitions(v_objectKernel(p), "*");    partition = v_entity(c_iterTakeFirst(partitions));    while(partition != NULL){        proceed = cmx_entityNewFromWalk(v_public(partition), &arg->entityArg);        if(proceed == TRUE){            xmlEntity = arg->entityArg.result;            arg->list = c_iterInsert(arg->list, xmlEntity);            arg->length += strlen(xmlEntity);        }        c_free(partition);        partition = v_entity(c_iterTakeFirst(partitions));    }    c_iterFree(partitions);}

v_resultv_subscriberEnable (    v_subscriber _this){    v_kernel kernel;    c_iter list;    c_char *partitionName;    v_result result = V_RESULT_ILL_PARAM;    if (_this) {        kernel = v_objectKernel(_this);        v_observableAddObserver(v_observable(kernel->groupSet),                                v_observer(_this), NULL);        if (_this->qos->partition != NULL) {            list = v_partitionPolicySplit(_this->qos->partition);            while((partitionName = c_iterTakeFirst(list)) != NULL) {                v_subscriberSubscribe(_this,partitionName);                os_free(partitionName);            }            c_iterFree(list);        }        result = V_RESULT_OK;    }    return result;}

/************************************************************** * Private functions **************************************************************/static c_boolv__partitionAdminAdd(    v_partitionAdmin da,    const char *partitionName,    v_partition *newPartition){    c_bool result = TRUE;    v_partition partition, found;    assert(v_partitionExpressionIsAbsolute(partitionName));    assert(newPartition != NULL);    partition = v_partitionNew(v_objectKernel(da), partitionName, NULL);    found = c_tableInsert(da->partitions, partition);    if (found != partition) {        c_free(partition);        result = FALSE;        *newPartition = NULL;    } else {        /* Do not free partition here because it is returned */        *newPartition = partition;    }    return result;}

v_result v_participantCandMCommandSetDisposeAllData(v_participant participant,                                                    v_message msg,                                                    char *topicExpr,                                                    char *partitionExpr){    v_kernel kernel;    v_topic topic;    c_base base;    v_controlAndMonitoringCommand *command;    struct v_commandDisposeAllData *disposeCmd;    assert(participant != NULL);    assert(C_TYPECHECK(participant,v_participant));    assert(msg != NULL );    assert(C_TYPECHECK(msg,v_message));    kernel = v_objectKernel(participant);    topic = v_builtinTopicLookup(kernel->builtin, V_C_AND_M_COMMAND_ID);    command = v_builtinControlAndMonitoringCommandData(kernel->builtin, msg);    command->u._d = V_COMMAND_DISPOSE_ALL_DATA;    base = c_getBase(c_object(topic));    disposeCmd = &command->u._u.dispose_all_data_info;    disposeCmd->topicExpr = c_stringNew(base, topicExpr);    disposeCmd->partitionExpr = c_stringNew(base, partitionExpr);    return ( ( disposeCmd->topicExpr != NULL               && disposeCmd->partitionExpr != NULL )             ? V_RESULT_OK             : V_RESULT_OUT_OF_MEMORY );}

v_resultv_participantSetQos(    v_participant p,    v_participantQos qos){    v_message builtinMsg;    v_kernel kernel;    v_qosChangeMask cm;    v_result result;    assert(C_TYPECHECK(p,v_participant));    /* Do not use C_TYPECHECK on qos parameter, since it might be allocated on heap! */    kernel = v_objectKernel(p);    c_lockWrite(&p->lock);    result = v_participantQosSet(p->qos, qos, &cm);    if ((result == V_RESULT_OK) && (cm != 0)) {        builtinMsg = v_builtinCreateParticipantInfo(kernel->builtin,p);        c_lockUnlock(&p->lock);        v_writeBuiltinTopic(kernel, V_PARTICIPANTINFO_ID, builtinMsg);        c_free(builtinMsg);    } else {        c_lockUnlock(&p->lock);    }    return result;}

static voiddeliveryServiceUnSubscribe(    void *o,    void *arg){    v_partition p = v_partition(o);    v_deliveryServiceEntry e = v_deliveryServiceEntry(arg);    v_kernel kernel;    v_group g;    c_value params[2];    c_iter list;    assert(C_TYPECHECK(e,v_deliveryServiceEntry));    assert(C_TYPECHECK(p,v_partition));    params[0] = c_objectValue(p);    params[1] = c_objectValue(e->topic);    kernel = v_objectKernel(e);    list = v_groupSetSelect(kernel->groupSet,                            "partition = %0 and topic = %1",                            params);    while ((g = c_iterTakeFirst(list)) != NULL) {        v_groupRemoveEntry(g,v_entry(e));        c_free(g);    }    c_iterFree(list);}

voidcmx_participantFindTopicAction(    v_public p,    c_voidp args){    cmx_walkEntityArg arg;    c_iter topics;    v_entity topic;    c_bool proceed;    c_char* xmlEntity;    arg = cmx_walkEntityArg(args);    topics = v_resolveTopics(v_objectKernel(p), cmx_walkParticipantArg(arg)->topicName);    topic = v_entity(c_iterTakeFirst(topics));    while(topic != NULL){        proceed = cmx_entityNewFromWalk(v_public(topic), &arg->entityArg);        if(proceed == TRUE){            xmlEntity = arg->entityArg.result;            arg->list = c_iterInsert(arg->list, xmlEntity);            arg->length += strlen(xmlEntity);        }        c_free(topic);        topic = v_entity(c_iterTakeFirst(topics));    }    c_iterFree(topics);}

v_resultv_deliveryServiceEnable(    v_deliveryService _this){#if 0    v_kernel kernel;    v_message builtinMsg;#endif    v_subscriber subscriber;    v_result result;    if (_this) {        result = V_RESULT_OK;        subscriber = v_subscriber(v_reader(_this)->subscriber);        v_subscriberAddReader(subscriber,v_reader(_this));#if 0        kernel = v_objectKernel(_this);        builtinMsg = v_builtinCreateSubscriptionInfo(kernel->builtin, _this);        v_writeBuiltinTopic(kernel, V_SUBSCRIPTIONINFO_ID, builtinMsg);        c_free(builtinMsg);#endif    } else {        result = V_RESULT_ILL_PARAM;    }    return result;}

voidv_deliveryServiceFree (    v_deliveryService _this){#if 0    v_message builtinMsg;    v_message unregisterMsg;#endif    assert(_this != NULL);    assert(C_TYPECHECK(_this,v_deliveryService));    v_readerFree(v_reader(_this));#if 0    /* First create message, only at the end dispose. Applications expect     * the disposed sample to be the last!     */    kernel = v_objectKernel(_this);    builtinMsg = v_builtinCreateSubscriptionInfo(kernel->builtin,_this);    unregisterMsg = v_builtinCreateSubscriptionInfo(kernel->builtin,_this);    v_writeDisposeBuiltinTopic(kernel, V_SUBSCRIPTIONINFO_ID, builtinMsg);    v_unregisterBuiltinTopic(kernel, V_SUBSCRIPTIONINFO_ID, unregisterMsg);    c_free(builtinMsg);    c_free(unregisterMsg);#endif}

c_boolv_entryAddGroup(    v_entry entry,    v_group group){    v_proxy proxy;    v_proxy found;    c_bool result;    assert(C_TYPECHECK(entry,v_entry));    assert(C_TYPECHECK(group,v_group));    proxy = v_proxyNew(v_objectKernel(group),                       v_publicHandle(v_public(group)), NULL);    found = c_insert(entry->groups, proxy);    if(found != proxy){        /* The group was already available in the groupset. This can happen if         * the reader gets notified of the group it has just created. In that         * case the administration should not be updated. */        result = FALSE;    } else {        result = TRUE;    }    c_free(proxy);    return result;}

v_listenerv_listenerNew(    v_participant p,    c_bool combine){    v_listener _this;    v_kernel kernel;    assert(C_TYPECHECK(p,v_participant));    kernel = v_objectKernel(p);    _this = v_listener(v_objectNew(kernel,K_LISTENER));    if (_this != NULL) {        v_publicInit(v_public(_this));        (void)c_mutexInit(c_getBase(_this), &_this->mutex);        c_condInit(c_getBase(_this), &_this->cv, &_this->mutex);        _this->participant = p;        _this->eventList = NULL;        _this->lastEvent = NULL;        v_participantAdd(p, v_object(_this));        _this->terminate = FALSE;        _this->waitCount = 0;        _this->combine = combine;    }    return _this;}

c_boolv_groupStreamSubscribe(    v_groupStream stream,    v_partition partition){    c_iter list;    v_kernel kernel;    c_value params[1];    v_group group;    assert(C_TYPECHECK(stream,v_groupStream));    kernel = v_objectKernel(v_entity(partition));    params[0] = c_objectValue(partition);    list = v_groupSetSelect(kernel->groupSet,"partition = %0 ",params);    group = c_iterTakeFirst(list);    while (group != NULL) {        v_groupStreamSubscribeGroup(stream, group);        c_free(group);        group = c_iterTakeFirst(list);    }    c_iterFree(list);    return TRUE;}

v_topicAdapterv_topicAdapterWrap(    v_participant p,    v_topic topic){    v_topicAdapter adapter = NULL;    v_kernel kernel;    assert(p != NULL);    assert(C_TYPECHECK(p,v_participant));    assert(topic != NULL);    assert(C_TYPECHECK(topic,v_topic));    kernel = v_objectKernel(p);    adapter = v_topicAdapter(v_objectNew(kernel,K_TOPIC_ADAPTER));    if (adapter != NULL) {        v_topicAdapterInit(adapter, topic, p, v_topicName(topic));    } else {        OS_REPORT(OS_ERROR, "v_topicAdapterWrap", V_RESULT_INTERNAL_ERROR,                  "Failed to allocate TopicAdapter for topic '%s'.",                   v_topicName(topic));    }    return adapter;}

u_resultu_readerGetMatchedPublications (    u_reader _this,    v_statusAction action,    c_voidp arg){    v_dataReader reader;    v_spliced spliced;    v_kernel kernel;    u_result result;    c_iter participants;    v_participant participant;    result = U_RESULT_PRECONDITION_NOT_MET;    if (_this != NULL) {        result = u_entityReadClaim(u_entity(_this), (v_entity*)(&reader));        if ((result == U_RESULT_OK) && (reader != NULL)) {            kernel = v_objectKernel(reader);            participants = v_resolveParticipants(kernel, V_SPLICED_NAME);            assert(c_iterLength(participants) == 1);            participant = v_participant(c_iterTakeFirst(participants));            spliced = v_spliced(participant);            c_free(participant);            c_iterFree(participants);            result = u_resultFromKernel(                         v_splicedGetMatchedPublications(spliced, v_dataReader(reader), action, arg));            u_entityRelease(u_entity(_this));        }    }    return result;}

/************************************************************** * constructor/destructor **************************************************************/voidv_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);}

voidv_participantInit(    v_participant p,    const c_char *name,    v_participantQos qos,    v_statistics s,    c_bool enable){    v_kernel kernel;    c_base base;    v_message builtinMsg;    c_type writerProxyType;    assert(C_TYPECHECK(p,v_participant));    assert(C_TYPECHECK(qos, v_participantQos));    kernel = v_objectKernel(p);    base = c_getBase(p);    v_observerInit(v_observer(p),name,s,enable);    p->entities = c_setNew(c_resolve(base,"kernelModule::v_entity"));    p->qos = c_keep(qos);    /* Currently default LIVELINESS policy is used: kind=AUTOMATIC,     * duration=INFINITE This setting implies no lease registration.    */    p->lease = NULL;    p->leaseManager = v_leaseManagerNew(kernel);    p->resendQuit = FALSE;    c_mutexInit(&p->resendMutex, SHARED_MUTEX);    c_condInit(&p->resendCond, &p->resendMutex, SHARED_COND);    writerProxyType = v_kernelType(kernel,K_PROXY);    p->resendWriters = c_tableNew(writerProxyType, "source.index,source.serial");    p->builtinSubscriber = NULL;    if (!v_observableAddObserver(v_observable(kernel),v_observer(p), NULL)) {        if (name != NULL) {            OS_REPORT_1(OS_WARNING,"Kernel Participant",0,                        "%s: Cannot observe Kernel events",name);        } else {            OS_REPORT(OS_WARNING,"Kernel Participant",0,                      "Cannot observe Kernel events");        }    }    c_mutexInit(&p->newGroupListMutex,SHARED_MUTEX);    p->newGroupList = c_listNew(c_resolve(base, "kernelModule::v_group"));    v_observerSetEventMask(v_observer(p), V_EVENT_NEW_GROUP);    c_lockInit(&p->lock,SHARED_LOCK);    c_mutexInit(&p->builtinLock,SHARED_MUTEX);    /* Here the Builtin Topic of the participant is published.     * This call mabe a noop in case builtin is disabled on kernel level.     */    builtinMsg = v_builtinCreateParticipantInfo(kernel->builtin,p);    v_writeBuiltinTopic(kernel, V_PARTICIPANTINFO_ID, builtinMsg);    c_free(builtinMsg);}

voidv_subscriberFree(    v_subscriber s){    v_kernel kernel;    v_participant p;    v_reader o;    v_entity found;    c_long sc;    kernel = v_objectKernel(s);    sc = (c_long)pa_decrement(&(s->shareCount));    if (sc > 0) return;    if(sc == 0){        v_observableRemoveObserver(v_observable(kernel->groupSet),v_observer(s), NULL);        if (s->qos->share.enable) {            found = v_removeShare(kernel,v_entity(s));            assert(found == v_entity(s));            c_free(found);        }        while ((o = c_take(s->readers)) != NULL) {            switch (v_objectKind(o)) {            case K_DATAREADER:                v_dataReaderFree(v_dataReader(o));            break;            case K_DELIVERYSERVICE:                v_deliveryServiceFree(v_deliveryService(o));            break;            case K_GROUPQUEUE:                v_groupQueueFree(v_groupQueue(o));            break;            case K_NETWORKREADER:                v_networkReaderFree(v_networkReader(o));            break;            default:                OS_REPORT_1(OS_ERROR,                            "v_subscriber", 0,                            "Unknown reader %d",                            v_objectKind(o));                assert(FALSE);            break;            }            c_free(o);        }        p = v_participant(s->participant);        if (p != NULL) {            v_participantRemove(p,v_entity(s));            s->participant = NULL;        }        v_publicFree(v_public(s));    } else {        OS_REPORT_1(OS_ERROR,  "v_subscriberFree", 0,                "subscriber already freed (shareCount is now %d).", sc);        assert(sc == 0);    }}

v_subscriberv_participantGetBuiltinSubscriber(    v_participant p){    v_subscriberQos sQos;    v_readerQos rQos;    v_kernel kernel;    c_bool create_builtin_readers = FALSE;    assert(p != NULL);    assert(C_TYPECHECK(p, v_participant));    c_mutexLock(&p->builtinLock);    if (p->builtinSubscriber == NULL) {        kernel = v_objectKernel(p);        sQos = v_subscriberQosNew(kernel, NULL);        sQos->presentation.access_scope = V_PRESENTATION_TOPIC;        c_free(sQos->partition);        sQos->partition = c_stringNew(c_getBase(c_object(kernel)),                                      V_BUILTIN_PARTITION);        sQos->entityFactory.autoenable_created_entities = TRUE;        p->builtinSubscriber = v_subscriberNew(p, V_BUILTINSUBSCRIBER_NAME,                                               sQos, TRUE);        v_subscriberQosFree(sQos);        create_builtin_readers = TRUE;        c_mutexUnlock(&p->builtinLock);        assert(p->builtinSubscriber != NULL);        rQos = v_readerQosNew(kernel, NULL);        rQos->durability.kind = V_DURABILITY_TRANSIENT;        rQos->reliability.kind = V_RELIABILITY_RELIABLE;        rQos->history.kind = V_HISTORY_KEEPLAST;        rQos->history.depth = 1;#define _CREATE_READER_(topicName) {/            q_expr expr; /            v_dataReader dr; /            expr = q_parse("select * from " topicName);/            dr = v_dataReaderNew(p->builtinSubscriber, topicName "Reader", /                                   expr, NULL, rQos, TRUE);/            c_free(dr); /            q_dispose(expr); /        }        _CREATE_READER_(V_PARTICIPANTINFO_NAME)        _CREATE_READER_(V_TOPICINFO_NAME)        _CREATE_READER_(V_PUBLICATIONINFO_NAME)        _CREATE_READER_(V_SUBSCRIPTIONINFO_NAME)#undef _CREATE_READER_        v_readerQosFree(rQos);    } else {        c_mutexUnlock(&p->builtinLock);    }    return c_keep(p->builtinSubscriber);}

static voidgetKernelId(    v_entity e,    c_voidp arg){    v_kernel *kernel = (v_kernel *)arg;    *kernel = v_objectKernel(e);}

v_networkReaderv_networkReaderNew(    v_subscriber subscriber,    const c_char *name,    v_readerQos qos,    c_bool ignoreReliabilityQoS){    /* Note: currently, no qos-es are supported. Everything is redirected     *       to the defaultQueue */    v_kernel kernel;    v_networkReader reader;    v_readerQos q;    v_statistics s;    c_type queueType;    c_long i;    assert(C_TYPECHECK(subscriber,v_subscriber));    /* Creation */    kernel = v_objectKernel(subscriber);    q = v_readerQosNew(kernel,qos);    if (q != NULL) {        reader = v_networkReader(v_objectNew(kernel,K_NETWORKREADER));        s = v_statistics(v_networkReaderStatisticsNew(kernel));        /* Initialization of parent */        v_readerInit(v_reader(reader), name, subscriber, q, s, TRUE);        c_free(q); /* ref now in v_reader(queue)->qos */        /* This function only ever called once per network instance so no         * need to store queueType as static variable.  Look up as needed (once)         */        queueType = c_resolve(c_getBase(subscriber),"kernelModule::v_networkQueue");        /* Initialization of self */        reader->queues = NULL;        reader->queues = c_arrayNew(queueType, NW_MAX_NOF_QUEUES);        reader->nofQueues = 0;        reader->defaultQueue = NULL;        reader->remoteActivity = FALSE;        reader->ignoreReliabilityQoS = ignoreReliabilityQoS;        reader->queueCache = c_arrayNew(queueType, 2*NW_MAX_QUEUE_CACHE_PRIO);        for( i= 0; i < 2*NW_MAX_QUEUE_CACHE_PRIO; i++) {            reader->queueCache[i] = NULL;        }        c_free(queueType);        /* Add to subscriber */        v_subscriberAddReader(subscriber,v_reader(reader));    } else {        OS_REPORT(OS_ERROR, "v_networkReaderNew", 0,            "NetworkReader not created: inconsistent qos");        reader = NULL;    }    return reader;}

voidjni_getTopicKeyExpression(    v_entity entity,    c_voidp args){    v_kernel vk;    c_iter vtopics;    c_array keyList;    c_char* keyExpr;    c_long nrOfKeys, totalSize, i;    c_string fieldName, actualFieldName;    struct jni_topicArg *arg;        arg = (struct jni_topicArg *)args;    vk = v_objectKernel(entity);        if(vk != NULL){        vtopics = v_resolveTopics(vk, arg->topicName);                    if(c_iterLength(vtopics) == 0){            c_iterFree(vtopics);        }        else{            keyList = v_topicMessageKeyList(c_iterTakeFirst(vtopics));            c_iterFree(vtopics);                            nrOfKeys = c_arraySize(keyList);            if (nrOfKeys>0) {                totalSize = 0;                                for (i=0;i<nrOfKeys;i++) {                    fieldName = c_fieldName(keyList[i]);                    totalSize += (strlen(fieldName)+1-9/*skip 'userdata.'*/);                }                keyExpr = (c_char *)os_malloc((size_t)(totalSize+1));                keyExpr[0] = 0;                                for (i=0;i<nrOfKeys;i++) {                    fieldName = c_fieldName(keyList[i]);                    actualFieldName = c_skipUntil(fieldName, ".");                    actualFieldName++; /*skip '.' */                    os_strcat(keyExpr,actualFieldName);                                        if (i<(nrOfKeys-1)) {                         os_strcat(keyExpr,",");                     }                }                arg->keyExpr = keyExpr;            } else{                /*No keys, do nothing.*/            }            arg->result = U_RESULT_OK;        }    }}

/************************************************************** * private functions **************************************************************/static voidv_serviceWatchSplicedaemon(    v_service service){    v_kernel k;    v_serviceManager m;    v_serviceState splicedState;    k = v_objectKernel(service);    m = v_getServiceManager(k);    splicedState = v_serviceManagerGetServiceState(m, V_SPLICED_NAME);    v_observableAddObserver(v_observable(splicedState), v_observer(service), NULL);}

c_boolv_publicInit(    v_public o){    v_kernel kernel;        assert(C_TYPECHECK(o,v_public));    kernel = v_objectKernel(o);    o->handle = v_handleServerRegister(kernel->handleServer,o);    o->userDataPublic = NULL;    return TRUE;}

v_subscriberQosv_subscriberGetQos(    v_subscriber s){    v_subscriberQos qos;    assert(s != NULL);    c_lockRead(&s->lock);    qos = v_subscriberQosNew(v_objectKernel(s), s->qos);    c_lockUnlock(&s->lock);    return qos;}

v_message v_participantCreateCandMCommand(v_participant participant){    v_message msg;    v_kernel kernel;    v_topic topic;    assert(participant != NULL);    assert(C_TYPECHECK(participant,v_participant));    kernel = v_objectKernel(participant);    topic = v_builtinTopicLookup(kernel->builtin, V_C_AND_M_COMMAND_ID);    msg = v_topicMessageNew(topic);    return msg;}

c_iterv_partitionLookupSubscribers(    v_partition partition){    c_iter participants;    c_iter result;    c_iter entities;    c_iter partitions;    v_participant participant;    v_entity entity;    v_entity partition2;    result = NULL;    participants = v_resolveParticipants(v_objectKernel(partition), "*");    participant = v_participant(c_iterTakeFirst(participants));    while (participant != NULL) {        c_lockRead(&participant->lock);        entities = c_select(participant->entities, 0);        c_lockUnlock(&participant->lock);        entity = v_entity(c_iterTakeFirst(entities));        while (entity != NULL) {            if(v_objectKind(entity) == K_SUBSCRIBER) {                partitions = v_subscriberLookupPartitions(v_subscriber(entity),                                                    v_partitionName(partition));                if (c_iterLength(partitions) > 0) {                    result = c_iterInsert(result, entity); /* transfer refcount */                } else {                    c_free(entity);                }                partition2 = v_entity(c_iterTakeFirst(partitions));                while (partition2 != NULL) {                    c_free(partition2);                    partition2 = v_entity(c_iterTakeFirst(partitions));                }                c_iterFree(partitions);            }            /* entity is already free or refcount transferred to result */            entity = v_entity(c_iterTakeFirst(entities));        }        c_iterFree(entities);        c_free(participant);        participant = v_participant(c_iterTakeFirst(participants));    }    c_iterFree(participants);    return result;}