/*
 *                         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 "v__index.h"
#include "os_heap.h"
#include "v__topic.h"
#include "v_state.h"
#include "v_event.h"
#include "v__dataReader.h"
#include "v_dataReaderEntry.h"
#include "v__dataReader.h"
#include "v_subscriber.h"
#include "v_reader.h"
#include "v_readerQos.h"
#include "v_status.h"
#include "c_stringSupport.h"
#include "v_entity.h"
#include "v_projection.h"
#include "v_dataView.h"
#include "v__dataReaderInstance.h"
#include "v__dataReaderSample.h"
#include "v_public.h"
#include "v_instance.h"

#include "os_report.h"
#include "vortex_os.h"

static c_type
sampleTypeNew(
    v_topic topic)
{
    c_metaObject o;
    c_type msgType,sampleType,foundType;
    c_base base;
    c_char *name;
    os_size_t length;
    int sres;

    assert(C_TYPECHECK(topic,v_topic));
    assert(topic);

    name = NULL;
    foundType = NULL;

    if (v_topicName(topic) == NULL) {
        OS_REPORT(OS_ERROR,
                  "v_index::sampleTypeNew failed",V_RESULT_ILL_PARAM,
                  "failed to retreive topic name");
        return NULL;
    }

    base = c_getBase(topic);

    if (base == NULL) {
        OS_REPORT(OS_ERROR,
                  "v_index::sampleTypeNew failed",V_RESULT_ILL_PARAM,
                  "failed to retreive base");
        return NULL;
    }

    msgType = c_keep(v_topicMessageType(topic));

    if (msgType == NULL) {
        OS_REPORT(OS_ERROR,
                  "v_index::sampleTypeNew failed",V_RESULT_ILL_PARAM,
                  "failed to retreive topic message type");
        return NULL;
    }

    sampleType = c_type(c_metaDefine(c_metaObject(base),M_CLASS));
    if (sampleType) {
        c_class(sampleType)->extends = v_dataReaderSample_t(base);
        o = c_metaDeclare(c_metaObject(sampleType),"message",M_ATTRIBUTE);
        if (o) {
            c_property(o)->type = c_keep(msgType);
            c_metaObject(sampleType)->definedIn = c_keep(base);
            c_metaFinalize(c_metaObject(sampleType));

#define SAMPLE_FORMAT "v_indexSample<%s>"
#define SAMPLE_NAME   "v_indexSample<>"
            /* sizeof contains \0 */
            length = sizeof(SAMPLE_NAME) + strlen(v_topicName(topic));
            name = os_malloc(length);
            sres = snprintf(name,length,SAMPLE_FORMAT,v_topicName(topic));
            assert(sres >= 0 && (os_size_t) sres == (length-1));
            OS_UNUSED_ARG(sres);
#undef SAMPLE_FORMAT
#undef SAMPLE_NAME

            foundType = c_type(c_metaBind(c_metaObject(base),
                                          name,
                                          c_metaObject(sampleType)));
            c_free(o);
        } else {
            foundType = NULL;
        }
        c_free(sampleType);
    } else {
        OS_REPORT(OS_ERROR,
                  "v_index::sampleTypeNew failed",V_RESULT_ILL_PARAM,
                  "failed to retreive topic sample type");
    }
    os_free(name);
    c_free(msgType);

    return foundType;
}

static c_type
createInstanceType (
    v_topic topic,
    c_char *keyExpr,
    c_array *keyListRef)
{
    c_metaObject o;
    c_type instanceType, baseType, foundType;
    c_type sampleType, keyType, keyInstanceType;
    c_base base;
    c_char *name;
    os_size_t length;
    int sres;

    assert(C_TYPECHECK(topic,v_topic));

    foundType = NULL;
    if (keyExpr) {
        keyType = v_topicKeyTypeCreate(topic,keyExpr,keyListRef);
    } else {
        keyExpr = v_topicKeyExpr(topic);
        keyType = v_topicKeyType(topic);
        *keyListRef = c_keep(v_topicMessageKeyList(topic));
    }
    sampleType = sampleTypeNew(topic);
    if (sampleType) {
        base = c_getBase(topic);
        baseType = v_dataReaderInstance_t(base);
        instanceType = c_type(c_metaDefine(c_metaObject(base),M_CLASS));
        if (instanceType != NULL) {
            c_class(instanceType)->extends = c_keep(c_class(baseType));
            o = c_metaDeclare(c_metaObject(instanceType),
                              "sample",M_ATTRIBUTE);
            c_property(o)->type = c_keep(sampleType);
            c_free(o);
            o = c_metaDeclare(c_metaObject(instanceType),
                              "oldest",M_ATTRIBUTE);
            c_property(o)->type = (c_type)c_metaResolveType(c_metaObject(base),
                                                    "c_voidp");
            assert(c_property(o)->type);
            c_free(o);
            c_metaFinalize(c_metaObject(instanceType));
#define INSTANCE_NAME   "v_indexInstance<v_indexSample<>>"
#define INSTANCE_FORMAT "v_indexInstance<v_indexSample<%s>>"
            /* The sizeof contains \0 */
            length = sizeof(INSTANCE_NAME) + strlen(v_topicName(topic));
            name = os_alloca(length);
            sres = snprintf(name,length,INSTANCE_FORMAT,v_topicName(topic));
            assert(sres >= 0 && (os_size_t) sres == (length-1));
            OS_UNUSED_ARG(sres);
#undef INSTANCE_NAME
#undef INSTANCE_FORMAT
            foundType = c_type(c_metaBind(c_metaObject(base),
                                          name,
                                          c_metaObject(instanceType)));
            if (foundType == NULL) {
                OS_REPORT(OS_ERROR,
                        "v_index::createInstanceType",V_RESULT_INTERNAL_ERROR,
                        "Could not create instance type");
            }

            os_freea(name);

            if (keyType != NULL) {
                keyInstanceType = c_type(c_metaDefine(c_metaObject(base),M_CLASS));
                if (keyInstanceType) {
                    c_class(keyInstanceType)->extends = c_keep(c_class(instanceType));
                    o = c_metaDeclare(c_metaObject(keyInstanceType),
                                      "key",M_ATTRIBUTE);
                    c_property(o)->type = c_keep(keyType);
                    c_free(o);
                    c_metaFinalize(c_metaObject(keyInstanceType));
#define INSTANCE_NAME   "v_indexKeyInstance<v_indexSample<>,>"
#define INSTANCE_FORMAT "v_indexKeyInstance<v_indexSample<%s>,%s>"
                    /* The sizeof contains \0 */
                    length = sizeof(INSTANCE_NAME) +
                             strlen(v_topicName(topic)) +
                             strlen(keyExpr);
                    name = os_alloca(length);
                    sres = snprintf(name,
                                    length,
                                    INSTANCE_FORMAT,
                                    v_topicName(topic),
                                    keyExpr);
                    assert(sres >= 0 && (os_size_t) sres == (length-1));
                    OS_UNUSED_ARG(sres);
#undef INSTANCE_NAME
#undef INSTANCE_FORMAT
                    c_free(foundType); /* Will be overwritten, so free */
                    foundType = c_type(c_metaBind(c_metaObject(base),
                                                  name,
                                                  c_metaObject(keyInstanceType)));
                    if (foundType == NULL) {
                        OS_REPORT(OS_ERROR,
                                  "v_index::createInstanceType",V_RESULT_INTERNAL_ERROR,
                                  "Could not create key instance type");
                    }
                    os_freea(name);
                    c_free(keyInstanceType);
                }
                c_free(keyType);
            }
            c_free(instanceType);
            c_free(baseType);
        } else {
            foundType = baseType; /* transfer refCount to caller */
        }
        c_free(sampleType);
    }

    return foundType;
}

static c_bool
indexCompare(
    void *o,
    c_iterActionArg arg)
{
    v_index index = v_index(o);
    v_topic topic = v_topic(arg);
    v_topic currentTopic = NULL;

    if (index->entry) {
        currentTopic = v_dataReaderEntryTopic(index->entry);
    }
    return (currentTopic == topic);
}

#define PREFIX "sample.message"
static c_array
createKeyList(
    c_type instanceType,
    c_array keyList)
{
    c_ulong size, i;
    c_array newKeyList = NULL;

    assert(instanceType);

    if (keyList) {
        size = c_arraySize(keyList);

        newKeyList = c_arrayNew(c_field_t(c_getBase(instanceType)), size);

        if (newKeyList) {
            for(i = 0; i<size; i++){
                c_field f = c_fieldNew(instanceType, PREFIX);
                assert(f);
                if (f) {
                    newKeyList[i] = c_fieldConcat(f, keyList[i]);
                    c_free(f);
                } else {
                    OS_REPORT(OS_CRITICAL,
                                "createKeyList", V_RESULT_INTERNAL_ERROR,
                                "Could not create c_field");
                }
            }
        } else {
            OS_REPORT(OS_FATAL,
                        "createKeyList", V_RESULT_INTERNAL_ERROR,
                        "Could not create array");
        }
    }
    return newKeyList;
}
#undef PREFIX

void
v_indexInit(
    v_index index,
    c_type instanceType,
    c_array keyList,
    v_reader reader)
{
    c_property keyProperty;
    c_structure keyStructure;
    c_char fieldName[16];
    c_char *keyExpr;
    c_size i,nrOfKeys,totalSize;

    assert(index != NULL);
    assert(C_TYPECHECK(index,v_index));
    assert(C_TYPECHECK(instanceType,c_type));

    keyProperty = c_property(c_metaResolve(c_metaObject(instanceType),"key"));
    if (keyProperty) {
        keyStructure = c_structure(keyProperty->type);
        nrOfKeys = c_arraySize(keyStructure->members);
        c_free(keyProperty);
    } else {
        nrOfKeys = 0;
    }

    if (nrOfKeys>0) {
        totalSize = nrOfKeys * strlen("key.field0,");
        if (nrOfKeys > 9) {
            /* For each key number greater than one digit
             * i.e. number of keys > 9 add one additional
             * character space to the total size.
             */
            totalSize += (nrOfKeys-9);
            if (nrOfKeys > 99) {
                /* For each key number greater than two digits
                 * i.e. number of keys > 99 add one additional
                 * character space to the total size.
                 */
                totalSize += (nrOfKeys-99);
            }
        }
        keyExpr = (char *)os_alloca(totalSize);
        keyExpr[0] = 0;
        for (i=0;i<nrOfKeys;i++) {
            os_sprintf(fieldName,"key.field%d",i);
            os_strcat(keyExpr,fieldName);
            if (i<(nrOfKeys-1)) { os_strcat(keyExpr,","); }
        }
    } else {
        keyExpr = NULL;
    }

    index->reader = reader;
    index->sourceKeyList = createKeyList(instanceType, keyList);
    index->messageKeyList = c_keep(keyList);    /* keyList is either topic->messageKeyList or a user-defined keylist */
    index->objects = c_tableNew(instanceType,keyExpr);
    index->notEmptyList = c_tableNew(instanceType,keyExpr);

    if(keyExpr){
        os_freea(keyExpr);
    }
    index->objectType = c_keep(instanceType);
}

v_index
v__indexNew(
    v_dataReader reader,
    q_expr _from,
    c_iter indexList,
    v_indexNewAction action,
    c_voidp arg)
{
    v_kernel kernel;
    v_index index;
    v_topic topic;
    c_type instanceType;
    c_iter list;
    c_char *keyExpr;
    c_array keyList;
    c_ulong nrOfTopics;

    assert(C_TYPECHECK(reader,v_dataReader));
    kernel = v_objectKernel(reader);

    if (q_isId(_from)) {
        list = v_resolveTopics(kernel,q_getId(_from));
        nrOfTopics = c_iterLength(list);
        if (nrOfTopics == 0) {
            OS_REPORT(OS_ERROR,
                        "v__indexNew", V_RESULT_ILL_PARAM,
                        "Unknown topic %s",
                        q_getId(_from));
            c_iterFree(list);
            return NULL;
        }
        if (nrOfTopics > 1) {
            OS_REPORT(OS_ERROR,
                        "v__indexNew", V_RESULT_ILL_PARAM,
                        "Multiple topic definitions of: %s",
                        q_getId(_from));
            topic = v_topic(c_iterTakeFirst(list));
            while (topic != NULL) {
                c_free(topic);
                topic = v_topic(c_iterTakeFirst(list));
            }
            c_iterFree(list);
            return NULL;
        }
        topic = c_iterTakeFirst(list);
        c_iterFree(list);
        index = v_index(c_iterReadAction(indexList, indexCompare, topic));
        if (index == NULL) {
            /* If the userKey is enabled then the instance type key field type
             * will be determined from the user key expression and topic.
             * Otherwise when no user key is specified the default Topic key
             * type will be used.
             */
            if (v_reader(reader)->qos->userKey.v.enable) {
                keyExpr = v_reader(reader)->qos->userKey.v.expression;
            } else {
                keyExpr = NULL;
            }
            instanceType = createInstanceType(topic,keyExpr,&keyList);
            if (instanceType) {
                index = v_index(v_objectNew(kernel,K_INDEX));
                v_indexInit(index, instanceType, keyList, v_reader(reader));
            }
            c_free(keyList);
            c_free(instanceType);

            if (index != NULL) {
                if (action != NULL && !action(index, topic, arg)) {
                    OS_REPORT(OS_ERROR,
                        "v_indexNew", V_RESULT_INTERNAL_ERROR,
                        "v_indexNewAction failed!");
                    c_free(index);
                    index = NULL;
                } else {
                    (void)c_iterAppend(indexList, index);
                }
            }
        }
    } else {
        OS_REPORT(OS_ERROR,
                  "v_indexNew failed",V_RESULT_ILL_PARAM,
                  "illegal from clause specified");
        assert(FALSE);
        index = NULL;
    }
    return index;
}

v_index
v_indexNew(
    v_dataReader reader,
    q_expr expr,
    v_indexNewAction action,
    c_voidp arg)
{
    v_index index;
    c_iter indexList;
    void *ptr;

    assert(C_TYPECHECK(reader,v_dataReader));

    indexList = c_iterNew(NULL);
    index = v__indexNew(reader, expr, indexList, action, arg);

    /* Clean up iterator */
    do {
        ptr = c_iterTakeFirst(indexList);
    } while (ptr != NULL);
    c_iterFree(indexList);

    assert(C_TYPECHECK(index,v_index));
    return index;
}

c_bool
v_indexWalk(
    v_index _this,
    c_action action,
    c_voidp arg)
{
    v_reader reader;
    c_bool result;
    assert(C_TYPECHECK(_this,v_index));

    reader = v_reader(_this->reader);
    if (reader->qos->userKey.v.enable) {
        result = c_tableWalk(_this->notEmptyList, action, arg);
    } else {
        result = c_tableWalk(_this->objects, action, arg);
    }
    return result;
}

