
/*---------------------------------------------------------------------------*/
/* apiwebsocket_client.cpp                                                   */
/* copyright (c) innovaphone 2017                                            */
/*                                                                           */
/*---------------------------------------------------------------------------*/

#include <platform.h>
#include "sdk/common/os/iomux.h"
#include "common/ilib/json.h"
#include "common/ilib/btree.h"
#include "common/ilib/hash.h"
#include "sdk/common/interface/socket.h"
#include "sdk/common/interface/webserver_plugin.h"
#include "sdk/common/interface/websocket_client.h"
#include "../lib/apiwebsocket_client.h"

#define TIMEOUT_RECONNECT 1000
#define TIMEOUT_RECONNECT_MAX 32000
#define BACKLOG_TIME_MAX  32000
#define TIMEOUT_KEEPALIVE 60

#if defined(DBG_FLAG_ALL) || defined(DBG_FLAG_API_WEBSOCKET)
#define DBG(x) debug->printf x
#else
#define DBG(x)
#endif

/*---------------------------------------------------------------------------*/
/* ApiWebsocketClient                                                        */
/*---------------------------------------------------------------------------*/

ApiWebsocketClient::ApiWebsocketClient(class IIoMux * const iomux, class ISocketProvider * localSocketProvider, class ISocketProvider * const tcp, class ISocketProvider * const tls, class IDns * const dns, class IInstanceLog * const log, const char * uri, const char * pwd, const char * dn, const char * localSocketPath) :
    reconnectTimer(iomux, this),
    backlogTimer(iomux, this)
{
    this->iomux = iomux;
    this->localSocketProvider = localSocketProvider;
    this->tcp = tcp;
    this->tls = tls;
    this->dns = dns;
    this->log = log;
    this->name = 0;
    this->dn = _strdup(dn);
    this->pwd = _strdup(pwd);
    this->uri = _strdup(uri);
    this->localSocketPath = _strdup(localSocketPath);
    up = false;
    closing = false;
    consumers = 0;
    providers = 0;
    providerRefs = 0;
    sendCount = 0;
    backlogSend = 0;
    timeoutTime = TIMEOUT_RECONNECT;
    websocket = NULL;
    localSocket = NULL;
    recvLen = 0;
    jsonLevel = 0;
    jsonQuotes = false;
    jsonEsc = false;
    DBG(("Websocket: Connect to %s", localSocketPath && localSocketPath[0] ? localSocketPath : uri && uri[0] ? uri : "no path provided"));
    if (localSocketPath && localSocketPath[0]) {
        localSocket = localSocketProvider->CreateSocket(iomux, this, log, false);
        localSocket->Connect(localSocketPath);
    }
    else if (uri && uri[0]) {
        websocket = IWebsocketClient::Create(iomux, this, tcp, tls, dns, log, TIMEOUT_KEEPALIVE);
        websocket->Connect(uri);
    }
}

ApiWebsocketClient::~ApiWebsocketClient()
{
    ASSERT(!websocket && !localSocket && !consumers && !providers, "~ApiWebsocketClient: Not all dependencies closed yet");
    while (providerRefs) {
        ApiWebsocketProviderRef * r = (ApiWebsocketProviderRef *)providerRefs;
        providerRefs = providerRefs->btree_get(r);
        delete r;
    }
    if (uri) {
        free(uri);
        uri = 0;
    }
    free(name);
    name = 0;
    free(dn);
    dn = 0;
    free(pwd);
    pwd = 0;
    if (localSocketPath) {
        free(localSocketPath);
        localSocketPath = 0;
    }
}

void ApiWebsocketClient::ApiWebsocketClientClose()
{
    TryClose();
}

void ApiWebsocketClient::TryClose()
{
    if (!closing) {
        closing = true;
        up = false;
        reconnectTimer.Cancel();
        if (websocket) websocket->Close();
        if (localSocket) localSocket->Shutdown();
        class ApiConsumer * c = 0;
        while ((c = (class ApiConsumer *)consumers->btree_find_next_right(c ? c->api : 0))) c->ApiConsumerClose();
        class ApiProvider * p = 0;
        while ((p = (class ApiProvider *)providers->btree_find_next_right(p ? p->api : 0))) p->ApiProviderClose();
    }
    if (!websocket && !localSocket && !consumers && !providers) {
        ApiWebsocketClientCloseComplete();
    }
}

void ApiWebsocketClient::ApiProviderClosed(class ApiProvider * provider)
{
    class ApiWebsocketBacklogBuffer * b, * bNext;

    providers = providers->btree_get(provider);
    if (!backlogQueue.empty()) {
        b = backlogQueue.front();
        bNext = b->goNext();
        if (b->provider == provider) {
            delete b;
            sendCount += backlogSend;
            backlogSend = 0;
            b = bNext;
        }
        while (b) {
            bNext = b->goNext();
            if (b->provider == provider) {
                delete b;
            }
            b = bNext;
        }
        if (backlogQueue.empty()) {
            backlogTimer.Cancel();
        }
    }
    delete provider;
    if (closing) TryClose();
}

void ApiWebsocketClient::ApiConsumerClosed(class ApiConsumer * consumer)
{
    class ApiWebsocketBacklogBuffer * b, * bNext;

    consumers = consumers->btree_get(consumer);
    if (!backlogQueue.empty()) {
        b = backlogQueue.front();
        bNext = b->goNext();
        if (b->consumer == consumer) {
            delete b;
            sendCount += backlogSend;
            backlogSend = 0;
            b = bNext;
        }
        while (b) {
            bNext = b->goNext();
            if (b->consumer == consumer) {
                delete b;
            }
            b = bNext;
        }
        if (backlogQueue.empty()) {
            backlogTimer.Cancel();
        }
    }
    delete consumer;
    if (closing) TryClose();
}

void ApiWebsocketClient::WebsocketConnectComplete(IWebsocketClient * const websocket)
{
    DBG(("Websocket: connected %s", uri));
    reconnectTimer.Cancel();
    timeoutTime = TIMEOUT_RECONNECT;
    char buffer[256];
    json_io msg(0);
    word base = msg.add_object(JSON_ID_ROOT, 0);
    msg.add_string(base, "mt", "AppChallenge");
    size_t len = msg.encode(base, buffer);
    websocket->Send(buffer, len);
    sendCount++;
    websocket->Recv();
}

void ApiWebsocketClient::WebsocketCloseComplete(IWebsocketClient * const websocket, closereason_t reason)
{
    DBG(("Websocket: disconnected %s", localSocketPath ? localSocketPath : uri));
    up = false;
    while (providerRefs) {
        ApiWebsocketProviderRef * r = (ApiWebsocketProviderRef *)providerRefs;
        providerRefs = providerRefs->btree_get(r);
        delete r;
    }
    sendCount = 0;
    backlogSend = 0;
    if (this->websocket) {
        delete this->websocket;
        this->websocket = 0;
    }
    ApiWebsocketClientDisconnected();
    class ApiProvider * p = 0;
    while ((p = (class ApiProvider *)providers->btree_find_next_right(p ? p->api : 0))) {
        p->ApiProviderDisconnected();
    }
    class ApiConsumer * c = 0;
    while ((c = (class ApiConsumer *)consumers->btree_find_next_right(c ? c->api : 0))) {
        c->ApiConsumerDisconnected();
    }
    if (closing) {
        TryClose();
    }
    else {
        DBG(("Websocket: reconnect after %u ms %s", timeoutTime, localSocketPath ? localSocketPath : uri));
        reconnectTimer.Start(timeoutTime);
    }
}

void ApiWebsocketClient::SendBacklog()
{
    class ApiWebsocketBacklogBuffer * b;

    if (!websocket && !localSocket) {
        if (localSocketPath) {
            localSocket = localSocketProvider->CreateSocket(iomux, this, log, false);
            localSocket->Connect(localSocketPath);
        }
        else {
            websocket = IWebsocketClient::Create(iomux, this, tcp, tls, dns, log, TIMEOUT_KEEPALIVE);
            websocket->Connect(uri);
        }
    }
    if (up && (sendCount == 0) && (backlogSend == 0) && !backlogQueue.empty()) {
        b = backlogQueue.front();
        while (b && b->consumer && !providerRefs->btree_find(b->consumer->api)) {
            b = b->goNext();
        }
        if (b) {
            b->remove();
            backlogQueue.push_front(b);
            if (localSocket) localSocket->Send(b->buffer, b->length);
            else websocket->Send(b->buffer, b->length);
            backlogSend++;
        }
    }
}

void ApiWebsocketClient::WebsocketSendResult(IWebsocketClient * const websocket)
{
    if (backlogSend != 0) {
        backlogSend--;
        delete backlogQueue.pop_front();
        if (backlogQueue.empty()) {
            backlogTimer.Cancel();
        }
        else {
            backlogTimer.Start(BACKLOG_TIME_MAX);
        }
    }
    else if (sendCount != 0) {
        sendCount--;
    }
    SendBacklog();
}

void ApiWebsocketClient::WebsocketRecvResult(IWebsocketClient * const websocket, void * buf, size_t len, bool text, bool isFragment)
{
    ASSERT(buf, "ApiWebsocketClient::WebsocketRecvResult");
    if (text && !isFragment) {
        //DBG(("Websocket: recv: %s", buf));
        json_io recv((char*)buf);
        if (recv.decode()) {
            word base = recv.get_object(JSON_ID_ROOT, 0);
            const char * mt = recv.get_string(base, "mt");
            if (mt) {
                if (!strcmp(mt, "AppChallengeResult")) {
                    const char * challenge = recv.get_string(base, "challenge");
                    if (!challenge) challenge = "";
                    const char * app = "";
                    const char * domain = "";
                    const char * sip = "";
                    const char * guid = "";
                    class hash hash;
                    byte digest_bin[HASH_SIZE_SHA256];
                    hash.init(HASH_SHA256);
                    hash.update(app, strlen(app));
                    hash.update(":", 1);
                    hash.update(domain, strlen(domain));
                    hash.update(":", 1);
                    hash.update(sip, strlen(sip));
                    hash.update(":", 1);
                    hash.update(guid, strlen(guid));
                    hash.update(":", 1);
                    hash.update(dn, strlen(dn));
                    hash.update(":", 1);
                    hash.update(challenge, strlen(challenge));
                    hash.update(":", 1);
                    hash.update(pwd, strlen(pwd));
                    hash.final(digest_bin);
                    char tmp[1024];
                    char * t = tmp;
                    char send_buffer[2048];
                    json_io send(send_buffer);
                    word sb = send.add_object(JSON_ID_ROOT, 0);
                    send.add_string(sb, "mt", "AppLogin");
                    send.add_string(sb, "app", app);
                    send.add_string(sb, "domain", domain);
                    send.add_string(sb, "sip", sip);
                    send.add_string(sb, "guid", guid);
                    send.add_string(sb, "dn", dn);
                    send.add_string(sb, "challenge", challenge);
                    send.add_hexstring(sb, "digest", digest_bin, HASH_SIZE_SHA256, t);
                    int len = send.encode();
                    websocket->Send(send_buffer, len);
                    sendCount++;
                }
                else if (!strcmp(mt, "LoginResult")) {
                    bool ok = recv.get_bool(base, "ok");
                    if (ok) {
                        DBG(("Websocket: Login OK"));
                    }
                    else {
                        DBG(("Websocket: login failed"));
                        DBG(("Websocket: closing"));
                        websocket->Close();
                    }
                }
                else if (!strcmp(mt, "Info")) {
                    const char * name = recv.get_string(base, "name");
                    if (this->name) free(this->name);
                    this->name = _strdup(name);
                    up = true;
                    ApiWebsocketClientConnected();
                    class ApiProvider * p = 0;
                    while ((p = (class ApiProvider *)providers->btree_find_next_right(p ? p->api : 0))) {
                        p->SendRegisterApi();
                        p->SendApiModel();
                        p->ApiProviderConnected();
                    }
                    class ApiConsumer * c = 0;
                    while ((c = (class ApiConsumer *)consumers->btree_find_next_right(c ? c->api : 0))) {
                        c->ApiConsumerConnected();
                    }
                    SendBacklog();
                }
                else if (!strcmp(mt, "ApiUpdate")) {
                    class ApiConsumer * c = 0;
                    word apis = recv.get_object(base, "apis");
                    if (apis != JSON_ID_NONE) {
                        while ((c = (class ApiConsumer *)consumers->btree_find_next_right(c ? c->api : 0))) {
                            word api = recv.get_object(apis, c->api);
                            if (api != JSON_ID_NONE) c->ApiConsumerUpdate(recv, api);
                        }
                        byte type, flags;
                        const char *name, *info;
                        word api;
                        api = 0;
                        while ((api = recv.get_next(apis, api, type, flags, name, info)) != JSON_ID_NONE) {
                            ApiWebsocketProviderRef * r = (ApiWebsocketProviderRef *)(providerRefs->btree_find(name));
                            if (!r) {
                                r = new ApiWebsocketProviderRef(name);
                                providerRefs = providerRefs->btree_put(r);
                            }
                            (r->refCount)++;
                        }
                        SendBacklog();
                    }
                    word defaultApiProviders = recv.get_object(base, "defaultApiProviders");
                    if (defaultApiProviders != JSON_ID_NONE) {
                        c = 0;
                        while ((c = (class ApiConsumer *)consumers->btree_find_next_right(c ? c->api : 0))) {
                            const char * apiProvider = recv.get_string(defaultApiProviders, c->api);
                            c->ApiConsumerDefaultProvider(apiProvider);
                        }
                    }
                }
                else if (!strcmp(mt, "ApiObsolete")) {
                    class ApiConsumer * c = 0;
                    word apis = recv.get_object(base, "apis");
                    if (apis != JSON_ID_NONE) {
                        while ((c = (class ApiConsumer *)consumers->btree_find_next_right(c ? c->api : 0))) {
                            word api = recv.get_object(apis, c->api);
                            if (api != JSON_ID_NONE) c->ApiConsumerObsolete(recv, api);
                        }
                        byte type, flags;
                        const char *name, *info;
                        word api;
                        api = 0;
                        while ((api = recv.get_next(apis, api, type, flags, name, info)) != JSON_ID_NONE) {
                            ApiWebsocketProviderRef * r = (ApiWebsocketProviderRef *)(providerRefs->btree_find(name));
                            if (r) {
                                (r->refCount)--;
                                if (r->refCount == 0) {
                                    providerRefs = providerRefs->btree_get(r);
                                    delete r;
                                }
                            }
                        }
                    }
                }
                else if (!strcmp(mt, "ApiRequest")) {
                    const char * api = recv.get_string(base, "apiId");
                    if (api) {
                        class ApiProvider * p = (class ApiProvider *)providers->btree_find(api);
                        if (p) {
                            const char * client = recv.get_string(base, "client");
                            const char * consumer = recv.get_string(base, "consumer");
                            const char * src = recv.get_string(base, "src");
                            word msg = recv.get_object(base, "msg");
                            p->ApiProviderRecv(client, consumer, src, recv, msg);
                        }
                    }
                }
                else if (!strcmp(mt, "ApiResult")) {
                    const char * api = recv.get_string(base, "apiId");
                    if (api) {
                        class ApiConsumer * c = (class ApiConsumer *)consumers->btree_find(api);
                        if (c) {
                            const char * provider = recv.get_string(base, "provider");
                            const char * src = recv.get_string(base, "src");
                            word msg = recv.get_object(base, "msg");
                            c->ApiConsumerRecv(provider, src, recv, msg);
                        }
                    }
                }
                else if (!strcmp(mt, "ApiConsumerClosed")) {
                    const char * api = recv.get_string(base, "apiId");
                    if (api) {
                        class ApiProvider * p = (class ApiProvider *)providers->btree_find(api);
                        if (p) {
                            const char * client = recv.get_string(base, "client");
                            const char * consumer = recv.get_string(base, "consumer");
                            p->ApiProviderConsumerClosed(client, consumer);
                        }
                    }
                    else {
                        class ApiProvider * p = 0;
                        while ((p = (class ApiProvider *)providers->btree_find_next_right(p ? p->api : 0))) {
                            const char * client = recv.get_string(base, "client");
                            const char * consumer = recv.get_string(base, "consumer");
                            p->ApiProviderConsumerClosed(client, consumer);
                        }
                    }
                }
            }
        }
    }
    if (websocket) {
        websocket->Recv();
    }
}

void * ApiWebsocketClient::WebsocketRecvBuffer(size_t len)
{
    recvBuf[len] = 0;
    return recvBuf;
}

void ApiWebsocketClient::WebsocketRecvCanceled(IWebsocketClient * const websocket, void * buf)
{
}

void ApiWebsocketClient::TimerOnTimeout(ITimer * timer)
{
    if (!closing) {
        if (timer == &reconnectTimer) {
            ASSERT(!websocket, "TimerOnTimeout");
            DBG(("Websocket: reconnect to %s", localSocketPath ? localSocketPath : uri));
            timeoutTime = MIN(timeoutTime * 2, TIMEOUT_RECONNECT_MAX);
            
            if (localSocketPath) {
                localSocket = localSocketProvider->CreateSocket(iomux, this, log, false);
                localSocket->Connect(localSocketPath);
            }
            else {
                websocket = IWebsocketClient::Create(iomux, this, tcp, tls, dns, log, TIMEOUT_KEEPALIVE);
                websocket->Connect(uri);
            }
        }
        if (timer == &backlogTimer) {
            DBG(("Websocket: backlog timeout %s", localSocketPath ? localSocketPath : uri));
            while (!backlogQueue.empty()) {
                sendCount += backlogSend;
                backlogSend = 0;
                delete backlogQueue.pop_front();
            }
        }
    }
}

void ApiWebsocketClient::SocketConnectComplete(ISocket * const socket)
{
    DBG(("Websocket: connected %s", localSocketPath));
    socket->Recv(&recvBuf[recvLen], sizeof(recvBuf) - recvLen - 1, true);
}


void ApiWebsocketClient::SocketSendResult(ISocket * const socket)
{
    WebsocketSendResult(websocket);
}

void ApiWebsocketClient::SocketRecvResult(ISocket * const socket, void * buf, size_t len)
{
    unsigned jsonStart;
    unsigned jsonPos;

    jsonStart = 0;
    jsonPos = recvLen;
    recvLen += (unsigned int)len;
    while (jsonPos < recvLen) {
        while ((jsonPos < recvLen) && ((recvBuf[jsonPos] != '}') || (jsonLevel > 1) || jsonQuotes)) {
            if (jsonQuotes) {
                if (jsonEsc) {
                    jsonEsc = false;
                }
                else if (recvBuf[jsonPos] == '\\') {
                    jsonEsc = true;
                }
                else if (recvBuf[jsonPos] == '"') {
                    jsonQuotes = false;
                }
            }
            else if (recvBuf[jsonPos] == '"') {
                jsonQuotes = true;
            }
            else if (recvBuf[jsonPos] == '{') {
                jsonLevel++;
            }
            else if (recvBuf[jsonPos] == '}') {
                jsonLevel--;
            }
            jsonPos++;
        }
        if ((jsonPos < recvLen) && (recvBuf[jsonPos] == '}') && (jsonLevel == 1) && !jsonQuotes) {
            jsonPos++;
            recvBuf[jsonPos] = '\0';

            WebsocketRecvResult(websocket, &recvBuf[jsonStart], jsonPos - jsonStart, true, false);

            recvBuf[jsonPos] = '{';
            jsonLevel = 0;
            jsonQuotes = false;
            jsonEsc = false;
            jsonStart = jsonPos;
        }
    }
    if (jsonStart != 0) {
        memmove(recvBuf, &recvBuf[jsonStart], recvLen - jsonStart);
        recvLen -= jsonStart;
    }
    socket->Recv(&recvBuf[recvLen], sizeof(recvBuf) - recvLen - 1, true);
}

void ApiWebsocketClient::SocketShutdown(ISocket * const socket, shutdownreason_t reason)
{
    delete socket;
    localSocket = NULL;
    WebsocketCloseComplete(websocket, WSCR_NORMAL_CLOSE);
}

bool ApiWebsocketClient::ApiWebsocketClientIsConnected()
{
    return up;
}

void ApiWebsocketClient::ApiWebsocketClientSend(const void * buf, size_t len)
{
    if (up) {
        if (localSocket) localSocket->Send(buf, len);
        else websocket->Send(buf, len);
        sendCount++;
    }
}

void ApiWebsocketClient::ApiWebsocketClientSend(const void * buf, size_t len, class ApiProvider * provider)
{
    if (backlogQueue.empty()) {
        backlogTimer.Start(BACKLOG_TIME_MAX);
    }
    backlogQueue.push_back(new ApiWebsocketBacklogBuffer(buf, len, provider, 0));
    SendBacklog();
}

void ApiWebsocketClient::ApiWebsocketClientSend(const void * buf, size_t len, class ApiConsumer * consumer)
{
    if (backlogQueue.empty()) {
        backlogTimer.Start(BACKLOG_TIME_MAX);
    }
    backlogQueue.push_back(new ApiWebsocketBacklogBuffer(buf, len, 0, consumer));
    SendBacklog();
}

/*---------------------------------------------------------------------------*/
/* ApiProvider                                                               */
/*---------------------------------------------------------------------------*/

ApiProvider::ApiProvider(class ApiWebsocketClient * const client, const char * api, const char * dn)
{
    DBG(("Provider(%s): created", api));
    this->client = client;
    this->api = _strdup(api);
    this->dn = _strdup(dn);
    this->model = 0;
    client->providers = client->providers->btree_put(this);
    SendRegisterApi();
    SendApiModel();
}

ApiProvider::~ApiProvider()
{
    DBG(("Provider(%s): deleted", api));
    free(api);
    api = 0;
    free(dn);
    dn = 0;
    free(model);
    model = 0;
    client = 0;
}

void ApiProvider::ApiProviderClosed()
{
    client->iomux->SetExec(this, 0);
}

void ApiProvider::IoExec(void * execContext)
{
    client->ApiProviderClosed(this);
}

void ApiProvider::ApiProviderSendJson(const char * client, const char * consumer, const char * src, const char * buffer)
{
    if (this->client) {
        json_io json(0);
        word sb = json.add_object(JSON_ID_ROOT, 0);
        json.add_string(sb, "mt", "ApiResult");
        json.add_string(sb, "src", src);
        json.add_string(sb, "apiId", api);
        json.add_string(sb, "provider", this->client->name);
        json.add_string(sb, "consumer", consumer);
        json.add_string(sb, "client", client);
        json.add_json(sb, "msg", buffer);
        char send_buffer[WEBSOCKET_MAX_PAYLOAD_SIZE];
        size_t len = json.encode(sb, send_buffer);
        DBG(("Provider(%s): send: %s", api, send_buffer));
        this->client->ApiWebsocketClientSend(send_buffer, len, this);
    }
    else {
        DBG(("Provider(%s): discard message", api));
    }
}

void ApiProvider::ApiProviderSend(const char * client, const char * consumer, const char * src, class json_io & msg, word base, char * buffer)
{
    if (this->client) {
        msg.encode(base, buffer);
        ApiProviderSendJson(client, consumer, src, buffer);
    }
    else {
        DBG(("Provider(%s): discard message", api));
    }
}

void ApiProvider::ApiProviderUpdate(class json_io & json, word base, char * buffer)
{
    json.encode(base, buffer);
    if (model) free(model);
    model = _strdup((char *)buffer);
    SendApiModel();
}

void ApiProvider::SendRegisterApi()
{
    if (client) {
        json_io json(0);
        word base = json.add_object(JSON_ID_ROOT, 0);
        json.add_string(base, "mt", "RegisterApi");
        json.add_string(base, "apiId", api);
        json.add_string(base, "provider", client->name);
        json.add_string(base, "title", dn ? dn : client->dn);
        char msg[2048];
        size_t len = json.encode(base, msg);
        DBG(("Provider(%s): send: %s", api, msg));
        client->ApiWebsocketClientSend(msg, len);
    }
}

void ApiProvider::SendApiModel()
{
    if (client) {
        json_io json(0);
        word base = json.add_object(JSON_ID_ROOT, 0);
        json.add_string(base, "mt", "ApiModel");
        json.add_string(base, "apiId", api);
        json.add_string(base, "provider", client->name);
        if (model) json.add_json(base, "model", model);
        else json.add_object(base, "model");
        char msg[2048];
        size_t len = json.encode(base, msg);
        DBG(("Provider(%s): send: %s", api, msg));
        client->ApiWebsocketClientSend(msg, len);
    }
}

int ApiProvider::btree_compare(void * key)
{
    return strcmp(api, (char *)key);
}

int ApiProvider::btree_compare(class btree * b)
{
    return strcmp(api, ((class ApiProvider *)b)->api);
}

/*---------------------------------------------------------------------------*/
/* ApiConsumer                                                               */
/*---------------------------------------------------------------------------*/

ApiConsumer::ApiConsumer(class ApiWebsocketClient * const client, const char * api)
{
    DBG(("Consumer(%s): created", api));
    this->client = client;
    this->api = _strdup(api);
    client->consumers = client->consumers->btree_put(this);
}

ApiConsumer::~ApiConsumer()
{
    DBG(("Consumer(%s): deleted", api));
    free(api);
    api = 0;
    client = 0;
}

void ApiConsumer::ApiConsumerClosed()
{
    client->iomux->SetExec(this, 0);
}

void ApiConsumer::IoExec(void * execContext)
{
    client->ApiConsumerClosed(this);
}

void ApiConsumer::ApiConsumerSendJson(const char * provider, const char * src, const char * buffer)
{
    if (client) {
        json_io json(0);
        word sb = json.add_object(JSON_ID_ROOT, 0);
        json.add_string(sb, "mt", "ApiRequest");
        json.add_string(sb, "src", src);
        json.add_string(sb, "apiId", api);
        json.add_string(sb, "provider", provider);
        json.add_string(sb, "consumer", client->name);
        json.add_json(sb, "msg", buffer);
        char send_buffer[WEBSOCKET_MAX_PAYLOAD_SIZE];
        size_t len = json.encode(sb, send_buffer);
        DBG(("Consumer(%s): send: %s", api, send_buffer));
        client->ApiWebsocketClientSend(send_buffer, len, this);
    }
    else {
        DBG(("Consumer(%s): discard message", api));
    }
}

void ApiConsumer::ApiConsumerSend(const char * provider, const char * src, class json_io & msg, word base, char * buffer)
{
    if (client) {
        msg.encode(base, buffer);
        ApiConsumerSendJson(provider, src, buffer);
    }
    else {
        DBG(("Consumer(%s): discard message", api));
    }
}

int ApiConsumer::btree_compare(void * key)
{
    return strcmp(api, (char *)key);
}

int ApiConsumer::btree_compare(class btree * b)
{
    return strcmp(api, ((class ApiConsumer *)b)->api);
}
