mqttasync_module.c

Go to the documentation of this file.

#include <Python.h>

#include "MQTTAsync.h"
#include "LinkedList.h"

static PyObject *MqttV3Error;


static PyObject* mqttv3_create(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        char* serverURI;
        char* clientId;
        int persistence_option = MQTTCLIENT_PERSISTENCE_DEFAULT;
        PyObject *pyoptions = NULL, *temp;
        MQTTAsync_createOptions options = MQTTAsync_createOptions_initializer;
        int rc;

        if (!PyArg_ParseTuple(args, "ss|iO", &serverURI, &clientId,
                        &persistence_option, &pyoptions))
                return NULL;

        if (persistence_option != MQTTCLIENT_PERSISTENCE_DEFAULT
                        && persistence_option != MQTTCLIENT_PERSISTENCE_NONE)
        {
                PyErr_SetString(PyExc_TypeError, "persistence must be DEFAULT or NONE");
                return NULL;
        }

        if (pyoptions)
        {

                if (!PyDict_Check(pyoptions))
                {
                        PyErr_SetString(PyExc_TypeError,
                                        "Create options parameter must be a dictionary");
                        return NULL;
                }

                if ((temp = PyDict_GetItemString(pyoptions, "sendWhileDisconnected"))
                                != NULL)
                {
                        if (PyInt_Check(temp))
                                options.sendWhileDisconnected = (int) PyInt_AsLong(temp);
                        else
                        {
                                PyErr_SetString(PyExc_TypeError, "sendWhileDisconnected value must be int");
                                return NULL;
                        }
                }

                if ((temp = PyDict_GetItemString(pyoptions, "maxBufferedMessages"))
                                != NULL)
                {
                        if (PyInt_Check(temp))
                                options.maxBufferedMessages = (int) PyInt_AsLong(temp);
                        else
                        {
                                PyErr_SetString(PyExc_TypeError, "maxBufferedMessages value must be int");
                                return NULL;
                        }
                }
                rc = MQTTAsync_createWithOptions(&c, serverURI, clientId, persistence_option, NULL, &options);
        }
        else
                rc = MQTTAsync_create(&c, serverURI, clientId, persistence_option, NULL);


        return Py_BuildValue("ik", rc, c);
}


static List* callbacks = NULL;
static List* connected_callbacks = NULL;

enum msgTypes
{
        CONNECT, PUBLISH, SUBSCRIBE, SUBSCRIBE_MANY, UNSUBSCRIBE
};

typedef struct
{
        MQTTAsync c;
        PyObject *context;
        PyObject *cl, *ma, *dc;
} CallbackEntry;


typedef struct
{
        MQTTAsync c;
        PyObject *context;
        PyObject *co;
} ConnectedEntry;


int clientCompare(void* a, void* b)
{
        CallbackEntry* e = (CallbackEntry*) a;
        return e->c == (MQTTAsync) b;
}


int connectedCompare(void* a, void* b)
{
        ConnectedEntry* e = (ConnectedEntry*) a;
        return e->c == (MQTTAsync) b;
}


void connected(void* context, char* cause)
{
        /* call the right Python function, using the context */
        PyObject *arglist;
        PyObject *result;
        ConnectedEntry* e = context;
        PyGILState_STATE gstate;

        gstate = PyGILState_Ensure();
        arglist = Py_BuildValue("Os", e->context, cause);
        result = PyEval_CallObject(e->co, arglist);
        Py_DECREF(arglist);
        PyGILState_Release(gstate);
}


void connectionLost(void* context, char* cause)
{
        /* call the right Python function, using the context */
        PyObject *arglist;
        PyObject *result;
        CallbackEntry* e = context;
        PyGILState_STATE gstate;

        gstate = PyGILState_Ensure();
        arglist = Py_BuildValue("Os", e->context, cause);
        result = PyEval_CallObject(e->cl, arglist);
        Py_DECREF(arglist);
        PyGILState_Release(gstate);
}


int messageArrived(void* context, char* topicName, int topicLen,
                MQTTAsync_message* message)
{
        PyObject *result = NULL;
        CallbackEntry* e = context;
        int rc = -99;
        PyGILState_STATE gstate;

        gstate = PyGILState_Ensure();
        if (topicLen == 0)
                result = PyObject_CallFunction(e->ma, "Os{ss#sisisisi}", e->context,
                                topicName, "payload", message->payload, message->payloadlen,
                                "qos", message->qos, "retained", message->retained, "dup",
                                message->dup, "msgid", message->msgid);
        else
                result = PyObject_CallFunction(e->ma, "Os#{ss#sisisisi}", e->context,
                                topicName, topicLen, "payload", message->payload,
                                message->payloadlen, "qos", message->qos, "retained",
                                message->retained, "dup", message->dup, "msgid",
                                message->msgid);
        if (result)
        {
                if (PyInt_Check(result))
                        rc = (int) PyInt_AsLong(result);
                Py_DECREF(result);
        }
        PyGILState_Release(gstate);
        MQTTAsync_free(topicName);
        MQTTAsync_freeMessage(&message);
        return rc;
}


void deliveryComplete(void* context, MQTTAsync_token dt)
{
        PyObject *arglist;
        PyObject *result;
        CallbackEntry* e = context;
        PyGILState_STATE gstate;

        gstate = PyGILState_Ensure();
        arglist = Py_BuildValue("Oi", e->context, dt);
        result = PyEval_CallObject(e->dc, arglist);
        Py_DECREF(arglist);
        PyGILState_Release(gstate);
}


static PyObject* mqttv3_setcallbacks(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        CallbackEntry* e = NULL;
        int rc;

        e = malloc(sizeof(CallbackEntry));

        if (!PyArg_ParseTuple(args, "kOOOO", &c, (PyObject**) &e->context, &e->cl,
                        &e->ma, &e->dc))
                return NULL;
        e->c = c;

        if ((e->cl != Py_None && !PyCallable_Check(e->cl))
                        || (e->ma != Py_None && !PyCallable_Check(e->ma))
                        || (e->dc != Py_None && !PyCallable_Check(e->dc)))
        {
                PyErr_SetString(PyExc_TypeError,
                                "3rd, 4th and 5th parameters must be callable or None");
                return NULL;
        }

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_setCallbacks(c, e, connectionLost, messageArrived, deliveryComplete);
        Py_END_ALLOW_THREADS

        if (rc == MQTTASYNC_SUCCESS)
        {
                ListElement* temp = NULL;
                if ((temp = ListFindItem(callbacks, c, clientCompare)) != NULL)
                {
                        ListDetach(callbacks, temp->content);
                        free(temp->content);
                }
                ListAppend(callbacks, e, sizeof(e));
                Py_XINCREF(e->cl);
                Py_XINCREF(e->ma);
                Py_XINCREF(e->dc);
                Py_XINCREF(e->context);
        }

        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_setconnected(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        ConnectedEntry* e = NULL;
        int rc;

        e = malloc(sizeof(ConnectedEntry));

        if (!PyArg_ParseTuple(args, "kOO", &c, (PyObject**) &e->context, &e->co))
                return NULL;
        e->c = c;

        if (e->co != Py_None && !PyCallable_Check(e->co))
        {
                PyErr_SetString(PyExc_TypeError,
                                "3rd parameter must be callable or None");
                return NULL;
        }

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_setConnected(c, e, connected);
        Py_END_ALLOW_THREADS

        if (rc == MQTTASYNC_SUCCESS)
        {
                ListElement* temp = NULL;
                if ((temp = ListFindItem(connected_callbacks, c, connectedCompare)) != NULL)
                {
                        ListDetach(connected_callbacks, temp->content);
                        free(temp->content);
                }
                ListAppend(connected_callbacks, e, sizeof(e));
                Py_XINCREF(e->co);
                Py_XINCREF(e->context);
        }

        return Py_BuildValue("i", rc);
}


typedef struct
{
        MQTTAsync c;
        PyObject *context;
        PyObject *onSuccess, *onFailure;
        enum msgTypes msgType;
} ResponseEntry;


void onSuccess(void* context, MQTTAsync_successData* response)
{
        PyObject *result = NULL;
        ResponseEntry* e = context;
        PyGILState_STATE gstate;

        gstate = PyGILState_Ensure();
        switch (e->msgType)
        {
        case CONNECT:
                result = PyObject_CallFunction(e->onSuccess, "O{sisiss}", (e->context) ? e->context : Py_None,
                        "MQTTVersion", response->alt.connect.MQTTVersion,
                        "sessionPresent", response->alt.connect.sessionPresent,
                        "serverURI", response->alt.connect.serverURI);
                break;

        case PUBLISH:
                result = PyObject_CallFunction(e->onSuccess, "O{si ss s{ss# sisi}}", (e->context) ? e->context : Py_None,
                        "token", response->token,
                        "destinationName", response->alt.pub.destinationName,
                        "message",
                        "payload", response->alt.pub.message.payload,
                        response->alt.pub.message.payloadlen,
                        "qos", response->alt.pub.message.qos,
                        "retained", response->alt.pub.message.retained);
                break;

        case SUBSCRIBE:
                result = PyObject_CallFunction(e->onSuccess, "O{sisi}", (e->context) ? e->context : Py_None,
                        "token", response->token,
                        "qos", response->alt.qos);
                break;

        case SUBSCRIBE_MANY:
                result = PyObject_CallFunction(e->onSuccess, "O{sis[i]}", (e->context) ? e->context : Py_None,
                        "token", response->token,
                        "qosList", response->alt.qosList[0]);
                break;

        case UNSUBSCRIBE:
                result = PyObject_CallFunction(e->onSuccess, "O{si}", (e->context) ? e->context : Py_None,
                                        "token", response->token);
                break;
        }
        if (result)
        {
                Py_DECREF(result);
                printf("decrementing reference count for result\n");
        }
        PyGILState_Release(gstate);
        free(e);
}


void onFailure(void* context, MQTTAsync_failureData* response)
{
        PyObject *result = NULL;
        PyObject *arglist = NULL;
        ResponseEntry* e = context;
        PyGILState_STATE gstate;

        gstate = PyGILState_Ensure();

        // TODO: convert response into Python structure
        if (e->context)
                arglist = Py_BuildValue("OO", e->context, response);
        else
                arglist = Py_BuildValue("OO", Py_None, response);

        result = PyEval_CallObject(e->onFailure, arglist);
        Py_DECREF(arglist);
        PyGILState_Release(gstate);
        free(e);
}


/* return true if ok, false otherwise */
int getResponseOptions(MQTTAsync c, PyObject *pyoptions, MQTTAsync_responseOptions* responseOptions,
                enum msgTypes msgType)
{
        PyObject *temp = NULL;

        if (!pyoptions)
                return 1;

        if (!PyDict_Check(pyoptions))
        {
                PyErr_SetString(PyExc_TypeError, "Response options must be a dictionary");
                return 0;
        }

        if ((temp = PyDict_GetItemString(pyoptions, "onSuccess")) != NULL)
        {
                if (PyCallable_Check(temp)) /* temp points to Python function */
                        responseOptions->onSuccess = (MQTTAsync_onSuccess*)temp;
                else
                {
                        PyErr_SetString(PyExc_TypeError,
                                        "onSuccess value must be callable");
                        return 0;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "onFailure")) != NULL)
        {
                if (PyCallable_Check(temp))
                        responseOptions->onFailure = (MQTTAsync_onFailure*)temp;
                else
                {
                        PyErr_SetString(PyExc_TypeError,
                                        "onFailure value must be callable");
                        return 0;
                }
        }

        responseOptions->context = PyDict_GetItemString(pyoptions, "context");

        if (responseOptions->onFailure || responseOptions->onSuccess)
        {
                ResponseEntry* r = malloc(sizeof(ResponseEntry));
                r->c = c;
                r->context = responseOptions->context;
                responseOptions->context = r;
                r->onSuccess = (PyObject*)responseOptions->onSuccess;
                responseOptions->onSuccess = onSuccess;
                r->onFailure = (PyObject*)responseOptions->onFailure;
                responseOptions->onFailure = onFailure;
                r->msgType = msgType;
        }

        return 1;  /* not an error, if we get here */
}


static PyObject* mqttv3_connect(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        PyObject *pyoptions = NULL, *temp;
        MQTTAsync_connectOptions connectOptions = MQTTAsync_connectOptions_initializer;
        MQTTAsync_willOptions willOptions = MQTTAsync_willOptions_initializer;
        int rc;

        if (!PyArg_ParseTuple(args, "k|O", &c, &pyoptions))
                return NULL;

        if (!pyoptions)
                goto skip;

        if (!PyDict_Check(pyoptions))
        {
                PyErr_SetString(PyExc_TypeError, "2nd parameter must be a dictionary");
                return NULL;
        }

        if ((temp = PyDict_GetItemString(pyoptions, "onSuccess")) != NULL)
        {
                if (PyCallable_Check(temp)) /* temp points to Python function */
                        connectOptions.onSuccess = (MQTTAsync_onSuccess*)temp;
                else
                {
                        PyErr_SetString(PyExc_TypeError,
                                        "onSuccess value must be callable");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "onFailure")) != NULL)
        {
                if (PyCallable_Check(temp))
                        connectOptions.onFailure = (MQTTAsync_onFailure*)temp;
                else
                {
                        PyErr_SetString(PyExc_TypeError,
                                        "onFailure value must be callable");
                        return NULL;
                }
        }

        connectOptions.context = PyDict_GetItemString(pyoptions, "context");

        if (connectOptions.onFailure || connectOptions.onSuccess)
        {
                ResponseEntry* r = malloc(sizeof(ResponseEntry));
                r->c = c;
                r->context = connectOptions.context;
                connectOptions.context = r;
                r->onSuccess = (PyObject*)connectOptions.onSuccess;
                connectOptions.onSuccess = onSuccess;
                r->onFailure = (PyObject*)connectOptions.onFailure;
                connectOptions.onFailure = onFailure;
                r->msgType = CONNECT;
        }

        if ((temp = PyDict_GetItemString(pyoptions, "keepAliveInterval")) != NULL)
        {
                if (PyInt_Check(temp))
                        connectOptions.keepAliveInterval = (int) PyInt_AsLong(temp);
                else
                {
                        PyErr_SetString(PyExc_TypeError,
                                        "keepAliveLiveInterval value must be int");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "cleansession")) != NULL)
        {
                if (PyInt_Check(temp))
                        connectOptions.cleansession = (int) PyInt_AsLong(temp);
                else
                {
                        PyErr_SetString(PyExc_TypeError, "cleansession value must be int");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "will")) != NULL)
        {
                if (PyDict_Check(temp))
                {
                        PyObject *wtemp = NULL;
                        if ((wtemp = PyDict_GetItemString(temp, "topicName")) == NULL)
                        {
                                PyErr_SetString(PyExc_TypeError,
                                                "will topicName value must be set");
                                return NULL;
                        }
                        else
                        {
                                if (PyString_Check(wtemp))
                                        willOptions.topicName = PyString_AsString(wtemp);
                                else
                                {
                                        PyErr_SetString(PyExc_TypeError,
                                                        "will topicName value must be string");
                                        return NULL;
                                }
                        }
                        if ((wtemp = PyDict_GetItemString(temp, "message")) == NULL)
                        {
                                PyErr_SetString(PyExc_TypeError,
                                                "will message value must be set");
                                return NULL;
                        }
                        else
                        {
                                if (PyString_Check(wtemp))
                                        willOptions.message = PyString_AsString(wtemp);
                                else
                                {
                                        PyErr_SetString(PyExc_TypeError,
                                                        "will message value must be string");
                                        return NULL;
                                }
                        }
                        if ((wtemp = PyDict_GetItemString(temp, "retained")) != NULL)
                        {
                                if (PyInt_Check(wtemp))
                                        willOptions.retained = (int) PyInt_AsLong(wtemp);
                                else
                                {
                                        PyErr_SetString(PyExc_TypeError,
                                                        "will retained value must be int");
                                        return NULL;
                                }
                        }
                        if ((wtemp = PyDict_GetItemString(temp, "qos")) != NULL)
                        {
                                if (PyInt_Check(wtemp))
                                        willOptions.qos = (int) PyInt_AsLong(wtemp);
                                else
                                {
                                        PyErr_SetString(PyExc_TypeError,
                                                        "will qos value must be int");
                                        return NULL;
                                }
                        }
                        connectOptions.will = &willOptions;
                }
                else
                {
                        PyErr_SetString(PyExc_TypeError, "will value must be dictionary");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "username")) != NULL)
        {
                if (PyString_Check(temp))
                        connectOptions.username = PyString_AsString(temp);
                else
                {
                        PyErr_SetString(PyExc_TypeError, "username value must be string");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "password")) != NULL)
        {
                if (PyString_Check(temp))
                        connectOptions.username = PyString_AsString(temp);
                else
                {
                        PyErr_SetString(PyExc_TypeError, "password value must be string");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "automaticReconnect")) != NULL)
        {
                if (PyInt_Check(temp))
                        connectOptions.automaticReconnect = (int) PyInt_AsLong(temp);
                else
                {
                        PyErr_SetString(PyExc_TypeError, "automatic reconnect value must be int");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "minRetryInterval")) != NULL)
        {
                if (PyInt_Check(temp))
                        connectOptions.minRetryInterval = (int) PyInt_AsLong(temp);
                else
                {
                        PyErr_SetString(PyExc_TypeError, "minRetryInterval value must be int");
                        return NULL;
                }
        }

        if ((temp = PyDict_GetItemString(pyoptions, "maxRetryInterval")) != NULL)
        {
                if (PyInt_Check(temp))
                        connectOptions.maxRetryInterval = (int) PyInt_AsLong(temp);
                else
                {
                        PyErr_SetString(PyExc_TypeError, "maxRetryInterval value must be int");
                        return NULL;
                }
        }

skip:
        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_connect(c, &connectOptions);
        Py_END_ALLOW_THREADS
        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_disconnect(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        MQTTAsync_disconnectOptions options = MQTTAsync_disconnectOptions_initializer;
        int rc;

        if (!PyArg_ParseTuple(args, "k|i", &c, &options.timeout))
                return NULL;

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_disconnect(c, &options);
        Py_END_ALLOW_THREADS
        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_isConnected(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        int rc;

        if (!PyArg_ParseTuple(args, "k", &c))
                return NULL;
        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_isConnected(c);
        Py_END_ALLOW_THREADS
        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_subscribe(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        MQTTAsync_responseOptions response = MQTTAsync_responseOptions_initializer;
        PyObject *pyoptions = NULL;
        char* topic;
        int qos = 2;
        int rc;

        if (!PyArg_ParseTuple(args, "ks|iO", &c, &topic, &qos, &pyoptions))
                return NULL;

        if (!getResponseOptions(c, pyoptions, &response, SUBSCRIBE))
                return NULL;

        Py_BEGIN_ALLOW_THREADS;
        rc = MQTTAsync_subscribe(c, topic, qos, &response);
        Py_END_ALLOW_THREADS;
        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_subscribeMany(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        MQTTAsync_responseOptions response = MQTTAsync_responseOptions_initializer;
        PyObject* topicList;
        PyObject* qosList;
        PyObject *pyoptions = NULL;

        int count;
        char** topics;
        int* qoss;

        int i, rc = 0;

        if (!PyArg_ParseTuple(args, "kOO|O", &c, &topicList, &qosList, &pyoptions))
                return NULL;

        if (!getResponseOptions(c, pyoptions, &response, SUBSCRIBE))
                return NULL;

        if (!PySequence_Check(topicList) || !PySequence_Check(qosList))
        {
                PyErr_SetString(PyExc_TypeError,
                                "3rd and 4th parameters must be sequences");
                return NULL;
        }

        if ((count = PySequence_Length(topicList)) != PySequence_Length(qosList))
        {
                PyErr_SetString(PyExc_TypeError,
                                "3rd and 4th parameters must be sequences of the same length");
                return NULL;
        }

        topics = malloc(count * sizeof(char*));
        for (i = 0; i < count; ++i)
                topics[i] = PyString_AsString(PySequence_GetItem(topicList, i));

        qoss = malloc(count * sizeof(int));
        for (i = 0; i < count; ++i)
                qoss[i] = (int) PyInt_AsLong(PySequence_GetItem(qosList, i));

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_subscribeMany(c, count, topics,  qoss, &response);
        Py_END_ALLOW_THREADS

        for (i = 0; i < count; ++i)
                PySequence_SetItem(qosList, i, PyInt_FromLong((long) qoss[i]));

        free(topics);
        free(qoss);

        if (rc == MQTTASYNC_SUCCESS)
                return Py_BuildValue("iO", rc, qosList);
        else
                return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_unsubscribe(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        MQTTAsync_responseOptions response = MQTTAsync_responseOptions_initializer;
        PyObject *pyoptions = NULL;
        char* topic;
        int rc;

        if (!PyArg_ParseTuple(args, "ks|O", &c, &topic, &pyoptions))
                return NULL;

        if (!getResponseOptions(c, pyoptions, &response, UNSUBSCRIBE))
                return NULL;

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_unsubscribe(c, topic, &response);
        Py_END_ALLOW_THREADS
        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_unsubscribeMany(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        MQTTAsync_responseOptions response = MQTTAsync_responseOptions_initializer;
        PyObject* topicList;
        PyObject *pyoptions = NULL;

        int count;
        char** topics;

        int i, rc = 0;

        if (!PyArg_ParseTuple(args, "kO|O", &c, &topicList, &pyoptions))
                return NULL;

        if (!getResponseOptions(c, pyoptions, &response, UNSUBSCRIBE))
                return NULL;

        if (!PySequence_Check(topicList))
        {
                PyErr_SetString(PyExc_TypeError, "3rd parameter must be sequences");
                return NULL;
        }

        count = PySequence_Length(topicList);
        topics = malloc(count * sizeof(char*));
        for (i = 0; i < count; ++i)
                topics[i] = PyString_AsString(PySequence_GetItem(topicList, i));

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_unsubscribeMany(c, count, topics, &response);
        Py_END_ALLOW_THREADS

        free(topics);

        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_send(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        char* destinationName;
        int payloadlen;
        void* payload;
        int qos = 0;
        int retained = 0;
        MQTTAsync_responseOptions response = MQTTAsync_responseOptions_initializer;
        PyObject *pyoptions = NULL;
        int rc;

        if (!PyArg_ParseTuple(args, "kss#|iiO", &c, &destinationName, &payload,
                        &payloadlen, &qos, &retained, &pyoptions))
                return NULL;

        if (!getResponseOptions(c, pyoptions, &response, PUBLISH))
                return NULL;

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_send(c, destinationName, payloadlen, payload, qos, retained, &response);
        Py_END_ALLOW_THREADS

        if (rc == MQTTASYNC_SUCCESS && qos > 0)
                return Py_BuildValue("ii", rc, response);
        else
                return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_sendMessage(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        char* destinationName;
        PyObject *message, *temp;
        MQTTAsync_message msg = MQTTAsync_message_initializer;
        MQTTAsync_responseOptions response = MQTTAsync_responseOptions_initializer;
        PyObject *pyoptions = NULL;
        int rc;

        if (!PyArg_ParseTuple(args, "ksO|O", &c, &destinationName, &message, &pyoptions))
                return NULL;

        if (!getResponseOptions(c, pyoptions, &response, PUBLISH))
                return NULL;

        if (!PyDict_Check(message))
        {
                PyErr_SetString(PyExc_TypeError, "3rd parameter must be a dictionary");
                return NULL;
        }

        if ((temp = PyDict_GetItemString(message, "payload")) == NULL)
        {
                PyErr_SetString(PyExc_TypeError, "dictionary must have payload key");
                return NULL;
        }

        if (PyString_Check(temp))
                PyString_AsStringAndSize(temp, (char**) &msg.payload,
                                (Py_ssize_t*) &msg.payloadlen);
        else
        {
                PyErr_SetString(PyExc_TypeError, "payload value must be string");
                return NULL;
        }

        if ((temp = PyDict_GetItemString(message, "qos")) == NULL)
                msg.qos = (int) PyInt_AsLong(temp);

        if ((temp = PyDict_GetItemString(message, "retained")) == NULL)
                msg.retained = (int) PyInt_AsLong(temp);

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_sendMessage(c, destinationName, &msg, &response);
        Py_END_ALLOW_THREADS

        if (rc == MQTTASYNC_SUCCESS && msg.qos > 0)
                return Py_BuildValue("ii", rc, response);
        else
                return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_waitForCompletion(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        unsigned long timeout = 1000L;
        MQTTAsync_token dt;
        int rc;

        if (!PyArg_ParseTuple(args, "ki|i", &c, &dt, &timeout))
                return NULL;

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_waitForCompletion(c, dt, timeout);
        Py_END_ALLOW_THREADS

        return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_getPendingTokens(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        MQTTAsync_token* tokens;
        int rc;

        if (!PyArg_ParseTuple(args, "k", &c))
                return NULL;

        Py_BEGIN_ALLOW_THREADS
        rc = MQTTAsync_getPendingTokens(c, &tokens);
        Py_END_ALLOW_THREADS

        if (rc == MQTTASYNC_SUCCESS)
        {
                int i = 0;
                PyObject* dts = PyList_New(0);

                while (tokens[i] != -1)
                        PyList_Append(dts, PyInt_FromLong((long) tokens[i]));

                return Py_BuildValue("iO", rc, dts);
        }
        else
                return Py_BuildValue("i", rc);
}


static PyObject* mqttv3_destroy(PyObject* self, PyObject *args)
{
        MQTTAsync c;
        ListElement* temp = NULL;

        if (!PyArg_ParseTuple(args, "k", &c))
                return NULL;

        if ((temp = ListFindItem(callbacks, c, clientCompare)) != NULL)
        {
                ListDetach(callbacks, temp->content);
                free(temp->content);
        }

        if ((temp = ListFindItem(connected_callbacks, c, connectedCompare)) != NULL)
        {
                ListDetach(connected_callbacks, temp->content);
                free(temp->content);
        }

        MQTTAsync_destroy(&c);

        Py_INCREF(Py_None);
        return Py_None;
}


static PyMethodDef MqttV3Methods[] =
        {
                { "create", mqttv3_create, METH_VARARGS, "Create an MQTTv3 client." },
                { "setcallbacks", mqttv3_setcallbacks, METH_VARARGS,
                                "Sets the callback functions for a particular client." },
                { "setconnected", mqttv3_setconnected, METH_VARARGS,
                                "Sets the connected callback function for a particular client." },
                { "connect", mqttv3_connect, METH_VARARGS,
                                "Connects to a server using the specified options." },
                { "disconnect", mqttv3_disconnect, METH_VARARGS,
                                "Disconnects from a server." },
                { "isConnected", mqttv3_isConnected, METH_VARARGS,
                                "Determines if this client is currently connected to the server." },
                { "subscribe", mqttv3_subscribe, METH_VARARGS,
                                "Subscribe to the given topic." },
                { "subscribeMany", mqttv3_subscribeMany, METH_VARARGS,
                                "Subscribe to the given topics." },
                { "unsubscribe", mqttv3_unsubscribe, METH_VARARGS,
                                "Unsubscribe from the given topic." },
                { "unsubscribeMany", mqttv3_unsubscribeMany, METH_VARARGS,
                                "Unsubscribe from the given topics." },
                { "send", mqttv3_send, METH_VARARGS,
                                "Publish a message to the given topic." },
                { "sendMessage", mqttv3_sendMessage, METH_VARARGS,
                                "Publish a message to the given topic." },
                { "waitForCompletion", mqttv3_waitForCompletion, METH_VARARGS,
                                "Waits for the completion of the delivery of the message represented by a delivery token." },
                { "getPendingTokens", mqttv3_getPendingTokens, METH_VARARGS,
                                "Returns the tokens pending of completion." },
                { "destroy", mqttv3_destroy, METH_VARARGS,
                                "Free memory allocated to a MQTT client. It is the opposite to create." },
                { NULL, NULL, 0, NULL } /* Sentinel */
        };


PyMODINIT_FUNC initpaho_mqtt3a(void)
{
        PyObject *m;

        PyEval_InitThreads();

        callbacks = ListInitialize();
        connected_callbacks = ListInitialize();

        m = Py_InitModule("paho_mqtt3a", MqttV3Methods);
        if (m == NULL)
                return;

        MqttV3Error = PyErr_NewException("paho_mqtt3a.error", NULL, NULL);
        Py_INCREF(MqttV3Error);
        PyModule_AddObject(m, "error", MqttV3Error);

        PyModule_AddIntConstant(m, "SUCCESS", MQTTASYNC_SUCCESS);
        PyModule_AddIntConstant(m, "FAILURE", MQTTASYNC_FAILURE);
        PyModule_AddIntConstant(m, "DISCONNECTED", MQTTASYNC_DISCONNECTED);
        PyModule_AddIntConstant(m, "MAX_MESSAGES_INFLIGHT",     MQTTASYNC_MAX_MESSAGES_INFLIGHT);
        PyModule_AddIntConstant(m, "BAD_UTF8_STRING", MQTTASYNC_BAD_UTF8_STRING);
        PyModule_AddIntConstant(m, "BAD_NULL_PARAMETER", MQTTASYNC_NULL_PARAMETER);
        PyModule_AddIntConstant(m, "BAD_TOPICNAME_TRUNCATED", MQTTASYNC_TOPICNAME_TRUNCATED);
        PyModule_AddIntConstant(m, "PERSISTENCE_DEFAULT", MQTTCLIENT_PERSISTENCE_DEFAULT);
        PyModule_AddIntConstant(m, "PERSISTENCE_NONE", MQTTCLIENT_PERSISTENCE_NONE);
        PyModule_AddIntConstant(m, "PERSISTENCE_USER", MQTTCLIENT_PERSISTENCE_USER);
        PyModule_AddIntConstant(m, "PERSISTENCE_ERROR",
        MQTTCLIENT_PERSISTENCE_ERROR);
}