[MOD/WIP] Threading, and bot "API".

Sloppy code!

XEPS-TBD.TXT

@@ -2,7 +2,8 @@ XEPs current supported are in src/XMPPThread.c, at the IQ disco advertising.
 
 Somewhat implemented XEPs:
     ~ https://xmpp.org/extensions/xep-0085.html
-        Only XMPP->Matrix
+        Only XMPP->Matrix at the moment. Still need to figure out how to 
+        get typing indicators as an AS.
     ~ https://xmpp.org/extensions/xep-0444.html
         This allows reactions, which Matrix also has support to. The two 
         systems don't seem *too* restrictive on one-another (unlike some 

src/Commands/Help.c

@@ -5,6 +5,7 @@
 #include <Cytoplasm/Str.h>
 
 #include <Matrix.h>
+#include <Bot.h>
 #include <AS.h>
 
 CommandHead(CmdHelp, cmd, argp)
@@ -12,31 +13,16 @@ CommandHead(CmdHelp, cmd, argp)
     ParseeCmdArg *args = argp;
     ParseeData *data = args->data;
     HashMap *json, *event = args->event;
-    char *profile = StrConcat(4,
+    BotInitialise();
-        "@", data->config->sender_localpart, 
-        ":", data->config->homeserver_host
-    );
-    char *id = GrabString(event, 1, "room_id");
-    char *msg;
 
-    Free(ASSend(
+    ReplyBasic("Commands: ");
-        data->config, id, profile, 
+#define X_COMMAND(path, name, desc) ReplySprintf("- %s: %s", path, desc);
-        "m.room.message",
-        MatrixCreateNotice("Commands: ")
-    ));
-
-#define X_COMMAND(path, name, desc) { \
-                                        msg = StrConcat(3, path, "-", desc);\
-                                        Free(ASSend( \
-                                            data->config, id, profile,  \
-                                            "m.room.message", \
-                                            MatrixCreateNotice(msg) \
-                                        )); \
-                                        Free(msg); \
-                                    } \
-                                    while (0);
     COMMANDS
 #undef X_COMMAND
 
-    Free(profile);
+    ReplySprintf("- Source code and licensing information: %s",
+        REPOSITORY
+    );
+    ReplyBasic("*Written with a shoelace and UHU glue by LDA <3 !*");
+    BotDestroy();
 }

src/Commands/ListBans.c

@@ -4,6 +4,7 @@
 #include <Cytoplasm/Str.h>
 
 #include <Matrix.h>
+#include <Bot.h>
 #include <AS.h>
 
 #include <stdlib.h>
@@ -14,36 +15,21 @@ CommandHead(CmdListBans, cmd, argp)
     ParseeData *data = args->data;
     HashMap *json, *event = args->event;
     DbRef *listed;
-    char *str = NULL, *tmp = NULL, *banned;
+    char *banned;
-    char *profile = StrConcat(4,
-        "@", data->config->sender_localpart, 
-        ":", data->config->homeserver_host
-    );
-    char *id = GrabString(event, 1, "room_id");
     JsonValue *val;
 
+    BotInitialise();
+
     listed = DbLock(data->db, 1, "global_bans");
     json = DbJson(listed);
 
+    ReplyBasic("Users on the no-fly list:");
     while (HashMapIterate(json, &banned, (void **) &val))
     {
-        tmp = str;
+        /* TODO: Add no-fly reasons */
-        str = StrConcat(4, str, "- ", banned, "\n");
+        ReplySprintf("- %s", banned);
-        Free(tmp);
     }
 
-    Free(ASSend(
-        data->config, id, profile, 
-        "m.room.message",
-        MatrixCreateNotice("No-fly listed users:")
-    ));
-    Free(ASSend(
-        data->config, id, profile, 
-        "m.room.message",
-        MatrixCreateNotice(str)
-    ));
-
-    Free(str);
     DbUnlock(data->db, listed);
-    Free(profile);
+    BotDestroy();
 }

src/Commands/Stats.c

@@ -5,6 +5,7 @@
 #include <Cytoplasm/Str.h>
 
 #include <Matrix.h>
+#include <Bot.h>
 #include <AS.h>
 
 CommandHead(CmdStats, cmd, argp)
@@ -12,11 +13,6 @@ CommandHead(CmdStats, cmd, argp)
     ParseeCmdArg *args = argp;
     ParseeData *data = args->data;
     HashMap *json, *event = args->event;
-    char *profile = StrConcat(4,
-        "@", data->config->sender_localpart, 
-        ":", data->config->homeserver_host
-    );
-    char *id = GrabString(event, 1, "room_id");
     char *msg;
     size_t alloc = MemoryAllocated();
     size_t kb = alloc   >> 10;
@@ -24,30 +20,22 @@ CommandHead(CmdStats, cmd, argp)
     size_t gb = mb      >> 10;
     size_t min = gb ? gb : (mb ? mb : (kb ? kb : alloc));
     char *unit = gb ? "GB" : (mb ? "MB" : (kb ? "KB" : "B"));
-    char *l = StrInt(min);
 
-    msg = StrConcat(3, 
+    BotInitialise();
-        "Information for " NAME " (",
+
-        CytoplasmGetVersionStr(), ")"
+
+    /* TODO: Separate these into different "categories" */
+    ReplySprintf("Information for %s v%s (Cytoplasm %s)",
+        NAME, VERSION, CytoplasmGetVersionStr()
     );
-    Free(ASSend(
-        data->config, id, profile, 
-        "m.room.message",
-        MatrixCreateNotice(msg)
-    ));
-    Free(msg);
 
-    msg = StrConcat(5, 
+    ReplySprintf("- Memory used: %d%s (reported by Cytoplasm)",
-        "- Memory usage: ", l, " ", unit,
+        min, unit
-        " (reported by Cytoplasm)"
     );
-    Free(ASSend(
+    ReplySprintf("- Source code and licensing information: %s",
-        data->config, id, profile, 
+        REPOSITORY
-        "m.room.message",
+    );
-        MatrixCreateNotice(msg)
+    ReplyBasic("*Written with a shoelace and UHU glue by LDA <3 !*");
-    ));
-    Free(msg);
-    Free(l);
 
-    Free(profile);
+    BotDestroy();
 }

src/Parsee/Data.c

@@ -433,6 +433,9 @@ ParseePushDMStanza(ParseeData *data, char *room_id, char *stanza_id, char *id, c
     }
     j = DbJson(ref);
 
+
+    if (stanza_id)
+    {
         stanzas = JsonValueAsObject(HashMapGet(j, "stanzas"));
         if (!stanzas)
         {
@@ -440,8 +443,6 @@ ParseePushDMStanza(ParseeData *data, char *room_id, char *stanza_id, char *id, c
             new_stanzas = true;
         }
 
-    if (stanza_id)
-    {
         obj = HashMapCreate();
         HashMapSet(obj, "age", JsonValueInteger(age));
         HashMapSet(obj, "event", JsonValueString(ev));

src/XMPP/Stanza.c

@@ -124,8 +124,6 @@ XMPPSendPlainID(XMPPComponent *comp, char *fr, char *to, char *msg, char *type,
     XMLAddChild(body, data);
 
     XMLEncode(comp->stream, message);
-    XMLEncode(StreamStdout(), message);
-    Log(LOG_INFO, "");
     StreamFlush(comp->stream);
     XMLFreeElement(message);
     Free(from);

src/XMPPThread.c

@@ -5,6 +5,7 @@
 
 #include <Cytoplasm/Memory.h>
 #include <Cytoplasm/Base64.h>
+#include <Cytoplasm/Util.h>
 #include <Cytoplasm/Log.h>
 #include <Cytoplasm/Str.h>
 #include <Cytoplasm/Sha.h>
@@ -269,11 +270,30 @@ ParseeVerifyAllStanza(ParseeData *args, XMLElement *stanza)
 
     return ret;
 }
+struct XMPPThread;
+typedef struct XMPPThreadInfo {
+    /* A FIFO of stanzas */
+    Array *stanzas;
+    pthread_mutex_t lock;
+
+    ParseeData *args;
+    XMPPComponent *jabber;
+
+    struct XMPPThread *dispatchers;
+    size_t available_dispatchers;
+
+    bool running;
+    pthread_mutex_t chk_lock;
+} XMPPThreadInfo;
+typedef struct XMPPThread {
+    pthread_t thr;
+    XMPPThreadInfo *info;
+} XMPPThread;
 
 /* TODO: Clean up all of this. We are currently separating DMs from MUCs, 
  * where we could unify all our code, and generalise everything. */
 static bool
-MessageStanza(ParseeData *args, XMLElement *stanza)
+MessageStanza(ParseeData *args, XMLElement *stanza, XMPPThread *thr)
 {
     XMPPComponent *jabber = args->jabber;
 
@@ -282,9 +302,14 @@ MessageStanza(ParseeData *args, XMLElement *stanza)
     XMLElement *data = NULL;
 
     char *to, *room, *from, *from_matrix, *decode_from;
-    char *chat_id, *mroom_id;
+    char *chat_id = NULL, *mroom_id = NULL;
     size_t i;
 
+    to = NULL;
+    from = NULL;
+    decode_from = NULL;
+    from_matrix = NULL;
+
     from = HashMapGet(stanza->attrs, "from");
 
 #define CHAT_STATES "http://jabber.org/protocol/chatstates"
@@ -292,29 +317,31 @@ MessageStanza(ParseeData *args, XMLElement *stanza)
     {
         decode_from = ParseeLookupJID(from);
         from_matrix = ParseeEncodeJID(args->config, decode_from, true);
-        chat_id = ParseeGetFromMUCID(args, from);
+        mroom_id = ParseeGetBridgedRoom(args, stanza);
-        mroom_id = ParseeGetRoomID(args, chat_id);
         
         ASType(args->config, from_matrix, mroom_id, true);
 
         Free(decode_from);
         Free(from_matrix);
         Free(mroom_id);
-        Free(chat_id);
+        mroom_id = NULL;
+        decode_from = NULL;
+        from_matrix = NULL;
     }
     else if (XMLookForTKV(stanza, "active", "xmlns", CHAT_STATES))
     {
         decode_from = ParseeLookupJID(from);
         from_matrix = ParseeEncodeJID(args->config, decode_from, true);
-        chat_id = ParseeGetFromMUCID(args, from);
+        mroom_id = ParseeGetBridgedRoom(args, stanza);
-        mroom_id = ParseeGetRoomID(args, chat_id);
         
         ASType(args->config, from_matrix, mroom_id, false);
 
         Free(decode_from);
         Free(from_matrix);
         Free(mroom_id);
-        Free(chat_id);
+        mroom_id = NULL;
+        decode_from = NULL;
+        from_matrix = NULL;
     }
 #undef CHAT_STATES
     body = XMLookForUnique(stanza, "body");
@@ -335,6 +362,19 @@ MessageStanza(ParseeData *args, XMLElement *stanza)
     room = ParseeFindDMRoom(args, to, from);
     data = ArrayGet(body->children, 0);
 
+    /* TODO: THIS IS A HACK. THIS CODE IGNORES ALL MUC MESSAGES EVER 
+     * SENT TO NON-PARSEE PUPPETS, AS A "FIX" TO THE MULTITHREADED 
+     * ISSUE. 
+     *
+     * I HATE THIS. I NEED TO FIND A BETTER WAY. */
+    if (!room)
+    {
+        if (strncmp(HashMapGet(stanza->attrs, "to"), "parsee@", 7))
+        {
+            goto end;
+        }
+    }
+
     mroom_id = ParseeGetBridgedRoom(args, stanza);
     if (mroom_id && !XMPPIsParseeStanza(stanza))
     {
@@ -355,10 +395,12 @@ MessageStanza(ParseeData *args, XMLElement *stanza)
             chat = true;
         }
 
+        pthread_mutex_lock(&thr->info->chk_lock);
         if (ParseeVerifyAllStanza(args, stanza) && !replaced)
         {
             XMLElement *oob, *oob_data;
 
+            pthread_mutex_unlock(&thr->info->chk_lock);
             ASRegisterUser(args->config, encoded);
             if (!chat)
             {
@@ -430,8 +472,10 @@ MessageStanza(ParseeData *args, XMLElement *stanza)
                     "m.room.message", ev
                 );
             }
+            pthread_mutex_lock(&thr->info->chk_lock);
             ParseePushAllStanza(args, stanza, event_id);
             Free(event_id);
+            pthread_mutex_unlock(&thr->info->chk_lock);
         }
         else if (replaced)
         {
@@ -441,10 +485,14 @@ MessageStanza(ParseeData *args, XMLElement *stanza)
                 "m.room.message", MatrixCreateReplace(event_id, data->data)
             ));
             ParseePushAllStanza(args, stanza, event_id);
+            pthread_mutex_unlock(&thr->info->chk_lock);
 
             Free(event_id);
         }
-
+        else
+        {
+            pthread_mutex_unlock(&thr->info->chk_lock);
+        }
         Free(res);
         Free(encoded);
     }
@@ -458,7 +506,9 @@ MessageStanza(ParseeData *args, XMLElement *stanza)
 
         ParseePushAllStanza(args, stanza, e_d->data);
     }
+end:
     Free(mroom_id);
+    mroom_id = NULL;
     Free(from_matrix);
     Free(decode_from);
     Free(room);
@@ -647,20 +697,38 @@ PresenceStanza(ParseeData *args, XMLElement *stanza)
 #undef MUC_USER_NS
 }
 
-void *
+static XMLElement *
-ParseeXMPPThread(void *argp)
+RetrieveStanza(XMPPThread *thread)
 {
-    ParseeData *args = argp;
+    XMLElement *ret = NULL;
-    XMPPComponent *jabber = args->jabber;
-    XMLElement *stanza = NULL;
-    while (true)
-    {
-        char *to, *room, *from, *from_matrix;
-        char *chat_id, *mroom_id;
 
-        stanza = XMLDecode(jabber->stream, false);
+    pthread_mutex_lock(&thread->info->lock);
+    ret = ArrayDelete(thread->info->stanzas, 0);
+    pthread_mutex_unlock(&thread->info->lock);
+
+    return ret;
+}
+static void
+PushStanza(XMPPThreadInfo *info, XMLElement *stanza)
+{
+    pthread_mutex_lock(&info->lock);
+    ArrayAdd(info->stanzas, stanza);
+    pthread_mutex_unlock(&info->lock);
+}
+
+static void *
+XMPPDispatcher(void *argp)
+{
+    XMPPThread *thread = argp;
+    ParseeData *args = thread->info->args;
+    XMPPComponent *jabber = thread->info->jabber;
+
+    while (thread->info->running)
+    {
+        XMLElement *stanza = RetrieveStanza(thread);
         if (!stanza)
         {
+            UtilSleepMillis(10);
             continue;
         }
 
@@ -674,28 +742,7 @@ ParseeXMPPThread(void *argp)
         else if (StrEquals(stanza->name, "message"))
         {
             size_t i;
-            if (XMPPIsKiller(stanza))
+            if (!MessageStanza(args, stanza, thread))
-            {
-                const char *killer = "killer";
-                const char *suspend="suspend";
-                from = HashMapGet(stanza->attrs, "from");
-                Log(LOG_INFO, "Killer.");
-                if (!strncmp(from, killer, strlen(killer)))
-                {
-                    XMLFreeElement(stanza);
-                    break;
-                }
-                else if (!strncmp(from, suspend, strlen(suspend)))
-                {
-                    XMLFreeElement(stanza);
-                    pthread_mutex_lock(&cond_var_lock);
-                    pthread_cond_wait(&cond_var, &cond_var_lock);
-                    pthread_mutex_unlock(&cond_var_lock);
-                    continue;
-                }
-            }
-
-            if (!MessageStanza(args, stanza))
             {
                 continue;
             }
@@ -713,6 +760,92 @@ ParseeXMPPThread(void *argp)
         }
         XMLFreeElement(stanza);
     }
+}
 
+void *
+ParseeXMPPThread(void *argp)
+{
+    ParseeData *args = argp;
+    XMPPComponent *jabber = args->jabber;
+    XMLElement *stanza = NULL;
+    XMPPThreadInfo info;
+    pthread_mutex_t stanzas_lock = PTHREAD_MUTEX_INITIALIZER;
+    size_t i;
+
+    /* Initialise the FIFO */
+    info.stanzas = ArrayCreate();
+    pthread_mutex_init(&info.lock, NULL);
+
+    /* TODO: Make that configurable. */
+    info.available_dispatchers = 8;
+    info.dispatchers = Malloc(
+        sizeof(*info.dispatchers) * info.available_dispatchers
+    );
+
+    info.args = args;
+    info.jabber = jabber;
+    info.running = true;
+    pthread_mutex_init(&info.chk_lock, NULL);
+
+    for (i = 0; i < info.available_dispatchers; i++)
+    {
+        pthread_t *thr = &info.dispatchers[i].thr;
+        info.dispatchers[i].info = &info;
+
+        pthread_create(thr, NULL, XMPPDispatcher, &info.dispatchers[i]);
+    }
+
+    while (true)
+    {
+        char *to, *room, *from, *from_matrix;
+        char *chat_id, *mroom_id;
+        ssize_t cntr;
+
+        stanza = XMLDecode(jabber->stream, false);
+        if (!stanza)
+        {
+            continue;
+        }
+
+        if (StrEquals(stanza->name, "message") && XMPPIsKiller(stanza))
+        {
+            const char *killer = "killer";
+            const char *suspend="suspend";
+            from = HashMapGet(stanza->attrs, "from");
+            if (!strncmp(from, killer, strlen(killer)))
+            {
+                XMLFreeElement(stanza);
+                break;
+            }
+            else if (!strncmp(from, suspend, strlen(suspend)))
+            {
+                /* TODO */
+                XMLFreeElement(stanza);
+                pthread_mutex_lock(&cond_var_lock);
+                pthread_cond_wait(&cond_var, &cond_var_lock);
+                pthread_mutex_unlock(&cond_var_lock);
+                continue;
+            }
+        }
+
+        PushStanza(&info, stanza);
+        //XMLFreeElement(stanza);
+    }
+
+    info.running = false;
+    for (i = 0; i < info.available_dispatchers; i++)
+    {
+        pthread_t thr = info.dispatchers[i].thr;
+        pthread_join(thr, NULL);
+    }
+    Free(info.dispatchers);
+
+    for (i = 0; i < ArraySize(info.stanzas); i++)
+    {
+        XMLFreeElement(ArrayGet(info.stanzas, i));
+    }
+    ArrayFree(info.stanzas);
+
+    pthread_mutex_destroy(&info.lock);
     return NULL;
 }

src/include/Bot.h

@@ -2,7 +2,49 @@
 #define PARSEE_BOT_H
 
 #include <Cytoplasm/HashMap.h>
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Log.h>
 
-#define BotInitialise()     profile = 
+#include <stdio.h>
+
+#include <Parsee.h>
+
+#define BotInitialise()     char *profile = StrConcat(4, \
+                                "@", data->config->sender_localpart, \
+                                ":", data->config->homeserver_host \
+                            ); \
+                            char *id = GrabString(event, 1, "room_id")
+
+#define BotDestroy()        Free(profile)
+
+#define ReplyBasic(rep)         do \
+                                { \
+                                    Free(ASSend( \
+                                        data->config, id, profile, \
+                                        "m.room.message", \
+                                        MatrixCreateNotice(rep) \
+                                    )); \
+                                } \
+                                while(0)
+
+#define ReplySprintf(fp,...)    do \
+                                { \
+                                    char *formatted = NULL; \
+                                    size_t fmt_len = 0; \
+                                    fmt_len = snprintf( \
+                                        NULL, 0, fp, __VA_ARGS__ \
+                                    ); \
+                                    formatted = Malloc(fmt_len + 2); \
+                                    snprintf( \
+                                       formatted, fmt_len + 1, fp, __VA_ARGS__ \
+                                    ); \
+                                    Free(ASSend( \
+                                        data->config, id, profile, \
+                                        "m.room.message", \
+                                        MatrixCreateNotice(formatted) \
+                                    )); \
+                                    Free(formatted); \
+                                } \
+                                while(0)
 
 #endif