Subscriptions.cpp

Go to the documentation of this file.

#include "Subscriptions.h"
#include "PsyTime.h"
#include "UnitTestFramework.h"
 
namespace cmlabs {
 
 
RetrieveSpec* TriggerSpec::getRetrieveSpec(const char* name) {
    if (!name || !strlen(name) || !retrieveCount)
        return NULL;
 
    RetrieveSpec* spec = (RetrieveSpec*) (((char*)this) +
        sizeof(TriggerSpec) + (filterCount*sizeof(FilterSpec)));
 
    uint32 n = 0;
    for (n = 0; n < retrieveCount; n++) {
        if (stricmp(name, spec->name) == 0)
            return spec;
        spec++;
    }
    return NULL;
}
 
QuerySpec* TriggerSpec::getQuerySpec(const char* name) {
    if (!name || !strlen(name) || !queryCount)
        return NULL;
 
    QuerySpec* spec = (QuerySpec*) (((char*)this) +
        sizeof(TriggerSpec) + (filterCount*sizeof(FilterSpec)) +
        (retrieveCount*sizeof(RetrieveSpec)));
 
    uint32 n = 0;
    for (n = 0; n < queryCount; n++) {
        if (stricmp(name, spec->name) == 0)
            return spec;
        spec++;
    }
    return NULL;
}
 
std::list<PostSpec*>* TriggerSpec::getPostSpecs(const char* name) {
    std::list<PostSpec*>* postSpecs = new std::list<PostSpec*>;
    uint32 n = 0;
    PostSpec* spec = (PostSpec*) (((char*)this) +
        sizeof(TriggerSpec) +
        (filterCount*sizeof(FilterSpec)) +
        (retrieveCount*sizeof(RetrieveSpec)) +
        (queryCount*sizeof(QuerySpec)));
 
    for (n = 0; n < postCount; n++) {
        if (!name || !strlen(name) || stricmp(name, spec->name) == 0)
            postSpecs->push_back(spec);
        spec++;
    }
    return postSpecs;
}
 
std::list<PostSpec*>* TriggerSpec::getPostSpecs(PsyType type) {
    std::list<PostSpec*>* postSpecs = new std::list<PostSpec*>;
    if (!type.isValid())
        return postSpecs;
    uint32 n = 0;
    PostSpec* spec = (PostSpec*) (((char*)this) +
        sizeof(TriggerSpec) +
        (filterCount*sizeof(FilterSpec)) +
        (retrieveCount*sizeof(RetrieveSpec)) +
        (queryCount*sizeof(QuerySpec)));
 
    for (n = 0; n < postCount; n++) {
        if (type.matches(spec->type))
            postSpecs->push_back(spec);
        spec++;
    }
    return postSpecs;
}
    
PostSpec* TriggerSpec::getPostSpec(const char* name) {
    uint32 n = 0;
    PostSpec* spec = (PostSpec*) (((char*)this) +
        sizeof(TriggerSpec) +
        (filterCount*sizeof(FilterSpec)) +
        (retrieveCount*sizeof(RetrieveSpec)) +
        (queryCount*sizeof(QuerySpec)));
 
    for (n = 0; n < postCount; n++) {
        if (stricmp(name, spec->name) == 0)
            return spec;
        spec++;
    }
    return NULL;
}
 
 
PostSpec* TriggerSpec::getPostSpec(PsyType type) {
 
    if (!type.isValid())
        return NULL;
 
    uint32 n = 0;
    PostSpec* spec = (PostSpec*) (((char*)this) +
        sizeof(TriggerSpec) +
        (filterCount*sizeof(FilterSpec)) +
        (retrieveCount*sizeof(RetrieveSpec)) +
        (queryCount*sizeof(QuerySpec)));
 
    for (n = 0; n < postCount; n++) {
        if (type.matches(spec->type))
            return spec;
        spec++;
    }
    return NULL;
}
 
SignalSpec* TriggerSpec::getSignalSpec(const char* name) {
    uint32 n = 0;
    SignalSpec* spec = (SignalSpec*) (((char*)this) +
        sizeof(TriggerSpec) +
        (filterCount*sizeof(FilterSpec)) +
        (retrieveCount*sizeof(RetrieveSpec)) +
        (queryCount*sizeof(QuerySpec)) +
        (postCount*sizeof(PostSpec)));
 
    for (n = 0; n < signalCount; n++) {
        if (stricmp(name, spec->name) == 0)
            return spec;
        spec++;
    }
    return NULL;
}
 
 
SignalSpec* TriggerSpec::getSignalSpec(PsyType type) {
 
    if (!type.isValid())
        return NULL;
 
    uint32 n = 0;
    SignalSpec* spec = (SignalSpec*) (((char*)this) +
        sizeof(TriggerSpec) +
        (filterCount*sizeof(FilterSpec)) +
        (retrieveCount*sizeof(RetrieveSpec)) +
        (queryCount*sizeof(QuerySpec)) +
        (postCount*sizeof(PostSpec)));
 
    for (n = 0; n < signalCount; n++) {
        if (type.matches(spec->type))
            return spec;
        spec++;
    }
    return NULL;
}
 
 
 
std::string TriggerSpec::toXML() {
    uint32 n;
    std::string str = utils::StringFormat("<trigger name=\"%s\" type=\"%s\" compid=\"%u\" interval=\"%u\" delay=\"%u\" from=\"%u\" to=\"%u\" tag=\"%u\" filtercount=\"%u\" signalcount=\"%u\" postcount=\"%u\" context=\"%s\" triggercontext=\"%s\"",
        name, type.toString().c_str(), componentID, interval, delay, from, to, tag, filterCount, signalCount, postCount, context.toString().c_str(), triggerContext.toString().c_str());
 
    if (!retrieveCount && !queryCount && !filterCount && !signalCount && !postCount) {
        str += " />\n";
        return str;
    }
    else
        str += ">\n";
 
    for (n=0; n<retrieveCount; n++) {
        RetrieveSpec* rSpec = getRetrieveSpec(n);
        if (rSpec)
            str += rSpec->toXML();
    }
 
    for (n=0; n<signalCount; n++) {
        SignalSpec* sSpec = getSignalSpec(n);
        if (sSpec)
            str += sSpec->toXML();
    }
 
    for (n=0; n<queryCount; n++) {
        QuerySpec* qSpec = getQuerySpec(n);
        if (qSpec)
            str += qSpec->toXML();
    }
 
    for (n=0; n<postCount; n++) {
        PostSpec* pSpec = getPostSpec(n);
        if (pSpec)
            str += pSpec->toXML();
    }
 
    str += "</trigger>\n";
    return str;
 
}
 
 
// ################# Unit test #################
 
bool TriggerSpec::UnitTest() {
 
    // A TriggerSpec is a variable-length packed structure: a fixed header
    // followed by filterCount FilterSpecs, retrieveCount RetrieveSpecs,
    // queryCount QuerySpecs, postCount PostSpecs and signalCount SignalSpecs.
    // We build one in a flat buffer, populate the sub-specs, then exercise the
    // index / name / type accessors and the XML serialisation.
 
    // 1. FilterSpec / RetrieveSpec / QuerySpec / PostSpec / SignalSpec XML round-trip
    unittest::progress(10, "spec XML round-trip");
    {
        FilterSpec f;
        memset(&f, 0, sizeof(FilterSpec));
        f.filter = TRIGGERFILTER_EQUALS;
        f.time = 123ULL;
        utils::strcpyavail(f.key, "fkey", MAXKEYNAMELEN, true);
        utils::strcpyavail(f.value, "fval", MAXVALUENAMELEN, true);
        std::string fxml = f.toXML();
        if (fxml.empty()) { unittest::fail("Subscriptions test: FilterSpec toXML empty\n"); return false; }
        FilterSpec f2;
        memset(&f2, 0, sizeof(FilterSpec));
        if (!f2.fromXML(fxml.c_str())) { unittest::fail("Subscriptions test: FilterSpec fromXML failed\n"); return false; }
        if (f2.filter != f.filter || f2.time != f.time ||
            strcmp(f2.key, "fkey") != 0 || strcmp(f2.value, "fval") != 0) {
            unittest::fail("Subscriptions test: FilterSpec round-trip mismatch\n"); return false;
        }
 
        SignalSpec s;
        memset(&s, 0, sizeof(SignalSpec));
        s.type.fromString("3.4");
        utils::strcpyavail(s.name, "sig", MAXKEYNAMELEN, true);
        std::string sxml = s.toXML();
        if (sxml.empty()) { unittest::fail("Subscriptions test: SignalSpec toXML empty\n"); return false; }
        SignalSpec s2;
        memset(&s2, 0, sizeof(SignalSpec));
        if (!s2.fromXML(sxml.c_str())) { unittest::fail("Subscriptions test: SignalSpec fromXML failed\n"); return false; }
        if (strcmp(s2.name, "sig") != 0 || s2.type != s.type) {
            unittest::fail("Subscriptions test: SignalSpec round-trip mismatch\n"); return false;
        }
    }
 
    // 2. PostSpec content build + addContentToMsg
    unittest::progress(25, "post content -> message");
    {
        PostSpec p;
        memset(&p, 0, sizeof(PostSpec));
        p.to = 7;
        p.type.fromString("1.2.3");
        utils::strcpyavail(p.name, "post1", MAXKEYNAMELEN, true);
        // content layout: ||<type>||<key>||<value>... with contentCount entries
        utils::strcpyavail(p.content, "||S||s_key||s_val||I||i_key||42||F||f_key||1.5", MAXCONTENTLEN, true);
        p.contentCount = 3;
 
        std::string pxml = p.toXML();
        if (pxml.empty() || pxml.find("post1") == std::string::npos) {
            unittest::fail("Subscriptions test: PostSpec toXML missing name\n"); return false;
        }
 
        DataMessage msg(CTRL_TEST, 0);
        uint32 added = p.addContentToMsg(&msg);
        if (added != 3) { unittest::fail("Subscriptions test: addContentToMsg count != 3\n"); return false; }
        if (msg.getAsString("s_key") != "s_val") { unittest::fail("Subscriptions test: addContentToMsg string mismatch\n"); return false; }
        int64 iv = 0;
        if (!msg.getInt("i_key", iv) || iv != 42) { unittest::fail("Subscriptions test: addContentToMsg int mismatch\n"); return false; }
        double dv = 0;
        if (!msg.getDouble("f_key", dv) || dv < 1.4 || dv > 1.6) { unittest::fail("Subscriptions test: addContentToMsg float mismatch\n"); return false; }
    }
 
    // 3. Build a packed TriggerSpec and exercise the accessors
    unittest::progress(45, "packed trigger accessors");
    {
        const uint16 nFilter   = 1;
        const uint16 nRetrieve = 1;
        const uint16 nQuery    = 1;
        const uint16 nPost     = 2;
        const uint16 nSignal   = 2;
 
        uint32 total = sizeof(TriggerSpec)
            + nFilter   * sizeof(FilterSpec)
            + nRetrieve * sizeof(RetrieveSpec)
            + nQuery    * sizeof(QuerySpec)
            + nPost     * sizeof(PostSpec)
            + nSignal   * sizeof(SignalSpec);
 
        char* buf = (char*)malloc(total);
        if (!buf) { unittest::fail("Subscriptions test: malloc failed\n"); return false; }
        memset(buf, 0, total);
 
        TriggerSpec* trig = (TriggerSpec*)buf;
        trig->reset(total);
        trig->id = 99;
        trig->componentID = 5;
        trig->type.fromString("10.20");
        utils::strcpyavail(trig->name, "mytrigger", MAXKEYNAMELEN, true);
        trig->filterCount   = nFilter;
        trig->retrieveCount = nRetrieve;
        trig->queryCount    = nQuery;
        trig->postCount     = nPost;
        trig->signalCount   = nSignal;
 
        // Populate sub-specs at the offsets the accessors expect.
        FilterSpec* filt = (FilterSpec*)(buf + sizeof(TriggerSpec));
        filt[0].filter = TRIGGERFILTER_HASKEY;
        utils::strcpyavail(filt[0].key, "fk", MAXKEYNAMELEN, true);
 
        RetrieveSpec* ret = (RetrieveSpec*)((char*)filt + nFilter * sizeof(FilterSpec));
        utils::strcpyavail(ret[0].name, "ret1", MAXKEYNAMELEN, true);
 
        QuerySpec* qry = (QuerySpec*)((char*)ret + nRetrieve * sizeof(RetrieveSpec));
        utils::strcpyavail(qry[0].name, "qry1", MAXKEYNAMELEN, true);
 
        PostSpec* posts = (PostSpec*)((char*)qry + nQuery * sizeof(QuerySpec));
        utils::strcpyavail(posts[0].name, "postA", MAXKEYNAMELEN, true);
        posts[0].type.fromString("1.2");
        utils::strcpyavail(posts[1].name, "postB", MAXKEYNAMELEN, true);
        posts[1].type.fromString("1.3");
 
        SignalSpec* sigs = (SignalSpec*)((char*)posts + nPost * sizeof(PostSpec));
        utils::strcpyavail(sigs[0].name, "sigA", MAXKEYNAMELEN, true);
        sigs[0].type.fromString("4.5");
        utils::strcpyavail(sigs[1].name, "sigB", MAXKEYNAMELEN, true);
        sigs[1].type.fromString("4.6");
 
        // --- index accessors return the same pointers we wrote ---
        if (trig->getRetrieveSpec((uint32)0) != ret) { unittest::fail("Subscriptions test: getRetrieveSpec(0) wrong ptr\n"); free(buf); return false; }
        if (trig->getQuerySpec((uint32)0) != qry) { unittest::fail("Subscriptions test: getQuerySpec(0) wrong ptr\n"); free(buf); return false; }
        if (trig->getPostSpec((uint32)0) != &posts[0]) { unittest::fail("Subscriptions test: getPostSpec(0) wrong ptr\n"); free(buf); return false; }
        if (trig->getPostSpec((uint32)1) != &posts[1]) { unittest::fail("Subscriptions test: getPostSpec(1) wrong ptr\n"); free(buf); return false; }
        if (trig->getSignalSpec((uint32)0) != &sigs[0]) { unittest::fail("Subscriptions test: getSignalSpec(0) wrong ptr\n"); free(buf); return false; }
        if (trig->getSignalSpec((uint32)1) != &sigs[1]) { unittest::fail("Subscriptions test: getSignalSpec(1) wrong ptr\n"); free(buf); return false; }
 
        // --- out-of-range index returns NULL ---
        if (trig->getRetrieveSpec(nRetrieve) != NULL) { unittest::fail("Subscriptions test: getRetrieveSpec out-of-range not NULL\n"); free(buf); return false; }
        if (trig->getQuerySpec(nQuery) != NULL) { unittest::fail("Subscriptions test: getQuerySpec out-of-range not NULL\n"); free(buf); return false; }
        if (trig->getPostSpec(nPost) != NULL) { unittest::fail("Subscriptions test: getPostSpec out-of-range not NULL\n"); free(buf); return false; }
        if (trig->getSignalSpec(nSignal) != NULL) { unittest::fail("Subscriptions test: getSignalSpec out-of-range not NULL\n"); free(buf); return false; }
 
        // --- name accessors ---
        if (trig->getRetrieveSpec("ret1") != ret) { unittest::fail("Subscriptions test: getRetrieveSpec(name) miss\n"); free(buf); return false; }
        if (trig->getRetrieveSpec("nope") != NULL) { unittest::fail("Subscriptions test: getRetrieveSpec(bad name) not NULL\n"); free(buf); return false; }
        if (trig->getQuerySpec("qry1") != qry) { unittest::fail("Subscriptions test: getQuerySpec(name) miss\n"); free(buf); return false; }
        if (trig->getQuerySpec("nope") != NULL) { unittest::fail("Subscriptions test: getQuerySpec(bad name) not NULL\n"); free(buf); return false; }
        if (trig->getPostSpec("postB") != &posts[1]) { unittest::fail("Subscriptions test: getPostSpec(name) miss\n"); free(buf); return false; }
        if (trig->getPostSpec("nope") != NULL) { unittest::fail("Subscriptions test: getPostSpec(bad name) not NULL\n"); free(buf); return false; }
        if (trig->getSignalSpec("sigA") != &sigs[0]) { unittest::fail("Subscriptions test: getSignalSpec(name) miss\n"); free(buf); return false; }
        if (trig->getSignalSpec("nope") != NULL) { unittest::fail("Subscriptions test: getSignalSpec(bad name) not NULL\n"); free(buf); return false; }
        // name lookup is case-insensitive (stricmp)
        if (trig->getPostSpec("POSTA") != &posts[0]) { unittest::fail("Subscriptions test: getPostSpec name not case-insensitive\n"); free(buf); return false; }
 
        // --- type accessors (PsyType::matches) ---
        PsyType t12; t12.fromString("1.2");
        PsyType t13; t13.fromString("1.3");
        PsyType t99; t99.fromString("9.9");
        if (trig->getPostSpec(t12) != &posts[0]) { unittest::fail("Subscriptions test: getPostSpec(type 1.2) miss\n"); free(buf); return false; }
        if (trig->getPostSpec(t13) != &posts[1]) { unittest::fail("Subscriptions test: getPostSpec(type 1.3) miss\n"); free(buf); return false; }
        if (trig->getPostSpec(t99) != NULL) { unittest::fail("Subscriptions test: getPostSpec(type 9.9) not NULL\n"); free(buf); return false; }
        PsyType t45; t45.fromString("4.5");
        if (trig->getSignalSpec(t45) != &sigs[0]) { unittest::fail("Subscriptions test: getSignalSpec(type 4.5) miss\n"); free(buf); return false; }
 
        // --- getPostSpecs list variants ---
        std::list<PostSpec*>* all = trig->getPostSpecs((const char*)NULL);
        if (!all || all->size() != nPost) { unittest::fail("Subscriptions test: getPostSpecs(NULL) wrong count\n"); delete all; free(buf); return false; }
        delete all;
 
        std::list<PostSpec*>* named = trig->getPostSpecs("postA");
        if (!named || named->size() != 1 || named->front() != &posts[0]) {
            unittest::fail("Subscriptions test: getPostSpecs(name) wrong\n"); delete named; free(buf); return false;
        }
        delete named;
 
        // wildcard match in PsyType: a post whose type matches 0xFFFF wildcard
        PsyType wild; wild.fromString("1");
        wild.setLevel(1, 0xFFFF); // 1.* matches both 1.2 and 1.3
        std::list<PostSpec*>* byType = trig->getPostSpecs(wild);
        if (!byType || byType->size() != 2) {
            unittest::fail("Subscriptions test: getPostSpecs(wildcard type) wrong count\n"); delete byType; free(buf); return false;
        }
        delete byType;
 
        // --- toXML produces a self-describing trigger document ---
        std::string xml = trig->toXML();
        if (xml.empty() || xml.find("mytrigger") == std::string::npos) {
            unittest::fail("Subscriptions test: TriggerSpec toXML missing name\n"); free(buf); return false;
        }
        if (xml.find("postA") == std::string::npos || xml.find("sigA") == std::string::npos) {
            unittest::fail("Subscriptions test: TriggerSpec toXML missing sub-specs\n"); free(buf); return false;
        }
 
        // --- lookup throughput metric ---
        unittest::progress(80, "lookup throughput");
        const uint32 iterations = 200000;
        uint64 t0 = GetTimeNow();
        volatile uint64 sink = 0;
        for (uint32 i = 0; i < iterations; i++) {
            if (trig->getPostSpec("postB")) sink++;
            if (trig->getSignalSpec("sigB")) sink++;
        }
        double us = (double)(GetTimeNow() - t0);
        if (us > 0.0)
            unittest::metric("name_lookup", (double)(iterations * 2) / us * 1e6, "ops/s", true);
        unittest::detail("trigger name lookups: %u in %.0f us\n", iterations * 2, us);
 
        free(buf);
    }
 
    unittest::progress(100, "done");
    return true;
}
 
void Register_Subscriptions_Tests() {
    UnitTestRunner::instance().registerTest("subscriptions", TriggerSpec::UnitTest,
        "Trigger/filter/retrieve/query/post/signal spec lookup and XML", "core");
}
 
} // namespace cmlabs