/*
 * Copyright (C) 2025 Amlogic, Inc. All rights reserved.
 *
 * All information contained herein is Amlogic confidential.
 *
 * This software is provided to you pursuant to Software License Agreement
 * (SLA) with Amlogic Inc ("Amlogic"). This software may be used
 * only in accordance with the terms of this agreement.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification is strictly prohibited without prior written permission from
 * Amlogic.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
#define LOG_TAG "CcSourceManager"

#include "CcSourceManager.h"

#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <sys/ioctl.h>
#include <sys/utsname.h>
#include <unistd.h>

#include "AmlogicUtil.h"
#include "CcSourceMediaProxy.h"
#include "SubtitleLog.h"
#include "UserdataDriver.h"
#include "VbiDriver.h"

#define AMSTREAM_IOC_MAGIC             'S'
#define AMSTREAM_IOC_UD_LENGTH         _IOR(AMSTREAM_IOC_MAGIC, 0x54, unsigned long)
#define AMSTREAM_IOC_UD_POC            _IOR(AMSTREAM_IOC_MAGIC, 0x55, int)
#define AMSTREAM_IOC_UD_FLUSH_USERDATA _IOR(AMSTREAM_IOC_MAGIC, 0x56, int)
#define AMSTREAM_IOC_UD_BUF_READ       _IOR(AMSTREAM_IOC_MAGIC, 0x57, int)
#define AMSTREAM_IOC_UD_AVAILABLE_VDEC _IOR(AMSTREAM_IOC_MAGIC, 0x5c, unsigned int)

#ifdef NEED_CCDATA_FROM_CCDATABASE
#define DAS_RET_CHK() \
    do { \
        if (ret == 0) { \
            continue; \
        } else if (ret == -1) { \
            SUBTITLE_LOGE("bad parameter"); \
            continue; \
        } else if (ret <= -3) { \
            SUBTITLE_LOGE("dataserver disconnect"); \
            ctx->das = NULL; \
            stage = STAGE_INIT; \
            continue; \
        } \
    } while (0)

#define DAS_STAGE_CHK() \
    do { \
        switch (stage) { \
            case STAGE_RB: \
                stage = STAGE_VDECID; \
                break; \
            case STAGE_VDECID: \
                stage = STAGE_SRC; \
                break; \
            case STAGE_SRC: \
                stage = STAGE_DAT; \
                break; \
            default: \
                stage = STAGE_DAT; \
                break; \
        } \
    } while (0)
#endif

static int getKernelVersion(void) {
    int version, subversion, patchlevel;
    struct utsname utsn;
    /* Return 0 on failure, and attempt to probe with empty kernal version */
    if (uname(&utsn)) return 0;
    if (sscanf(utsn.release, "%d.%d.%d", &version, &subversion, &patchlevel) != 3) return 0;
    return version;
}

static void swapData(uint8_t *userData, int udSize) {
    int swapBlocks = -1;
    int i = 0, j = 0, k = 0, m = 0;
    unsigned char cTemp = '\0';
    swapBlocks = udSize >> 3;
    // Swap byte order
    for (i = 0; i < swapBlocks; i++) {
        j = i << 3;
        k = j + 7;
        for (m = 0; m < 4; m++) {
            cTemp = userData[j];
            userData[j++] = userData[k];
            userData[k--] = cTemp;
        }
    }
}

static const char *retToString(int ret) {
    if (ret == -2) {
        return "timeout";
    } else if (ret == -3) {
        return "disconnect";
    }

    static char numStr[20] = {0};
    sprintf(numStr, "%d", ret);
    return numStr;
}

static void dumpCcData(char *who, uint8_t *buff, int size) {
    int i = -1;
    char buf[4096] = {'\0'};

    if (size > 1024) size = 1024;
    for (i = 0; i < size; i++) {
        sprintf(buf + i * 3, "%02x ", buff[i]);
    }
    SUBTITLE_LOGI("[ccSource]: %s, dataSize:%d data:%s", who, size, buf);
}

static void *ccUserdataDeviceTask(void *args) {
    int fd = -1;
    int r = -1, ret = -1;
    uint8_t data[MAX_CC_DATA_LEN] = {'\0'};
    int flush = 1;
    int vdecIds = -1;
    int readVdecId = -1;
    int kernelVersion = -1;
    struct pollfd pfd;
    struct CcSourceUserdataParam userParaInfo;
    CcSourceCtx *ctx = (CcSourceCtx *)args;

    if (!ctx->running) return NULL;

    fd = open(AMSTREAM_USDRDATA_DEV_FILE, O_RDONLY);
    if (fd == -1) {
        SUBTITLE_LOGE("cannot open userdata device");
        return NULL;
    }

    pfd.events = POLLIN | POLLERR;
    pfd.fd = fd;
    kernelVersion = getKernelVersion();
    while (ctx->running) {
        ret = poll(&pfd, 1, USERDATA_POLL_TIMEOUT);
        if (ret != 1) continue;

        if (!ctx->running) break;

        if (!(pfd.revents & POLLIN)) continue;
        // For multi-instances supports
        vdecIds = 0;
        if (-1 == ioctl(fd, AMSTREAM_IOC_UD_AVAILABLE_VDEC, &vdecIds)) {
            SUBTITLE_LOGE("get available vdec failed");
            continue;
        }

        int buildVersion = property_get_int32("ro.build.version.sdk", 28);
        if (kernelVersion >= 5 || buildVersion >= 29) {
            SUBTITLE_LOGD("kernelVersion:%d, buildVersion:%d", kernelVersion, buildVersion);
            readVdecId = vdecIds;
        } else {
            readVdecId = ffs(vdecIds) - 1;
        }

        if (flush) {
            ioctl(fd, AMSTREAM_IOC_UD_FLUSH_USERDATA, &readVdecId);
            flush = 0;
            continue;
        }

        do {
            if (!ctx->running) break;

            memset(&userParaInfo, 0, sizeof(struct CcSourceUserdataParam));
            userParaInfo.pbuf_addr = (void *)(size_t)data;
            userParaInfo.buf_len = sizeof(data);
            userParaInfo.instance_id = readVdecId;

            if (-1 == ioctl(fd, AMSTREAM_IOC_UD_BUF_READ, &userParaInfo)) {
                SUBTITLE_LOGE("call AMSTREAM_IOC_UD_BUF_READ failed\n");
                continue;
            }

            r = userParaInfo.data_size;
            if (r > MAX_CC_DATA_LEN) {
                userParaInfo.data_size = MAX_CC_DATA_LEN;
                r = MAX_CC_DATA_LEN;
            }

            if (r <= 0) continue;

            if (isSubtitleDumpAvailable()) dumpCcData("userdata ioctl", data, r);

            swapData(data, r);
            if (ctx->param.uDataCb) {
                if (isSubtitleDumpAvailable()) dumpCcData("userdata after swap", data, r);
                ctx->param.uDataCb(ctx, &userParaInfo);
            }
        } while (userParaInfo.meta_info.records_in_que > 1);
    }
    close(fd);
    return NULL;
}

static void *ccVbiDeviceTask(void *args) {
    struct vbi_data_s vbi[50];
    int fd = -1;
    int type = VBI_TYPE_USCC;
    int last = -1;
    int ret = -1;
    CcSourceCtx *ctx = (CcSourceCtx *)args;

    fd = open(CC_VBI_DEV_FILE, O_RDWR);
    if (fd == -1) {
        SUBTITLE_LOGE("cannot open \"%s\"", CC_VBI_DEV_FILE);
        return NULL;
    }

    if (ioctl(fd, VBI_IOC_SET_TYPE, &type) == -1)
        SUBTITLE_LOGI("VBI_IOC_SET_TYPE error:%s", strerror(errno));
    if (ioctl(fd, VBI_IOC_START) == -1) SUBTITLE_LOGI("VBI_IOC_START error:%s", strerror(errno));

    AmlogicGetClock(&last);
    ctx->vCtx.last = last;

    while (ctx->running) {
        struct pollfd pfd;
        pfd.fd = fd;
        pfd.events = POLLIN;

        ret = poll(&pfd, 1, CC_POLL_TIMEOUT);
        if (ctx->running && (ret > 0) && (pfd.events & POLLIN)) {
            struct vbi_data_s *pd;

            ret = read(fd, vbi, sizeof(vbi));
            pd = vbi;
            SUBTITLE_LOGI("am_vbi_data_thread running read data == %d", ret);

            if (ret >= (int)sizeof(struct vbi_data_s)) {
                while (ret >= (int)sizeof(struct vbi_data_s)) {
                    SUBTITLE_LOGD("device vbi line %d: %02x %02x", pd->line_num, pd->b[0],
                                  pd->b[1]);
                    if (ctx->param.vDataCb) {
                        ctx->param.vDataCb(ctx, pd);
                    }
                    pd++;
                    ret -= sizeof(struct vbi_data_s);
                }
            }
            if (ctx->param.vEvtHandler) {
                ctx->param.vEvtHandler(ctx);
            }
        }
    }
    SUBTITLE_LOGI("am_vbi_data_thread exit");
    ioctl(fd, VBI_IOC_STOP);
    close(fd);
    return NULL;
}

static void v4lInfoToUserdataParam(struct v4l_stream_common_s *v4lInfo,
                                   struct CcSourceUserdataParam *pUserdataParam) {
    memset(pUserdataParam, 0, sizeof(struct CcSourceUserdataParam));

    pUserdataParam->version = v4lInfo->version;
    pUserdataParam->instance_id = v4lInfo->instance_id;
    pUserdataParam->data_size = v4lInfo->data_size;

    // Meta data convert
    struct userdata_meta_info_t *userdataInfo = &pUserdataParam->meta_info;
    struct userdata_cc_meta_info_t *v4lMeta = &v4lInfo->cc_meta_data;
    *userdataInfo = *(struct userdata_meta_info_t *)v4lMeta;

    // Buffer data
    pUserdataParam->pbuf_addr = &v4lInfo->data[0];
}

static void *ccMediaProxyDeviceTask(void *args) {
    CcSourceCtx *ctx = (CcSourceCtx *)args;
    if (!ctx || !ctx->running) return NULL;

    int fd = -1;
    fd = open(MEDIA_PROXY_DEV_FILE, O_RDONLY);
    if (fd == -1) {
        SUBTITLE_LOGE("cannot open mediaproxy device");
        return NULL;
    }

    int len = 64;
    if (ioctl(fd, MEDIAPROXY_SET_FIFO_LEN, &len) < 0) {
        SUBTITLE_LOGE("Failed to set fifo len %d", len);
        close(fd);
        return NULL;
    }

    int msgType = MEDIA_VIDEO_CC_INFO;
    if (ioctl(fd, MEDIAPROXY_SUBSCRIBE, &msgType) < 0) {
        SUBTITLE_LOGE("Failed to subscribe msg type %x", msgType);
        close(fd);
        return NULL;
    }

    int role = MP_ROLE_CONSUMER;
    if (ioctl(fd, MEDIAPROXY_CONNECT, &role) < 0) {
        SUBTITLE_LOGE("Failed to connect mediaproxy");
        close(fd);
        return NULL;
    }

    int vdecId = ctx->param.playerId;
    SUBTITLE_LOGI("set vdec id:%d to media proxy", vdecId);
    if (ioctl(fd, MEDIAPROXY_SET_VDEC_ID, &vdecId) < 0) {
        SUBTITLE_LOGE("Failed to set vid %d", vdecId);
        if (ioctl(fd, MEDIAPROXY_DISCONNECT, &role) < 0) {
            SUBTITLE_LOGE("Failed to disconnect mediaproxy");
        }
        close(fd);
        return NULL;
    }

    int bufSize = 6 * 1024;
    void *bufPtr = malloc(bufSize);
    if (!bufPtr) {
        SUBTITLE_LOGE("no memory for reader, size:%d", bufSize);
        close(fd);
        return NULL;
    }

    struct aml_buffer_info buf;
    buf.message_type = MEDIA_VIDEO_CC_INFO;
    buf.data.buf_info.size = bufSize;
    buf.data.buf_info.ptr = (long long)bufPtr;

    struct CcSourceUserdataParam usrParam;
    struct CcSourceUserdataParam *pUsrParam = &usrParam;

    while (ctx->running) {
        struct pollfd pfd;
        int ret;

        pfd.fd = fd;
        pfd.events = POLLIN | POLLERR;

        ret = poll(&pfd, 1, USERDATA_POLL_TIMEOUT);

        if (!ctx->running || ret <= 0 || !(pfd.events & POLLIN)) continue;

        do {
            ret = ioctl(fd, MEDIAPROXY_GET_BUF_INFO, &buf);

            if (ret < 0) {
                SUBTITLE_LOGE("mp ioctl READ fail, err:%m");
                continue;
            }

            struct v4l_stream_common_s *v4lInfo =
                (struct v4l_stream_common_s *)buf.data.buf_info.ptr;

            SUBTITLE_LOGD("mediaproxy: ver:%u, inst:%u, type:%u, data_size:%u, sizeof(v4l_s):%zu",
                          v4lInfo->version, v4lInfo->instance_id, v4lInfo->meta_type,
                          v4lInfo->data_size, sizeof(struct v4l_stream_common_s));

            if (v4lInfo->meta_type != V4L_STREAM_CC_INFO || v4lInfo->data_size <= 0) {
                continue;
            }

            v4lInfoToUserdataParam(v4lInfo, pUsrParam);

            if (ctx->param.uDataCb) {
                if (isSubtitleDumpAvailable()) {
                    dumpCcData("mediaproxy data:", pUsrParam->pbuf_addr, pUsrParam->data_size);
                }
                ctx->param.uDataCb(ctx, pUsrParam);
            }
        } while (0 /*user_para_info.meta_info.records_in_que > 1*/);
    }

    if (ioctl(fd, MEDIAPROXY_DISCONNECT, &role) < 0) {
        SUBTITLE_LOGE("Failed to disconnect mediaproxy");
    }
    close(fd);
    free(bufPtr);
    return NULL;
}

#ifdef NEED_CCDATA_FROM_CCDATABASE
static void *ccDataserverUserdataTask(void *args) {
    enum {
        STAGE_INIT,
        STAGE_RB = STAGE_INIT,
        STAGE_VDECID,
        STAGE_SRC,
        STAGE_DAT,
    } stage = STAGE_INIT;

    CcSourceCtx *ctx = (CcSourceCtx *)args;
    SUBTITLE_LOGI("dataserver userdata task started.");

    int buf_size = 1024 * 10;
    uint8_t *buf = (uint8_t *)malloc(buf_size);
    if (!buf) {
        SUBTITLE_LOGE("memory fail");
        goto exit;
    }

    int ret = -1, last_ret = -100;
    while (ctx->running) {
        if (!ctx->das) {
            SUBTITLE_LOGE("dataserver not connected, try reconnect...");
            sleep(1);

            if (!ctx->running) goto exit;

            ctx->das = ctx->das_connect();
            continue;
        }

        // Prepare for the data
        uint8_t msg[sizeof(ds_msg_head_t) + 64];
        ds_msg_head_t *pmsg = (ds_msg_head_t *)msg;
        pmsg->magic = DS_MAGIC;
        pmsg->type = DS_T_CNTL;

        switch (stage) {
            case STAGE_RB:
                pmsg->size = snprintf((char *)pmsg->body, 64, ds_cmd_enable_ringbuffer, 1024 * 512);
                pmsg->size++;
                ret = dataserver_write_message(ctx->das, pmsg, sizeof(ds_msg_head_t) + pmsg->size);
                DAS_RET_CHK();
                break;
            case STAGE_VDECID:
                SUBTITLE_LOGI("dataserver set playerId:%d", ctx->param.playerId);
                pmsg->size = snprintf((char *)pmsg->body, 64,
                                      ds_cmd_set_source_para_userdata_vdec_id, ctx->param.playerId);
                pmsg->size++;
                ret = dataserver_write_message(ctx->das, pmsg, sizeof(ds_msg_head_t) + pmsg->size);
                DAS_RET_CHK();
                break;
            case STAGE_SRC:
                SUBTITLE_LOGI("dataserver set data source file:%s", ctx->devFile);
                pmsg->size = snprintf((char *)pmsg->body, 64, ds_cmd_set_source, ctx->devFile,
                                      ctx->param.owner);
                pmsg->size++;
                ret = dataserver_write_message(ctx->das, pmsg, sizeof(ds_msg_head_t) + pmsg->size);
                DAS_RET_CHK();
                break;
            default:
                // Stage is in init
                break;
        }

        if (!ctx->running) goto exit;

        ret = dataserver_read_message(ctx->das, buf, buf_size, 1000);

        if (ret == -2 || ret == -3) {
            if (ret != last_ret) {
                SUBTITLE_LOGI("%s", retToString(ret));
            }
            last_ret = ret;
            continue;
        }

        last_ret = ret;

        if (!ctx->running) goto exit;

        DAS_RET_CHK();

        pmsg = (ds_msg_head_t *)buf;
        switch (stage) {
            case STAGE_RB:
            case STAGE_VDECID:
            case STAGE_SRC:
                if (pmsg->type == DS_T_RESP) {
                    if (strncmp((char *)pmsg->body, "ok", 2) == 0) {
                        // Next stage
                        DAS_STAGE_CHK();
                        break;
                    }
                }

                // Stage fail, should exit and wait for next start.
                SUBTITLE_LOGE("fail in stage %d", stage);
                dataserver_disconnect(ctx->das);
                dataserver_destroy(ctx->das);
                ctx->das = NULL;
                stage = STAGE_INIT;
                goto exit;
            case STAGE_DAT:
            default:
                if (pmsg->type == DS_T_DATA) {
                    unsigned int left = pmsg->size;

                    ds_data_userdata_t *ud = (ds_data_userdata_t *)pmsg->body;
                    uint32_t version = ud->version;
                    // See SWPL-224528, PTS is changed from 32bit to 33bit from version 2.
                    // - Version 1 uses ds_data_userdata_t::meta_info::vpts as the PTS;
                    // - Version 2 uses ds_data_userdata_t::meta_info::vpts_64 as the PTS.
                    if (version != 2) {
                        SUBTITLE_LOGE(
                            "[subtitle_unsupported_flag] received unsupported version %d, "
                            "drop the current data.",
                            version);
                        continue;
                    }

                    while (left >= sizeof(ds_data_userdata_t)) {
                        if (strcmp(ctx->devFile, AMSTREAM_USDRDATA_DEV_FILE) == 0) {
                            if (isSubtitleDumpAvailable()) {
                                dumpCcData("dataserver amstream ioctl", ud->data, ud->data_size);
                            }
                            // Swap data every 8 bytes.
                            swapData(ud->data, ud->data_size);
                        }
                        if (ctx->param.uDataCb) {
                            if (isSubtitleDumpAvailable())
                                dumpCcData("dataserver userdata", ud->data, ud->data_size);
                            ctx->param.uDataCb(ctx, ud);
                        }

                        int udsize = sizeof(ds_data_userdata_t) + ud->data_size;
                        left -= udsize;
                        ud = (ds_data_userdata_t *)(pmsg->body + udsize);
                    }
                } else if (pmsg->type == DS_T_EVNT) {
                    SUBTITLE_LOGE("data overflow, restart data pipe");
                    dataserver_disconnect(ctx->das);
                    dataserver_destroy(ctx->das);
                    ctx->das = NULL;
                    stage = STAGE_INIT;
                    continue;
                }
                break;
        }
    }
exit:
    if (buf) {
        free(buf);
    }

    SUBTITLE_LOGI("dataserver userdata task exit.");
    return NULL;
}
#endif

#ifdef NEED_CCDATA_FROM_CCDATABASE
static void *ccDataServerVbiTask(void *args) {
    enum {
        STAGE_INIT,
        STAGE_PAR = STAGE_INIT,
        STAGE_SRC,
        STAGE_DAT,
    } stage = STAGE_INIT;

    CcSourceCtx *ctx = (CcSourceCtx *)args;

    SUBTITLE_LOGI("dataserver vbi task started.");

    int buf_size = sizeof(ds_data_vbi_t) * 50;

    uint8_t *buf = (uint8_t *)malloc(buf_size);

    int last = -1;
    AmlogicGetClock(&last);
    ctx->vCtx.last = last;
    if (!buf) {
        SUBTITLE_LOGE("memory fail");
        goto exit;
    }

    int ret = -1, last_ret = -100;
    while (ctx->running) {
        if (!ctx->das) {
            SUBTITLE_LOGE("dataserver not connected, try reconnect...");
            sleep(1);

            if (!ctx->running) goto exit;

            ctx->das = ctx->das_connect();
            continue;
        }

        // Prepare for the data
        uint8_t msg[sizeof(ds_msg_head_t) + 64];
        ds_msg_head_t *pmsg = (ds_msg_head_t *)msg;
        pmsg->magic = DS_MAGIC;
        pmsg->type = DS_T_CNTL;

        switch (stage) {
            case STAGE_PAR:
                pmsg->size =
                    snprintf((char *)pmsg->body, 64, ds_cmd_set_source_para_vbi_type, 1 /*USCC*/);
                pmsg->size++;
                ret = dataserver_write_message(ctx->das, (uint8_t *)pmsg,
                                               sizeof(ds_msg_head_t) + pmsg->size);
                DAS_RET_CHK();
                break;
            case STAGE_SRC:
                pmsg->size = snprintf((char *)pmsg->body, 64, ds_cmd_set_source, CC_VBI_DEV_FILE,
                                      ctx->param.owner);
                pmsg->size++;
                ret = dataserver_write_message(ctx->das, pmsg, sizeof(ds_msg_head_t) + pmsg->size);
                DAS_RET_CHK();
                break;
            default:
                // Stage is in init
                break;
        }

        if (!ctx->running) goto exit;

        ret = dataserver_read_message(ctx->das, buf, buf_size, 1000);
        if (ret == -2 || ret == -3) {
            if (ret != last_ret) {
                SUBTITLE_LOGI("%s", retToString(ret));
            }
            last_ret = ret;
            continue;
        }

        last_ret = ret;

        if (!ctx->running) goto exit;

        DAS_RET_CHK();

        pmsg = (ds_msg_head_t *)buf;
        switch (stage) {
            case STAGE_SRC:
            case STAGE_PAR:
                if (pmsg->type == DS_T_RESP) {
                    if (strncmp((char *)pmsg->body, "ok", 2) == 0) {
                        // Next stage
                        stage = (stage == STAGE_PAR) ? STAGE_SRC : STAGE_DAT;
                        break;
                    }
                }

                // Stage fail, should exit and wait for next start
                SUBTITLE_LOGE("fail in stage %d", stage);
                dataserver_disconnect(ctx->das);
                dataserver_destroy(ctx->das);
                ctx->das = NULL;
                stage = STAGE_INIT;
                goto exit;
            case STAGE_DAT:
            default:
                if (pmsg->type == DS_T_DATA) {
                    ds_data_vbi_t *vbi = (ds_data_vbi_t *)pmsg->body;
                    unsigned int left = pmsg->size;

                    while (left >= sizeof(ds_data_vbi_t)) {
                        if (ctx->param.vDataCb) {
                            SUBTITLE_LOGD("dataserver vbi line %d: %02x %02x", vbi->line_num,
                                          vbi->b[0], vbi->b[1]);
                            ctx->param.vDataCb(ctx, vbi);
                        }
                        vbi++;
                        left -= sizeof(ds_data_vbi_t);
                    }
                    if (ctx->param.vEvtHandler) {
                        ctx->param.vEvtHandler(ctx);
                    }

                } else if (pmsg->type == DS_T_EVNT) {
                    SUBTITLE_LOGE("data overflow, restart data pipe");
                    dataserver_disconnect(ctx->das);
                    dataserver_destroy(ctx->das);
                    ctx->das = NULL;
                    stage = STAGE_INIT;
                    continue;
                }
                break;
        }
    }

exit:
    if (buf) {
        free(buf);
    }
    SUBTITLE_LOGI("vbi task exit.");
    return NULL;
}
#endif

CcSourceMgrHnd ccSourceManagerCreate() {
    SUBTITLE_LOGI_FUNC_IN();
    CcSourceCtx *ctx;
    ctx = (CcSourceCtx *)calloc(1, sizeof(CcSourceCtx));
    if (!ctx) {
        SUBTITLE_LOGE("memory fail.");
        return NULL;
    }
    pthread_mutex_init(&ctx->lock, NULL);
    SUBTITLE_LOGI("end");
    return ctx;
}

CcSourceStatus ccSourceManagerStop(CcSourceMgrHnd handle) {
    SUBTITLE_LOGI_FUNC_IN();
    CcSourceCtx *ctx = (CcSourceCtx *)handle;
    if (ctx == NULL) {
        SUBTITLE_LOGE("source context is null, return!");
        return CC_SOURCE_INVALID;
    }

    bool join = false;
    pthread_mutex_lock(&ctx->lock);
    if (ctx->running) {
        ctx->running = false;
        join = true;
    }
    pthread_mutex_unlock(&ctx->lock);

#ifdef NEED_CCDATA_FROM_CCDATABASE
    // Disconnect includes waking fmq message block.
    if (ctx->das) {
        dataserver_disconnect(ctx->das);
    }
#endif

    if (join) {
        pthread_join(ctx->thread, NULL);
    }

#ifdef NEED_CCDATA_FROM_CCDATABASE
    if (ctx->das) {
        dataserver_destroy(ctx->das);
        ctx->das = NULL;
    }
#endif
    SUBTITLE_LOGI("end");
    return CC_SOURCE_OK;
}

CcSourceStatus ccSourceManagerDestroy(CcSourceMgrHnd handle) {
    SUBTITLE_LOGI_FUNC_IN();
    CcSourceCtx *ctx = (CcSourceCtx *)handle;
    if (ctx == NULL) {
        SUBTITLE_LOGE("source context is null, return!");
        return CC_SOURCE_INVALID;
    }
    pthread_mutex_destroy(&ctx->lock);
    free(ctx);
    ctx = NULL;
    SUBTITLE_LOGI("end");
    return CC_SOURCE_OK;
}

CcSourceStatus ccSourceManagerStart(CcSourceMgrHnd handle, CcSourceStartParam *param) {
    SUBTITLE_LOGI_FUNC_IN();
    CcSourceCtx *ctx = (CcSourceCtx *)handle;
    if (ctx == NULL) {
        SUBTITLE_LOGE("ctx is null, return!");
        return CC_SOURCE_INVALID;
    }

    pthread_mutex_lock(&ctx->lock);
    ctx->running = true;
    pthread_mutex_unlock(&ctx->lock);

    ctx->param = *param;
    SUBTITLE_LOGI("srcType:%d", ctx->param.srcType);

    // CC Digital Data Source strategy:
    // Android/linux:
    //   ccdataserver exists:
    //      1.if vdec is non-v4l, then read data from AMSTREAM_USDRDATA_DEV_FILE
    //        by ccdataserver
    //      2.if vdec is v4l, then read data from MEDIA_PROXY_DEV_FILE by ccdataserver
    //   ccdataserver doesn't exist:
    //      1.if vdec is non-v4l, then read data from AMSTREAM_USDRDATA_DEV_FILE
    //      2.if vdec is v4l, then read data from MEDIA_PROXY_DEV_FILE
    //
    // CC Analog Data Source strategy
    // Android/linux:
    //   ccdataserver exists:
    //      1. get Line 21 data from /dev/vbi by ccdataserver
    //   ccdataserver doesn't exist:
    //      1. get Line 21 data from /dev/vbi directly
    //
    // ccdataserver enable/disable by product config:
    //      if need ratings(xds) to enable ccdataserver otherwise disable.
    switch (ctx->param.srcType) {
        case CC_SOURCE_TYPE_AMSTREAM_DEVICE:
            pthread_create(&ctx->thread, NULL, ccUserdataDeviceTask, ctx);
            break;
        case CC_SOURCE_TYPE_VBI_DEVICE:
            pthread_create(&ctx->thread, NULL, ccVbiDeviceTask, ctx);
            break;
        case CC_SOURCE_TYPE_MEDIA_PROXY:
            pthread_create(&ctx->thread, NULL, ccMediaProxyDeviceTask, ctx);
            break;
        case CC_SOURCE_TYPE_CCDATASERVER_MEDIA_PROXY:
#ifdef NEED_CCDATA_FROM_CCDATABASE
            SUBTITLE_LOGI("dataserver_connect");
            ctx->das_connect = (ds_conn_t * (*)()) dataserver_connect;
            ctx->das = ctx->das_connect();
            snprintf(ctx->devFile, sizeof(ctx->devFile), "%s", MEDIA_PROXY_DEV_FILE);
            pthread_create(&ctx->thread, NULL, ccDataserverUserdataTask, ctx);
#endif
            break;
        case CC_SOURCE_TYPE_CCDATASERVER_AMSTREAM_DEVICE:
#ifdef NEED_CCDATA_FROM_CCDATABASE
            SUBTITLE_LOGI("dataserver_connect");
            ctx->das_connect = (ds_conn_t * (*)()) dataserver_connect;
            ctx->das = ctx->das_connect();
            snprintf(ctx->devFile, sizeof(ctx->devFile), "%s", AMSTREAM_USDRDATA_DEV_FILE);
            pthread_create(&ctx->thread, NULL, ccDataserverUserdataTask, ctx);
#endif
            break;
        case CC_SOURCE_TYPE_CCDATASERVER_VBI_DEVICE:
#ifdef NEED_CCDATA_FROM_CCDATABASE
            SUBTITLE_LOGI("dataserver_connect");
            ctx->das_connect = (ds_conn_t * (*)()) dataserver_connect;
            ctx->das = ctx->das_connect();
            pthread_create(&ctx->thread, NULL, ccDataServerVbiTask, ctx);
#endif
            break;
        default:
            pthread_create(&ctx->thread, NULL, ccUserdataDeviceTask, ctx);
            break;
    }
    SUBTITLE_LOGI("end");
    return CC_SOURCE_OK;
}