#include <assert.h>

#include "AudioVideoPlaybackController.h"
#include "WebSocketMediaStreamProducerImpl.h"
#include "MediaClientImpl.h"
#include "Base64.hh"

#include <stdlib.h>
#include <stdio.h>          /* Basic I/O routines          */
#include <strings.h>

namespace videonext { namespace media {

struct lws_static_init
{
    lws_static_init()
    {
        lws_set_log_level(LLL_INFO|LLL_NOTICE|LLL_DEBUG|LLL_WARN|LLL_ERR, WebSocketMediaStreamProducerImpl::lws_log_callback);
    }
};

static lws_static_init lsi;


WebSocketMediaStreamProducerImpl::WebSocketMediaStreamProducerImpl(MediaStreamHandler *mediaStreamHandler, const std::string &url,
                                                                   unsigned cacheSize, unsigned streamId,
                                                                   unsigned bufferLen, float speed)
    : RTSPMediaStreamProducerImpl(mediaStreamHandler, url, cacheSize, streamId, false, bufferLen, speed)
    , lwsServiceTask_(0), wsiContext_(0), jumpToTimestamp_(0), fd_(0), speed_(speed)
    , hevc_vconf_size_(0), hevc_vconf_(NULL), cloud_(false), audioMuted_(false)
{
    pts_.tv_sec = pts_.tv_usec = 0;

    fprintf(stderr, "WebSocketMediaStreamProducerImpl::WebSocketMediaStreamProducerImpl(): %p\n", this);

    sinks_[0] = sinks_[1] = 0;

    
}

/*virtual*/WebSocketMediaStreamProducerImpl::~WebSocketMediaStreamProducerImpl()
{
    fprintf(stderr, "WebSocketMediaStreamProducerImpl::~WebSocketMediaStreamProducerImpl(): %p\n", this);
    if (wsiContext_)
        lws_context_destroy(wsiContext_);

    if (sinks_[0])
        fPlaybackController->removeSink(sinks_[0]);

    if (sinks_[1])
        fPlaybackController->removeSink(sinks_[1]);

    shutdown();

    if (fPlaybackController)
    {
        Medium::close(fPlaybackController);
        fPlaybackController = NULL;
    }

    if (sinks_[0])
    {
        Medium::close(sinks_[0]);
        sinks_[0] = 0;
    }

    if (sinks_[1])
    {
        Medium::close(sinks_[1]);
        sinks_[1] = 0;
    }

    if(hevc_vconf_)
    {
        delete[] hevc_vconf_;
        hevc_vconf_ = NULL;
    }
}

/*virtual*/void WebSocketMediaStreamProducerImpl::open()
{
    bzero(&protocols_[0], sizeof(protocols_[0]));
    protocols_[0].name = "";
    protocols_[0].callback = WebSocketMediaStreamProducerImpl::lws_protocol_callback;
    protocols_[0].per_session_data_size = sizeof(void*);
    protocols_[0].rx_buffer_size = 1024*1024;
    protocols_[0].user = this;
    bzero(&protocols_[1], sizeof(protocols_[1])); /* terminator */


    struct lws_context_creation_info info = {};
    info.port = CONTEXT_PORT_NO_LISTEN;
    info.protocols = protocols_;
    info.pt_serv_buf_size = 32 * 1024;
    info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
//    info.client_ssl_ca_filepath = "/private/etc/ssl/cert.pem";
//    info.client_ssl_ca_filepath = "/etc/pki/tls/cert.pem";

    // info.ssl_cipher_list =     "ECDHE-ECDSA-AES256-GCM-SHA384:"
    //     		       "ECDHE-RSA-AES256-GCM-SHA384:"
    //     		       "DHE-RSA-AES256-GCM-SHA384:"
    //     		       "ECDHE-RSA-AES256-SHA384:"
    //     		       "HIGH:!aNULL:!eNULL:!EXPORT:"
    //     		       "!DES:!MD5:!PSK:!RC4:!HMAC_SHA1:"
    //     		       "!SHA1:!DHE-RSA-AES128-GCM-SHA256:"
    //     		       "!DHE-RSA-AES128-SHA256:"
    //     		       "!AES128-GCM-SHA256:"
    //     		       "!AES128-SHA256:"
    //     		       "!DHE-RSA-AES256-SHA256:"
    //     		       "!AES256-GCM-SHA384:"
    //     		       "!AES256-SHA256";

    wsiContext_ = lws_create_context(&info);
    if (!wsiContext_)
    {
        fEnv->setResultErr(EINTERNALERROR);
        changeState(STOPPED, EINTERNALERROR);

        fEnv->log("lws init failed\n");
        return;
    }

    lwsServiceTask_ = fEnv->taskScheduler()
        .scheduleDelayedTask(5000,
                             (TaskFunc*)lwsTask, this);

    changeState(OPENING);

    fEnv->taskScheduler().doEventLoop(&fStopFlag); // does not return

    printf("WebSocketMediaStreamProducerImpl::open() DONE\n");
}

void WebSocketMediaStreamProducerImpl::lwsTask(void* clientData)
{
    WebSocketMediaStreamProducerImpl* thiz = (WebSocketMediaStreamProducerImpl*)clientData;

    lws_service(thiz->wsiContext_, 1);

    thiz->fMutex.lock();
    if (!thiz->requests_.empty())
    {
        thiz->fMutex.unlock();
        lws_callback_all_protocol(thiz->wsiContext_,
                                  &thiz->protocols_[0], LWS_CALLBACK_USER);
    }
    else
    {
        thiz->fMutex.unlock();
    }

    thiz->lwsServiceTask_ = thiz->fEnv->taskScheduler()
        .scheduleDelayedTask(5000,
                             (TaskFunc*)lwsTask, thiz);
}

/*virtual*/void WebSocketMediaStreamProducerImpl::pause()
{
    if (!fPlaybackController->hasSinks())
        return;

    if (fPlayingArchive && !fStepMode)
    {
        fPlaybackController->setPlayFramesFromServer(false);
        fPlaybackController->setPaused(true);
        fPlaybackController->setServerPaused(true);
        clientPause();
    }
}

/*virtual*/void WebSocketMediaStreamProducerImpl::resume()
{
    if (!fPlaybackController->hasSinks())
        return;

    fPlaybackController->setServerPaused(false);
    fPlaybackController->setPaused(false);

    clientPlay();
}

void WebSocketMediaStreamProducerImpl::jumpToTimestamp(int timestamp, bool mark)
{
   if (!fPlaybackController->hasSinks())
       return;

   if (!fPlayingArchive)
       return;

   jumpToTimestamp_ = timestamp;

   clientJump();
}

/*virtual*/ void WebSocketMediaStreamProducerImpl::setSpeed(float speed)
{
    if (!fPlaybackController->hasSinks())
       return;

   if (!fPlayingArchive)
       return;

   if (speed_ == speed)
       return;

    speed_ = speed;

    clientSetSpeed();
}

/*virtual*/ void WebSocketMediaStreamProducerImpl::muteAudio(bool mute)
{
    audioMuted_ = mute;

    if (!sinks_[1])
        return;

    fPlaybackController->audioSink()->mute(mute);
}

void WebSocketMediaStreamProducerImpl::shutdown()
{
    MutexGuard g(fCacheMutex);

    if (fd_ > 0)
    {
        fEnv->taskScheduler().turnOffBackgroundReadHandling(fd_);
        fd_ = 0;
    }

    if (fDidShutdown)
    {
        return;
    }

    fEnv->taskScheduler().unscheduleDelayedTask(lwsServiceTask_);
}

/*static*/ int WebSocketMediaStreamProducerImpl::lws_protocol_callback(struct lws *wsi, enum lws_callback_reasons reason,
                                                                       void *user, void *in, size_t len)
{
    const struct lws_protocols *p = lws_get_protocol(wsi);

    if (!p)
        return 0;
    WebSocketMediaStreamProducerImpl *thiz = (WebSocketMediaStreamProducerImpl *)p->user;

    //    fprintf(stderr, "p=%p, reason=%d\n", p, reason);
    //    thiz->fEnv->log("p=%p, reason=%d\n", p, reason);
    switch (reason)
    {
        case LWS_CALLBACK_PROTOCOL_INIT:
        {
            fprintf(stderr, "LWS_CALLBACK_PROTOCOL_INIT: thiz: %p\n", thiz);
            thiz->clientConnect(wsi);
            break;
        }
        case LWS_CALLBACK_CLIENT_ESTABLISHED:
            fprintf(stderr, "LWS_CALLBACK_CLIENT_ESTABLISHED: thiz: %p\n", thiz);
            thiz->clientInit();
            break;

        case LWS_CALLBACK_CLIENT_WRITEABLE:
        {
            //fprintf(stderr, "LWS_CALLBACK_CLIENT_WRITEABLE: thiz: %p\n", thiz);
            //thiz->fEnv->log("LWS_CALLBACK_CLIENT_WRITEABLE: thiz: %p\n", thiz);

            MutexGuard g(thiz->fMutex);
            if (!thiz->requests_.empty())
            {
                Json::FastWriter fastWriter;
                Json::Value r = thiz->requests_.front();
                thiz->requests_.pop();
                std::string msg = std::string(LWS_PRE, ' ') + fastWriter.write(r);

                thiz->fEnv->log("Send request: %s\n", msg.c_str());

                thiz->fMutex.unlock();
                int m = lws_write(wsi, (unsigned char*)msg.data() + LWS_PRE, msg.size() - LWS_PRE,
                                  LWS_WRITE_TEXT);
                thiz->fMutex.lock();
                if (m < (int)(msg.size() - LWS_PRE))
                {
                    thiz->fEnv->log("ERROR writing to ws socket (%d bytes)\n", m);
                    thiz->fEnv->setResultMsg("Error writing to ws socket");
                    thiz->changeState(STOPPED, EINTERNALERROR);
                }
            }
            break;
        }
        case LWS_CALLBACK_CLIENT_RECEIVE:
//            fprintf(stderr, "LWS_CALLBACK_CLIENT_RECEIVE: thiz: %p, %ld bytes\n", thiz, len);
//            thiz->fEnv->log("LWS_CALLBACK_CLIENT_RECEIVE: thiz: %p, %ld bytes\n", thiz, len);
            thiz->clientReceive((const char*)in, len);
            break;

        case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
            thiz->fEnv->log("CLIENT_CONNECTION_ERROR: %s\n",
                      in ? (char *)in : "(null)");
            if (in)
                thiz->fEnv->setResultMsg((char*)in);
            thiz->changeState(STOPPED, EINTERNALERROR);
            break;

        case LWS_CALLBACK_WSI_DESTROY:
            if (thiz->fd_ > 0)
            {
                thiz->fEnv->taskScheduler().turnOffBackgroundReadHandling(thiz->fd_);
                thiz->fd_ = 0;
            }

            if (!thiz->fStopFlag)
            {
                fprintf(stderr, "LWS_CALLBACK_WSI_DESTROY: thiz: %p\n", thiz);
                thiz->fEnv->setResultMsg("Unexpected end of stream");
                thiz->changeState(STOPPED, EINTERNALERROR);
            }
            break;

        case LWS_CALLBACK_USER:
        {
            fprintf(stderr, "LWS_CALLBACK_USER: thiz: %p\n", thiz);
            lws_callback_on_writable(wsi);
            break;
        }

	default:
            break;
    }

    return 0;
}

/*static*/ void WebSocketMediaStreamProducerImpl::lws_log_callback(int level, const char *line)
{
    fprintf(stderr, "%s", line);
}


int WebSocketMediaStreamProducerImpl::clientConnect(struct lws *wsi)
{
    fprintf(stderr, "URL: %s\n", url_.c_str());
    char proto[16] = {0};
    char authorization[128] = {0};
    char hostname[128] = {0};
    int  port = 0;
    char path[1024] = {0};
    url_split(proto, sizeof proto,
              authorization, sizeof authorization,
              hostname, sizeof hostname,
              &port,
              path, sizeof path, url_.c_str());



    std::vector<std::string> t;
    tokenize(path, "?", &t);

    assert(t.size());

    if (t.size() == 2)
    {
        std::string query = t[1];
        t.clear();
        tokenize(query, "&", &t);
        for (std::vector<std::string>::iterator it = t.begin(); it != t.end(); ++it)
        {
            // LOG_DEBUG("Processing: %s", it->c_str());
            std::vector<std::string> kv;
            tokenize(*it, "=", &kv);
            if (kv.empty())
                continue;

            if (kv[0] == "cloud" && (kv[1] == "true" || kv[1] == "1"))
            {
                cloud_ = true;
                continue;
            }

            if (kv[0] == "speed")
            {
                speed_ = atof(kv[1].c_str());
                if (fPlaybackController)
                    fPlaybackController->enableJitterBuffer(speed_ == 1.0f);
            }

            // LOG_DEBUG("Inserting '%s' => '%s", kv[0].c_str(), kv[1].c_str());
            requestParams_.insert(std::make_pair(kv[0], kv.size() == 1 ? "" : kv[1]));
        }
    }

    fEnv->log("proto: %s, hostname: %s, port: %d, path: %s, cloud: %d\n", proto, hostname, port, path, cloud_);

    // sanity checks
    if (requestParams_["objid"].empty()
        || requestParams_["streamnum"].empty())
    {
        fEnv->setResultErrMsg("Missing 'objid' or 'streamnum' argument in the URL");
        changeState(STOPPED, EINTERNALERROR);
        return -1;
    }

    struct lws_client_connect_info i = {};
    i.context = wsiContext_;
    i.port    = port;
    i.address = hostname;
    i.path    = "/";
    i.host    = i.address;
    i.origin  = i.address;
    if (url_.find("wss://") != std::string::npos)
        i.ssl_connection = LCCSCF_USE_SSL|LCCSCF_ALLOW_SELFSIGNED|LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK|LCCSCF_ALLOW_EXPIRED;
    i.protocol = protocols_[0].name;
    i.pwsi = &wsi;

    fEnv->log("Connecting to %s:%d\n", i.address, i.port);

    if (fd_ > 0)
    {
        fEnv->taskScheduler().turnOffBackgroundReadHandling(fd_);
        fd_ = 0;
    }

    void* r = lws_client_connect_via_info(&i);
    if(r)
    {
        fd_ = lws_get_socket_fd(wsi);
        fprintf(stderr, "SOCKET_FD=%d, wsi=%p\n", fd_, r);
        fEnv->taskScheduler().turnOnBackgroundReadHandling(fd_, (TaskScheduler::BackgroundHandlerProc*)&incomingDataHandler, this);
    }

    return !r;
}

int WebSocketMediaStreamProducerImpl::clientInit()
{
    MutexGuard g(fMutex);

    lastRequest_.clear();
    lastRequest_["cmd"]      = "init";
    lastRequest_["objid"]    = requestParams_["objid"];
    lastRequest_["streamid"] = requestParams_["streamnum"];
    lastRequest_["ticket"]   = requestParams_["ticket"];
    lastRequest_["params"]["raw"]   = true;
    lastRequest_["params"]["audio"] = (requestParams_.find("audio") != requestParams_.end() && requestParams_["audio"]=="1");
    if (!requestParams_["startTime"].empty() && !requestParams_["endTime"].empty())
    {
        lastRequest_["params"]["startTime"] = Json::UInt64(atoll(requestParams_["startTime"].c_str()));
        lastRequest_["params"]["endTime"]   = Json::UInt64(atoll(requestParams_["endTime"].c_str()));
        lastRequest_["params"]["speed"]     = speed_;
    }

    if (fPlaybackController)
        fPlaybackController->enableJitterBuffer(speed_ == 1.0f);

    requests_.push(lastRequest_);

    lws_callback_all_protocol(wsiContext_,
                              &protocols_[0], LWS_CALLBACK_USER);
    return 0;
}

int WebSocketMediaStreamProducerImpl::clientPlay()
{
    MutexGuard g(fMutex);

    lastRequest_.clear();
    lastRequest_["cmd"]      = "play";

    requests_.push(lastRequest_);

    return 0;
}

int WebSocketMediaStreamProducerImpl::clientPause()
{
    MutexGuard g(fMutex);

    lastRequest_.clear();
    lastRequest_["cmd"]      = "pause";

    requests_.push(lastRequest_);

    return 0;
}

int WebSocketMediaStreamProducerImpl::clientJump()
{
    MutexGuard g(fMutex);

    lastRequest_.clear();
    lastRequest_["cmd"]      = "set";
    lastRequest_["params"]["ts"] = (uint64_t)atoll(unixtsToDatetime(jumpToTimestamp_).c_str());

    requests_.push(lastRequest_);

    if (fPlaybackController)
    {
        fPlaybackController->flushJitterBuffer();
    }

    return 0;
}

int WebSocketMediaStreamProducerImpl::clientSetSpeed()
{
    MutexGuard g(fMutex);

    lastRequest_.clear();
    lastRequest_["cmd"]      = "set";
    lastRequest_["params"]["speed"] = speed_;

    if (fPlaybackController)
        fPlaybackController->enableJitterBuffer(speed_ == 1.0f);

    requests_.push(lastRequest_);


    return 0;
}

/*static*/ void WebSocketMediaStreamProducerImpl::incomingDataHandler(WebSocketMediaStreamProducerImpl* src, int mask)
{
    //fprintf(stderr, "\n*** incomingDataHandler(): mask=%d ***\n", mask);
    if (SOCKET_READABLE == (mask & SOCKET_READABLE))
        lws_service(src->wsiContext_, 1);
}

int WebSocketMediaStreamProducerImpl::clientReceive(const char *in, size_t len)
{
    if (in[0] == '{') // json
    {
        std::string msg(in, len);

//        fEnv->log("Got js: %s\n", msg.c_str());
//        fprintf(stderr, "Got js: %s\n", msg.c_str());

        Json::Value js;
        Json::Reader reader;
        if (!reader.parse(msg, js))
        {
            fEnv->log("Failed to parse json: %s\n", reader.getFormatedErrorMessages().c_str());
            changeState(STOPPED, EINTERNALERROR);
            return -1;
        }

        if (!js["error"].empty())
        {
            fEnv->setResultMsg(js["error"].asString().c_str());
            changeState(STOPPED, EINTERNALERROR);
            return -1;
        }

        if (lastRequest_["cmd"] == "init")
        {
            fEnv->log("Got js: %s\n", msg.c_str());

            if (js["mime"].asString().find("video/x-jpeg") != std::string::npos)
            {
                StreamInfo *streamInfo = new StreamInfo;
                streamInfo->streamId = streamId_;
                streamInfo->RTPCodecStr = "JPEG";
                streamInfo->configData = 0;
                streamInfo->configDataSize = 0;
                streamInfo->duration = 0;
                streamInfo->type = VIDEO;

                sinks_[0]  = VideoFrameConsumer::createNew(*fEnv, fPlaybackController,
                                                           fFrameSinkBufferSize,
                                                           decoderType_, preferedPixelFormat_,
                                                           streamInfo, fVideoTranscoded);

                fPlaybackController->addSink(sinks_[0], AudioVideoPlaybackController::VIDEO);

            }
            else if (js["mime"].asString().find("video/mp4") != std::string::npos)
            {
                StreamInfo *streamInfo = new StreamInfo;
                streamInfo->streamId = streamId_;
                streamInfo->RTPCodecStr = "H264";
                streamInfo->configData = 0;
                streamInfo->configDataSize = 0;
                streamInfo->duration = 0;
                streamInfo->type = VIDEO;

                sinks_[0]  = VideoFrameConsumer::createNew(*fEnv, fPlaybackController,
                                                          fFrameSinkBufferSize,
                                                          decoderType_, preferedPixelFormat_,
                                                          streamInfo, fVideoTranscoded);

                unsigned vconf_size = 0;
                unsigned char* vconf = NULL;
                if(!js["vconf"].empty() && (vconf = base64Decode((char*)js["vconf"].asString().c_str(), vconf_size)) && vconf_size > 4)
                {
            	    //Json::FastWriter writer;
            	    //fprintf(stderr, "vconf=%p, sz=%ld, vconf: %s\n", vconf, vconf_size, writer.write(js["vconf"]).c_str());
                    //unsigned vconf_size;
                    //unsigned char *vconf = base64Decode((char*)js["vconf"].asString().c_str(), vconf_size);

                    fprintf(stderr, "vconf:\n");
                    for (unsigned i= 0; i < (vconf_size>180?180:vconf_size); i++)
                    {
                        fprintf(stderr, "%02X ", (uint8_t)((unsigned char*)vconf)[i]);
                    }
                    fprintf(stderr, "\n");

                    fPlaybackController->addSink(sinks_[0], AudioVideoPlaybackController::VIDEO);

                    sinks_[0]->addExtraData(vconf+4, vconf_size-4);

                    delete[] vconf;
                } else
                    return 0;

                unsigned aconf_size = 0;
                unsigned char* aconf = NULL;
                if (!js["aconf"].empty() && (aconf = base64Decode((char*)js["aconf"].asString().c_str(), aconf_size)))
                {
                    fprintf(stderr, "aconf:\n");
                    for (unsigned i= 0; i < (aconf_size>180?180:aconf_size); i++)
                    {
                        fprintf(stderr, "%02X ", (uint8_t)((unsigned char*)aconf)[i]);
                    }
                    fprintf(stderr, "\n");


            	    //Json::FastWriter writer;
            	    //fprintf(stderr, "aconf: %s\n", writer.write(js["aconf"]).c_str());
                    StreamInfo *streamInfo = new StreamInfo;
                    streamInfo->streamId = streamId_;
                    streamInfo->RTPCodecStr = "MPEG4-GENERIC";
                    streamInfo->duration = 0;
                    streamInfo->type = AUDIO;

                    streamInfo->configData = new unsigned char[aconf_size+8];
                    memcpy(streamInfo->configData, aconf, aconf_size);
                    streamInfo->configDataSize = aconf_size;
                    delete[] aconf;

                    sinks_[1]  = AudioFrameConsumer::createNew(*fEnv, fPlaybackController,
                                                               fFrameSinkBufferSize,
                                                               decoderType_, streamInfo);
                    fPlaybackController->addSink(sinks_[1], AudioVideoPlaybackController::AUDIO);
                    sinks_[1]->mute(audioMuted_);
                }
            }
            else if (js["mime"].asString().find("video/x-hevc") != std::string::npos)
            {
                StreamInfo *streamInfo = new StreamInfo;
                streamInfo->streamId = streamId_;
                streamInfo->RTPCodecStr = "HEVC";
                streamInfo->configData = 0;
                streamInfo->configDataSize = 0;
                streamInfo->duration = 0;
                streamInfo->type = VIDEO;

                sinks_[0]  = VideoFrameConsumer::createNew(*fEnv, fPlaybackController,
                                                          fFrameSinkBufferSize,
                                                          decoderType_, preferedPixelFormat_,
                                                          streamInfo, fVideoTranscoded);

                unsigned vconf_size = 0;
                unsigned char* vconf = NULL;
                if(!js["vconf"].empty() && (vconf = base64Decode((char*)js["vconf"].asString().c_str(), vconf_size)) && vconf_size > 4)
                {
                    if (hevc_vconf_)
                        delete[] hevc_vconf_;
                    hevc_vconf_ = base64Decode((char*)js["vconf"].asString().c_str(), hevc_vconf_size_);
            	    //Json::FastWriter writer;
            	    //fprintf(stderr, "vconf=%p, sz=%ld, vconf: %s\n", vconf, vconf_size, writer.write(js["vconf"]).c_str());
                    //unsigned vconf_size;
                    //unsigned char *vconf = base64Decode((char*)js["vconf"].asString().c_str(), vconf_size);

                    fprintf(stderr, "vconf: %d bytes\n", vconf_size);
                    for (unsigned i= 0; i < (vconf_size>180?180:vconf_size); i++)
                    {
                        fprintf(stderr, "%02X ", (uint8_t)((unsigned char*)vconf)[i]);
                    }
                    fprintf(stderr, "\n");

                    fPlaybackController->addSink(sinks_[0], AudioVideoPlaybackController::VIDEO);

                    sinks_[0]->addExtraData(vconf+4, vconf_size-4);

                    delete[] vconf;
                } else
                    return 0;

                unsigned aconf_size = 0;
                unsigned char* aconf = NULL;
                if (!js["aconf"].empty() && (aconf = base64Decode((char*)js["aconf"].asString().c_str(), aconf_size)))
                {
                    fprintf(stderr, "aconf:\n");
                    for (unsigned i= 0; i < (aconf_size>180?180:aconf_size); i++)
                    {
                        fprintf(stderr, "%02X ", (uint8_t)((unsigned char*)aconf)[i]);
                    }
                    fprintf(stderr, "\n");


            	    //Json::FastWriter writer;
            	    //fprintf(stderr, "aconf: %s\n", writer.write(js["aconf"]).c_str());
                    StreamInfo *streamInfo = new StreamInfo;
                    streamInfo->streamId = streamId_;
                    streamInfo->RTPCodecStr = "MPEG4-GENERIC";
                    streamInfo->duration = 0;
                    streamInfo->type = AUDIO;

                    streamInfo->configData = new unsigned char[aconf_size+8];
                    memcpy(streamInfo->configData, aconf, aconf_size);
                    streamInfo->configDataSize = aconf_size;
                    delete[] aconf;

                    sinks_[1]  = AudioFrameConsumer::createNew(*fEnv, fPlaybackController,
                                                               fFrameSinkBufferSize,
                                                               decoderType_, streamInfo);
                    fPlaybackController->addSink(sinks_[1], AudioVideoPlaybackController::AUDIO);
                    sinks_[1]->mute(audioMuted_);
                }
            }
            else
            {
                fEnv->setResultMsg("Unknown mime type");
                changeState(STOPPED, EINTERNALERROR);
                return -1;
            }

            for (int i = 0; i < 2; i++)
            {
                if (sinks_[i] && !sinks_[i]->init())
                {
                    delete sinks_[i];
                    sinks_[i] = 0;
                }
            }

            changeState(OPENED);
        }
        else
        {
            if (js["pts"].isString())
                datetimeToTimeval(js["pts"].asString().c_str(), pts_);

            struct timeval tv;
            Json::FastWriter jsWriter;
            Json::Value& jsMetadata = js["metadata"];
            if(!jsMetadata.isNull()) {
                Json::Value* pjs = &jsMetadata;
                bool bArr = jsMetadata.isArray();
                size_t count = bArr ? jsMetadata.size()
                                    : jsMetadata.isObject() ? 1 : 0;
                for(Json::Value::ArrayIndex i = 0; i < count; i++) {
                    if(bArr)
                        pjs = &jsMetadata[i];
                    Json::Value& jsPTS = (*pjs)["pts"];
                    Json::Value& jsList = (*pjs)["list"];
                    if(!jsPTS.isString() || !jsList.isObject() || jsList.empty())
                        continue;

                    datetimeToTimeval(jsPTS.asString().c_str(), tv);

                    std::string objsData = jsWriter.write(*pjs);
                    if(*objsData.rbegin() == '\n')
                        objsData.resize(objsData.size() - 1);
                    // auto ret // 'auto' type specifier is a C++11 extension
                    std::pair<OBJECTS_DATA_ASCENDING::iterator, bool> ret
                            = objectsData_.insert(std::make_pair(pts_.tv_sec > tv.tv_sec || (pts_.tv_sec == tv.tv_sec && pts_.tv_usec >= tv.tv_usec)
                                                                 ? pts_ : tv, std::string("[") + objsData));
                    if(!ret.second)
                        ret.first->second.append(std::string(",") + objsData);
                }
            }

            if (js["event"].isString())
            {
                if (js["event"] == "EOS")
                {
                    fEnv->log("End of stream\n");
                    fEnv->setResultErr(EENDOFSTREAM);
                    fPlaybackController->setState2(IDLE, EENDOFSTREAM);
                }
                else if (js["event"] == "EOA")
                {
                    fEnv->log("End of archive\n");
                    fEnv->setResultErr(EENDOFARCHIVE);
                    fPlaybackController->setState2(IDLE, EENDOFARCHIVE);
                }
                else if (js["event"] == "EOC")
                {
                    // FIXME: will not happen until there is appropriate logic in the streamer. Cut also here
                    fEnv->log("End of chunk\n");
                    fprintf(stderr, "End of chunk\n");
                    fEnv->setResultErr(EENDOFCHUNK);
                    changeState(IDLE, EENDOFCHUNK);
                }
            }
        }
    }
    else // frame
    {
        // fEnv->log("Got binary data %d bytes\n", len);
        // fprintf(stderr, "Got binary data\n");
        // for (int i= 0; i < (len>200?200:len); i++)
        // {
        //     fprintf(stderr, "%02X ", (uint8_t)((unsigned char*)in)[i]);
        // }
        // fprintf(stderr, "\n");


        // static int iii = 0;
        // if (!iii++)
        //     return 0;


        // struct timeval pts;
        // gettimeofday(&pts, 0);

        if (in[0] == 0) // VIDEO
        {
            // auto p // 'auto' type specifier is a C++11 extension
            bool sent(false);
            std::pair<OBJECTS_DATA_ASCENDING::iterator, OBJECTS_DATA_ASCENDING::iterator> p
                    = objectsData_.equal_range(pts_);
            if((*(unsigned char*)&in[5] >> 1 & 0x3F) == 32 && len-5 > hevc_vconf_size_) {
                bool equal(true);
                for (unsigned i=4; i<hevc_vconf_size_; i++) {
                    if (*(unsigned char*)&in[5+i-4] != *(unsigned char*)&hevc_vconf_[i])
                    {
                        equal = false;
                        break;
                    }
                }
                uint8_t* ptr = (uint8_t*)&in[5+hevc_vconf_size_-4];
                if(equal && ptr[0] == 0 && ptr[1] == 0 && ptr[2] == 0 && ptr[3] == 1)
                {
                    fPlaybackController->videoSink()->addData((unsigned char*)&in[5], hevc_vconf_size_, pts_, false, NULL);
                    fPlaybackController->videoSink()->addData((unsigned char*)&in[5+hevc_vconf_size_+4], len-5-hevc_vconf_size_-4, pts_, false, p.first != objectsData_.end() ? &p.first->second.append("]") : NULL);
                    sent = true;
                }
            }
            if(!sent)
                fPlaybackController->videoSink()->addData((unsigned char*)&in[5], len-5, pts_, false, p.first != objectsData_.end() ? &p.first->second.append("]") : NULL);
            objectsData_.erase(objectsData_.begin(),
                               p.first == p.second && p.second != objectsData_.end() ? ++p.second : p.second);
        }
        else
        {
            if (sinks_[1])
                fPlaybackController->audioSink()->addData((unsigned char*)&in[1], len-1, pts_, false);
        }

    }

    return 0;
}

}}
