Merge changes from topic "play"

am: 297193ceed

Change-Id: I022154516e912fd25163009b3f5d2f4d1d1c8639

tv/tuner/1.0/IDemux.hal

@@ -180,5 +180,220 @@ interface IDemux {
      *         UNKNOWN_ERROR if failed for other reasons.
      */
     close() generates (Result result);
-};
 
+    /**
+     * Add output to the demux
+     *
+     * It is used by the client to record output data from selected filters.
+     *
+     * @param bufferSize the buffer size of the output to be added. It's used to
+     * create a FMQ(Fast Message Queue) to hold data from selected filters.
+     * @param cb the callback for the demux to be used to send notifications
+     * back to the client.
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         OUT_OF_MEMORY if failed for not enough memory.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    addOutput(uint32_t bufferSize, IDemuxCallback cb) generates (Result result);
+
+    /**
+     * Get the descriptor of the output's FMQ
+     *
+     * It is used by the client to get the descriptor of the output's Fast
+     * Message Queue. The data in FMQ is muxed packets output from selected
+     * filters. The packet's format is specifed by DemuxDataFormat in
+     * DemuxOutputSettings.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         UNKNOWN_ERROR if failed for other reasons.
+     * @return queue the descriptor of the output's FMQ
+     */
+    getOutputQueueDesc() generates (Result result, fmq_sync<uint8_t> queue);
+
+    /**
+     * Configure the demux's output.
+     *
+     * It is used by the client to configure the demux's output for recording.
+     *
+     * @param settings the settings of the demux's output.
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    configureOutput(DemuxOutputSettings settings) generates (Result result);
+
+    /**
+     * Attach one filter to the demux's output.
+     *
+     * It is used by the client to mux one filter's output to demux's output.
+     *
+     * @param filterId the ID of the attached filter.
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    attachOutputTsFilter(DemuxFilterId filterId) generates (Result result);
+
+    /**
+     * Detach one filter from the demux's output.
+     *
+     * It is used by the client to remove one filter's output from demux's
+     * output.
+     *
+     * @param filterId the ID of the detached filter.
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    detachOutputTsFilter(DemuxFilterId filterId) generates (Result result);
+
+    /**
+     * Start to take data to the demux's output.
+     *
+     * It is used by the client to ask the output to start to take data from
+     * attached filters.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    startOutput() generates (Result result);
+
+    /**
+     * Stop to take data to the demux's output.
+     *
+     * It is used by the client to ask the output to stop to take data from
+     * attached filters.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    stopOutput() generates (Result result);
+
+    /**
+     * Flush unconsumed data in the demux's output.
+     *
+     * It is used by the client to ask the demux to flush the data which is
+     * already produced but not consumed yet in the demux's output.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    flushOutput() generates (Result result);
+
+    /**
+     * Remove the demux's output.
+     *
+     * It is used by the client to remove the demux's output.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    removeOutput() generates (Result result);
+
+    /**
+     * Add input to the demux
+     *
+     * It is used by the client to add the demux's input for playback content.
+     *
+     * @param bufferSize the buffer size of the demux's input to be added.
+     * It's used to create a FMQ(Fast Message Queue) to hold input data.
+     * @param cb the callback for the demux to be used to send notifications
+     * back to the client.
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         OUT_OF_MEMORY if failed for not enough memory.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    addInput(uint32_t bufferSize, IDemuxCallback cb) generates (Result result);
+
+    /**
+     * Get the descriptor of the input's FMQ
+     *
+     * It is used by the client to get the descriptor of the input's Fast
+     * Message Queue. The data in FMQ is fed by client. Data format is specifed
+     * by DemuxDataFormat in DemuxInputSettings.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         UNKNOWN_ERROR if failed for other reasons.
+     * @return queue the descriptor of the output's FMQ
+     */
+    getInputQueueDesc() generates (Result result, fmq_sync<uint8_t> queue);
+
+    /**
+     * Configure the demux's input.
+     *
+     * It is used by the client to configure the demux's input for playback.
+     *
+     * @param settings the settings of the demux's input.
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    configureInput(DemuxInputSettings settings) generates (Result result);
+
+    /**
+     * Start to consume the data from the demux's input.
+     *
+     * It is used by the client to ask the demux to start to consume data from
+     * the demux's input.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    startInput() generates (Result result);
+
+    /**
+     * Stop to consume the data from the demux's input.
+     *
+     * It is used by the client to ask the demux to stop to consume data from
+     * the demux's input.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    stopInput() generates (Result result);
+
+    /**
+     * Flush unconsumed data in the demux's input.
+     *
+     * It is used by the client to ask the demux to flush the data which is
+     * already produced but not consumed yet in the demux's input.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    flushInput() generates (Result result);
+
+    /**
+     * Remove the demux's input.
+     *
+     * It is used by the client to remove the demux's input.
+     *
+     * @return result Result status of the operation.
+     *         SUCCESS if successful,
+     *         INVALID_STATE if failed for wrong state.
+     *         UNKNOWN_ERROR if failed for other reasons.
+     */
+    removeInput() generates (Result result);
+};

tv/tuner/1.0/IDemuxCallback.hal

@@ -15,5 +15,19 @@ interface IDemuxCallback {
      * @param status a new status of the demux filter.
      */
     oneway onFilterStatus(DemuxFilterId filterId, DemuxFilterStatus status);
+
+    /**
+     * Notify the client a new status of the demux's output.
+     *
+     * @param status a new status of the demux's output.
+     */
+    oneway onOutputStatus(DemuxOutputStatus status);
+
+    /**
+     * Notify the client a new status of the demux's input.
+     *
+     * @param status a new status of the demux's input.
+     */
+    oneway onInputStatus(DemuxInputStatus status);
 };
 

tv/tuner/1.0/default/Demux.cpp

@@ -73,34 +73,6 @@ Demux::Demux(uint32_t demuxId) {
 
 Demux::~Demux() {}
 
-bool Demux::createAndSaveMQ(uint32_t bufferSize, uint32_t filterId) {
-    ALOGV("%s", __FUNCTION__);
-
-    // Create a synchronized FMQ that supports blocking read/write
-    std::unique_ptr<FilterMQ> tmpFilterMQ =
-            std::unique_ptr<FilterMQ>(new (std::nothrow) FilterMQ(bufferSize, true));
-    if (!tmpFilterMQ->isValid()) {
-        ALOGW("Failed to create FMQ of filter with id: %d", filterId);
-        return false;
-    }
-
-    mFilterMQs.resize(filterId + 1);
-    mFilterMQs[filterId] = std::move(tmpFilterMQ);
-
-    EventFlag* mFilterEventFlag;
-    if (EventFlag::createEventFlag(mFilterMQs[filterId]->getEventFlagWord(), &mFilterEventFlag) !=
-        OK) {
-        return false;
-    }
-    mFilterEventFlags.resize(filterId + 1);
-    mFilterEventFlags[filterId] = mFilterEventFlag;
-    mFilterWriteCount.resize(filterId + 1);
-    mFilterWriteCount[filterId] = 0;
-    mThreadRunning.resize(filterId + 1);
-
-    return true;
-}
-
 Return<Result> Demux::setFrontendDataSource(uint32_t frontendId) {
     ALOGV("%s", __FUNCTION__);
 
@@ -113,23 +85,42 @@ Return<void> Demux::addFilter(DemuxFilterType type, uint32_t bufferSize,
                               const sp<IDemuxCallback>& cb, addFilter_cb _hidl_cb) {
     ALOGV("%s", __FUNCTION__);
 
-    uint32_t filterId = mLastUsedFilterId + 1;
+    uint32_t filterId;
-    mLastUsedFilterId += 1;
+
+    if (!mUnusedFilterIds.empty()) {
+        filterId = *mUnusedFilterIds.begin();
+
+        mUnusedFilterIds.erase(filterId);
+    } else {
+        filterId = ++mLastUsedFilterId;
+
+        mDemuxCallbacks.resize(filterId + 1);
+        mFilterMQs.resize(filterId + 1);
+        mFilterEvents.resize(filterId + 1);
+        mFilterEventFlags.resize(filterId + 1);
+        mFilterThreadRunning.resize(filterId + 1);
+        mFilterThreads.resize(filterId + 1);
+    }
+
+    mUsedFilterIds.insert(filterId);
 
     if ((type != DemuxFilterType::PCR || type != DemuxFilterType::TS) && cb == nullptr) {
         ALOGW("callback can't be null");
         _hidl_cb(Result::INVALID_ARGUMENT, filterId);
         return Void();
     }
+
     // Add callback
-    mDemuxCallbacks.resize(filterId + 1);
     mDemuxCallbacks[filterId] = cb;
 
-    // Mapping from the filter ID to the filter type
+    // Mapping from the filter ID to the filter event
-    mFilterTypes.resize(filterId + 1);
+    DemuxFilterEvent event{
-    mFilterTypes[filterId] = type;
+            .filterId = filterId,
+            .filterType = type,
+    };
+    mFilterEvents[filterId] = event;
 
-    if (!createAndSaveMQ(bufferSize, filterId)) {
+    if (!createFilterMQ(bufferSize, filterId)) {
         _hidl_cb(Result::UNKNOWN_ERROR, -1);
         return Void();
     }
@@ -141,8 +132,8 @@ Return<void> Demux::addFilter(DemuxFilterType type, uint32_t bufferSize,
 Return<void> Demux::getFilterQueueDesc(uint32_t filterId, getFilterQueueDesc_cb _hidl_cb) {
     ALOGV("%s", __FUNCTION__);
 
-    if (filterId < 0 || filterId > mLastUsedFilterId) {
+    if (mUsedFilterIds.find(filterId) == mUsedFilterIds.end()) {
-        ALOGW("No filter with id: %d exists", filterId);
+        ALOGW("No filter with id: %d exists to get desc", filterId);
         _hidl_cb(Result::INVALID_ARGUMENT, FilterMQ::Descriptor());
         return Void();
     }
@@ -160,35 +151,29 @@ Return<Result> Demux::configureFilter(uint32_t /* filterId */,
 
 Return<Result> Demux::startFilter(uint32_t filterId) {
     ALOGV("%s", __FUNCTION__);
+    Result result;
 
-    if (filterId < 0 || filterId > mLastUsedFilterId) {
+    if (mUsedFilterIds.find(filterId) == mUsedFilterIds.end()) {
-        ALOGW("No filter with id: %d exists", filterId);
+        ALOGW("No filter with id: %d exists to start filter", filterId);
         return Result::INVALID_ARGUMENT;
     }
 
-    DemuxFilterType filterType = mFilterTypes[filterId];
+    switch (mFilterEvents[filterId].filterType) {
-    Result result;
-    DemuxFilterEvent event{
-            .filterId = filterId,
-            .filterType = filterType,
-    };
-
-    switch (filterType) {
         case DemuxFilterType::SECTION:
-            result = startSectionFilterHandler(event);
+            result = startFilterLoop(filterId);
             break;
         case DemuxFilterType::PES:
-            result = startPesFilterHandler(event);
+            result = startPesFilterHandler(filterId);
             break;
         case DemuxFilterType::TS:
             result = startTsFilterHandler();
             return Result::SUCCESS;
         case DemuxFilterType::AUDIO:
         case DemuxFilterType::VIDEO:
-            result = startMediaFilterHandler(event);
+            result = startMediaFilterHandler(filterId);
             break;
         case DemuxFilterType::RECORD:
-            result = startRecordFilterHandler(event);
+            result = startRecordFilterHandler(filterId);
             break;
         case DemuxFilterType::PCR:
             result = startPcrFilterHandler();
@@ -212,9 +197,13 @@ Return<Result> Demux::flushFilter(uint32_t /* filterId */) {
     return Result::SUCCESS;
 }
 
-Return<Result> Demux::removeFilter(uint32_t /* filterId */) {
+Return<Result> Demux::removeFilter(uint32_t filterId) {
     ALOGV("%s", __FUNCTION__);
 
+    // resetFilterRecords(filterId);
+    mUsedFilterIds.erase(filterId);
+    mUnusedFilterIds.insert(filterId);
+
     return Result::SUCCESS;
 }
 
@@ -239,25 +228,291 @@ Return<void> Demux::getAvSyncTime(AvSyncHwId /* avSyncHwId */, getAvSyncTime_cb
 Return<Result> Demux::close() {
     ALOGV("%s", __FUNCTION__);
 
+    set<uint32_t>::iterator it;
+    mInputThread = 0;
+    mOutputThread = 0;
+    mFilterThreads.clear();
+    mUnusedFilterIds.clear();
+    mUsedFilterIds.clear();
+    mDemuxCallbacks.clear();
+    mFilterMQs.clear();
+    mFilterEvents.clear();
+    mFilterEventFlags.clear();
+    mLastUsedFilterId = -1;
+
     return Result::SUCCESS;
 }
 
-bool Demux::writeSectionsAndCreateEvent(DemuxFilterEvent& event, uint32_t sectionNum) {
+Return<Result> Demux::addOutput(uint32_t bufferSize, const sp<IDemuxCallback>& cb) {
-    event.events.resize(sectionNum);
+    ALOGV("%s", __FUNCTION__);
-    for (int i = 0; i < sectionNum; i++) {
+
-        DemuxFilterSectionEvent secEvent;
+    // Create a synchronized FMQ that supports blocking read/write
-        secEvent = {
+    std::unique_ptr<FilterMQ> tmpFilterMQ =
-                // temp dump meta data
+            std::unique_ptr<FilterMQ>(new (std::nothrow) FilterMQ(bufferSize, true));
-                .tableId = 0,
+    if (!tmpFilterMQ->isValid()) {
-                .version = 1,
+        ALOGW("Failed to create output FMQ");
-                .sectionNum = 1,
+        return Result::UNKNOWN_ERROR;
-                .dataLength = 530,
-        };
-        event.events[i].section(secEvent);
-        if (!writeDataToFilterMQ(fakeDataInputBuffer, event.filterId)) {
-            return false;
-        }
     }
+
+    mOutputMQ = std::move(tmpFilterMQ);
+
+    if (EventFlag::createEventFlag(mOutputMQ->getEventFlagWord(), &mOutputEventFlag) != OK) {
+        return Result::UNKNOWN_ERROR;
+    }
+
+    mOutputCallback = cb;
+
+    return Result::SUCCESS;
+}
+
+Return<void> Demux::getOutputQueueDesc(getOutputQueueDesc_cb _hidl_cb) {
+    ALOGV("%s", __FUNCTION__);
+
+    if (!mOutputMQ) {
+        _hidl_cb(Result::NOT_INITIALIZED, FilterMQ::Descriptor());
+        return Void();
+    }
+
+    _hidl_cb(Result::SUCCESS, *mOutputMQ->getDesc());
+    return Void();
+}
+
+Return<Result> Demux::configureOutput(const DemuxOutputSettings& /* settings */) {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::attachOutputTsFilter(uint32_t /*filterId*/) {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::detachOutputTsFilter(uint32_t /* filterId */) {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::startOutput() {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::stopOutput() {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::flushOutput() {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::removeOutput() {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::addInput(uint32_t bufferSize, const sp<IDemuxCallback>& cb) {
+    ALOGV("%s", __FUNCTION__);
+
+    // Create a synchronized FMQ that supports blocking read/write
+    std::unique_ptr<FilterMQ> tmpInputMQ =
+            std::unique_ptr<FilterMQ>(new (std::nothrow) FilterMQ(bufferSize, true));
+    if (!tmpInputMQ->isValid()) {
+        ALOGW("Failed to create input FMQ");
+        return Result::UNKNOWN_ERROR;
+    }
+
+    mInputMQ = std::move(tmpInputMQ);
+
+    if (EventFlag::createEventFlag(mInputMQ->getEventFlagWord(), &mInputEventFlag) != OK) {
+        return Result::UNKNOWN_ERROR;
+    }
+
+    mInputCallback = cb;
+
+    return Result::SUCCESS;
+}
+
+Return<void> Demux::getInputQueueDesc(getInputQueueDesc_cb _hidl_cb) {
+    ALOGV("%s", __FUNCTION__);
+
+    if (!mInputMQ) {
+        _hidl_cb(Result::NOT_INITIALIZED, FilterMQ::Descriptor());
+        return Void();
+    }
+
+    _hidl_cb(Result::SUCCESS, *mInputMQ->getDesc());
+    return Void();
+}
+
+Return<Result> Demux::configureInput(const DemuxInputSettings& /* settings */) {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::startInput() {
+    ALOGV("%s", __FUNCTION__);
+
+    pthread_create(&mInputThread, NULL, __threadLoopInput, this);
+    pthread_setname_np(mInputThread, "demux_input_waiting_loop");
+
+    // TODO start another thread to send filter status callback to the framework
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::stopInput() {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::flushInput() {
+    ALOGV("%s", __FUNCTION__);
+
+    return Result::SUCCESS;
+}
+
+Return<Result> Demux::removeInput() {
+    ALOGV("%s", __FUNCTION__);
+
+    mInputMQ = nullptr;
+
+    return Result::SUCCESS;
+}
+
+Result Demux::startFilterLoop(uint32_t filterId) {
+    struct ThreadArgs* threadArgs = (struct ThreadArgs*)malloc(sizeof(struct ThreadArgs));
+    threadArgs->user = this;
+    threadArgs->filterId = filterId;
+
+    pthread_t mFilterThread;
+    pthread_create(&mFilterThread, NULL, __threadLoopFilter, (void*)threadArgs);
+    mFilterThreads[filterId] = mFilterThread;
+    pthread_setname_np(mFilterThread, "demux_filter_waiting_loop");
+
+    return Result::SUCCESS;
+}
+
+Result Demux::startSectionFilterHandler(uint32_t filterId, vector<uint8_t> data) {
+    if (!writeSectionsAndCreateEvent(filterId, data)) {
+        ALOGD("[Demux] filter %d fails to write into FMQ. Ending thread", filterId);
+        return Result::UNKNOWN_ERROR;
+    }
+
+    return Result::SUCCESS;
+}
+
+Result Demux::startPesFilterHandler(uint32_t filterId) {
+    // TODO generate multiple events in one event callback
+    DemuxFilterPesEvent pesEvent;
+    pesEvent = {
+            // temp dump meta data
+            .streamId = 0,
+            .dataLength = 530,
+    };
+    mFilterEvents[filterId].events.resize(1);
+    mFilterEvents[filterId].events[0].pes(pesEvent);
+    /*pthread_create(&mThreadId, NULL, __threadLoop, this);
+    pthread_setname_np(mThreadId, "demux_section_filter_waiting_loop");*/
+    if (!writeDataToFilterMQ(fakeDataInputBuffer, filterId)) {
+        return Result::INVALID_STATE;
+    }
+
+    if (mDemuxCallbacks[filterId] == nullptr) {
+        return Result::NOT_INITIALIZED;
+    }
+
+    mDemuxCallbacks[filterId]->onFilterEvent(mFilterEvents[filterId]);
+    return Result::SUCCESS;
+}
+
+Result Demux::startTsFilterHandler() {
+    // TODO handle starting TS filter
+    return Result::SUCCESS;
+}
+
+Result Demux::startMediaFilterHandler(uint32_t filterId) {
+    DemuxFilterMediaEvent mediaEvent;
+    mediaEvent = {
+            // temp dump meta data
+            .pts = 0,
+            .dataLength = 530,
+            .secureMemory = nullptr,
+    };
+    mFilterEvents[filterId].events.resize(1);
+    mFilterEvents[filterId].events[0].media() = mediaEvent;
+    // TODO handle write FQM for media stream
+    return Result::SUCCESS;
+}
+
+Result Demux::startRecordFilterHandler(uint32_t filterId) {
+    DemuxFilterRecordEvent recordEvent;
+    recordEvent = {
+            // temp dump meta data
+            .tpid = 0,
+            .packetNum = 0,
+    };
+    recordEvent.indexMask.tsIndexMask() = 0x01;
+    mFilterEvents[filterId].events.resize(1);
+    mFilterEvents[filterId].events[0].ts() = recordEvent;
+    return Result::SUCCESS;
+}
+
+Result Demux::startPcrFilterHandler() {
+    // TODO handle starting PCR filter
+    return Result::SUCCESS;
+}
+
+bool Demux::createFilterMQ(uint32_t bufferSize, uint32_t filterId) {
+    ALOGV("%s", __FUNCTION__);
+
+    // Create a synchronized FMQ that supports blocking read/write
+    std::unique_ptr<FilterMQ> tmpFilterMQ =
+            std::unique_ptr<FilterMQ>(new (std::nothrow) FilterMQ(bufferSize, true));
+    if (!tmpFilterMQ->isValid()) {
+        ALOGW("Failed to create FMQ of filter with id: %d", filterId);
+        return false;
+    }
+
+    mFilterMQs[filterId] = std::move(tmpFilterMQ);
+
+    EventFlag* filterEventFlag;
+    if (EventFlag::createEventFlag(mFilterMQs[filterId]->getEventFlagWord(), &filterEventFlag) !=
+        OK) {
+        return false;
+    }
+    mFilterEventFlags[filterId] = filterEventFlag;
+
+    return true;
+}
+
+bool Demux::writeSectionsAndCreateEvent(uint32_t filterId, vector<uint8_t> data) {
+    // TODO check how many sections has been read
+    std::lock_guard<std::mutex> lock(mFilterEventLock);
+    int size = mFilterEvents[filterId].events.size();
+    mFilterEvents[filterId].events.resize(size + 1);
+    if (!writeDataToFilterMQ(data, filterId)) {
+        return false;
+    }
+    DemuxFilterSectionEvent secEvent;
+    secEvent = {
+            // temp dump meta data
+            .tableId = 0,
+            .version = 1,
+            .sectionNum = 1,
+            .dataLength = 530,
+    };
+    mFilterEvents[filterId].events[size].section(secEvent);
     return true;
 }
 
@@ -269,116 +524,82 @@ bool Demux::writeDataToFilterMQ(const std::vector<uint8_t>& data, uint32_t filte
     return false;
 }
 
-Result Demux::startSectionFilterHandler(DemuxFilterEvent event) {
+bool Demux::filterAndOutputData() {
-    struct ThreadArgs* threadArgs = (struct ThreadArgs*)malloc(sizeof(struct ThreadArgs));
+    ALOGD("[Demux] start to dispatch data to filters");
-    threadArgs->user = this;
+    // Read input data from the input FMQ
-    threadArgs->event = &event;
+    int size = mInputMQ->availableToRead();
+    vector<uint8_t> dataOutputBuffer;
+    dataOutputBuffer.resize(size);
+    mInputMQ->read(dataOutputBuffer.data(), size);
 
-    pthread_create(&mThreadId, NULL, __threadLoop, (void*)threadArgs);
+    Result result;
-    pthread_setname_np(mThreadId, "demux_filter_waiting_loop");
+    // Filter the data and feed the output to each filter
-
+    set<uint32_t>::iterator it;
-    return Result::SUCCESS;
+    for (it = mUsedFilterIds.begin(); it != mUsedFilterIds.end(); it++) {
-}
+        switch (mFilterEvents[*it].filterType) {
-
+            case DemuxFilterType::SECTION:
-Result Demux::startPesFilterHandler(DemuxFilterEvent& event) {
+                result = startSectionFilterHandler(*it, dataOutputBuffer);
-    // TODO generate multiple events in one event callback
+                break;
-    DemuxFilterPesEvent pesEvent;
+            case DemuxFilterType::PES:
-    pesEvent = {
+                result = startPesFilterHandler(*it);
-            // temp dump meta data
+                break;
-            .streamId = 0,
+            case DemuxFilterType::TS:
-            .dataLength = 530,
+                result = startTsFilterHandler();
-    };
+                break;
-    event.events.resize(1);
+            case DemuxFilterType::AUDIO:
-    event.events[0].pes(pesEvent);
+            case DemuxFilterType::VIDEO:
-    /*pthread_create(&mThreadId, NULL, __threadLoop, this);
+                result = startMediaFilterHandler(*it);
-    pthread_setname_np(mThreadId, "demux_section_filter_waiting_loop");*/
+                break;
-    if (!writeDataToFilterMQ(fakeDataInputBuffer, event.filterId)) {
+            case DemuxFilterType::RECORD:
-        return Result::INVALID_STATE;
+                result = startRecordFilterHandler(*it);
+                break;
+            case DemuxFilterType::PCR:
+                result = startPcrFilterHandler();
+                break;
+            default:
+                return false;
+        }
     }
 
-    if (mDemuxCallbacks[event.filterId] == nullptr) {
+    return result == Result::SUCCESS;
-        return Result::NOT_INITIALIZED;
-    }
-
-    mDemuxCallbacks[event.filterId]->onFilterEvent(event);
-    return Result::SUCCESS;
 }
 
-Result Demux::startTsFilterHandler() {
+void* Demux::__threadLoopFilter(void* threadArg) {
-    // TODO handle starting TS filter
-    return Result::SUCCESS;
-}
-
-Result Demux::startMediaFilterHandler(DemuxFilterEvent& event) {
-    DemuxFilterMediaEvent mediaEvent;
-    mediaEvent = {
-            // temp dump meta data
-            .pts = 0,
-            .dataLength = 530,
-            .secureMemory = nullptr,
-    };
-    event.events.resize(1);
-    event.events[0].media() = mediaEvent;
-    // TODO handle write FQM for media stream
-    return Result::SUCCESS;
-}
-
-Result Demux::startRecordFilterHandler(DemuxFilterEvent& event) {
-    DemuxFilterRecordEvent recordEvent;
-    recordEvent = {
-            // temp dump meta data
-            .tpid = 0,
-            .packetNum = 0,
-    };
-    recordEvent.indexMask.tsIndexMask() = 0x01;
-    event.events.resize(1);
-    event.events[0].ts() = recordEvent;
-    return Result::SUCCESS;
-}
-
-Result Demux::startPcrFilterHandler() {
-    // TODO handle starting PCR filter
-    return Result::SUCCESS;
-}
-
-void* Demux::__threadLoop(void* threadArg) {
     Demux* const self = static_cast<Demux*>(((struct ThreadArgs*)threadArg)->user);
-    self->filterThreadLoop(((struct ThreadArgs*)threadArg)->event);
+    self->filterThreadLoop(((struct ThreadArgs*)threadArg)->filterId);
     return 0;
 }
 
-void Demux::filterThreadLoop(DemuxFilterEvent* event) {
+void* Demux::__threadLoopInput(void* user) {
-    uint32_t filterId = event->filterId;
+    Demux* const self = static_cast<Demux*>(user);
-    ALOGD("[Demux] filter %d threadLoop start.", filterId);
+    self->inputThreadLoop();
-    mThreadRunning[filterId] = true;
+    return 0;
+}
 
-    while (mThreadRunning[filterId]) {
+void Demux::filterThreadLoop(uint32_t filterId) {
+    ALOGD("[Demux] filter %d threadLoop start.", filterId);
+    mFilterThreadRunning[filterId] = true;
+
+    // For the first time of filter output, implementation needs to send the filter
+    // Event Callback without waiting for the DATA_CONSUMED to init the process.
+    while (mFilterThreadRunning[filterId]) {
+        if (mFilterEvents[filterId].events.size() == 0) {
+            ALOGD("[Demux] wait for filter data output.");
+            usleep(1000 * 1000);
+            continue;
+        }
+        // After successfully write, send a callback and wait for the read to be done
+        mDemuxCallbacks[filterId]->onFilterEvent(mFilterEvents[filterId]);
+        mFilterEvents[filterId].events.resize(0);
+        break;
+    }
+
+    while (mFilterThreadRunning[filterId]) {
         uint32_t efState = 0;
         // We do not wait for the last round of writen data to be read to finish the thread
         // because the VTS can verify the reading itself.
         for (int i = 0; i < SECTION_WRITE_COUNT; i++) {
-            DemuxFilterEvent filterEvent{
+            while (mFilterThreadRunning[filterId]) {
-                    .filterId = filterId,
-                    .filterType = event->filterType,
-            };
-            if (!writeSectionsAndCreateEvent(filterEvent, 2)) {
-                ALOGD("[Demux] filter %d fails to write into FMQ. Ending thread", filterId);
-                break;
-            }
-            mFilterWriteCount[filterId]++;
-            if (mDemuxCallbacks[filterId] == nullptr) {
-                ALOGD("[Demux] filter %d does not hava callback. Ending thread", filterId);
-                break;
-            }
-            // After successfully write, send a callback and wait for the read to be done
-            mDemuxCallbacks[filterId]->onFilterEvent(filterEvent);
-            // We do not wait for the last read to be done
-            // VTS can verify the read result itself.
-            if (i == SECTION_WRITE_COUNT - 1) {
-                ALOGD("[Demux] filter %d writing done. Ending thread", filterId);
-                break;
-            }
-            while (mThreadRunning[filterId]) {
                 status_t status = mFilterEventFlags[filterId]->wait(
                         static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_CONSUMED), &efState,
                         WAIT_TIMEOUT, true /* retry on spurious wake */);
@@ -388,15 +609,60 @@ void Demux::filterThreadLoop(DemuxFilterEvent* event) {
                 }
                 break;
             }
-        }
 
-        mFilterWriteCount[filterId] = 0;
+            if (mDemuxCallbacks[filterId] == nullptr) {
-        mThreadRunning[filterId] = false;
+                ALOGD("[Demux] filter %d does not hava callback. Ending thread", filterId);
+                break;
+            }
+
+            while (mFilterThreadRunning[filterId]) {
+                std::lock_guard<std::mutex> lock(mFilterEventLock);
+                if (mFilterEvents[filterId].events.size() == 0) {
+                    continue;
+                }
+                // After successfully write, send a callback and wait for the read to be done
+                mDemuxCallbacks[filterId]->onFilterEvent(mFilterEvents[filterId]);
+                mFilterEvents[filterId].events.resize(0);
+                break;
+            }
+            // We do not wait for the last read to be done
+            // VTS can verify the read result itself.
+            if (i == SECTION_WRITE_COUNT - 1) {
+                ALOGD("[Demux] filter %d writing done. Ending thread", filterId);
+                break;
+            }
+        }
+        mFilterThreadRunning[filterId] = false;
     }
 
     ALOGD("[Demux] filter thread ended.");
 }
 
+void Demux::inputThreadLoop() {
+    ALOGD("[Demux] input threadLoop start.");
+    mInputThreadRunning = true;
+
+    while (mInputThreadRunning) {
+        uint32_t efState = 0;
+        status_t status =
+                mInputEventFlag->wait(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_READY),
+                                      &efState, WAIT_TIMEOUT, true /* retry on spurious wake */);
+        if (status != OK) {
+            ALOGD("[Demux] wait for data ready on the input FMQ");
+            continue;
+        }
+        // Our current implementation filter the data and write it into the filter FMQ immedaitely
+        // after the DATA_READY from the VTS/framework
+        if (!filterAndOutputData()) {
+            ALOGD("[Demux] input data failed to be filtered. Ending thread");
+            break;
+        }
+    }
+
+    mInputThreadRunning = false;
+    ALOGD("[Demux] input thread ended.");
+}
+
 }  // namespace implementation
 }  // namespace V1_0
 }  // namespace tuner

tv/tuner/1.0/default/Demux.h

@@ -19,6 +19,7 @@
 
 #include <android/hardware/tv/tuner/1.0/IDemux.h>
 #include <fmq/MessageQueue.h>
+#include <set>
 
 using namespace std;
 
@@ -43,6 +44,8 @@ class Demux : public IDemux {
   public:
     Demux(uint32_t demuxId);
 
+    ~Demux();
+
     virtual Return<Result> setFrontendDataSource(uint32_t frontendId) override;
 
     virtual Return<Result> close() override;
@@ -68,8 +71,58 @@ class Demux : public IDemux {
 
     virtual Return<void> getAvSyncTime(AvSyncHwId avSyncHwId, getAvSyncTime_cb _hidl_cb) override;
 
+    virtual Return<Result> addInput(uint32_t bufferSize, const sp<IDemuxCallback>& cb) override;
+
+    virtual Return<void> getInputQueueDesc(getInputQueueDesc_cb _hidl_cb) override;
+
+    virtual Return<Result> configureInput(const DemuxInputSettings& settings) override;
+
+    virtual Return<Result> startInput() override;
+
+    virtual Return<Result> stopInput() override;
+
+    virtual Return<Result> flushInput() override;
+
+    virtual Return<Result> removeInput() override;
+
+    virtual Return<Result> addOutput(uint32_t bufferSize, const sp<IDemuxCallback>& cb) override;
+
+    virtual Return<void> getOutputQueueDesc(getOutputQueueDesc_cb _hidl_cb) override;
+
+    virtual Return<Result> configureOutput(const DemuxOutputSettings& settings) override;
+
+    virtual Return<Result> attachOutputTsFilter(uint32_t filterId) override;
+
+    virtual Return<Result> detachOutputTsFilter(uint32_t filterId) override;
+
+    virtual Return<Result> startOutput() override;
+
+    virtual Return<Result> stopOutput() override;
+
+    virtual Return<Result> flushOutput() override;
+
+    virtual Return<Result> removeOutput() override;
+
   private:
-    virtual ~Demux();
+    // A struct that passes the arguments to a newly created filter thread
+    struct ThreadArgs {
+        Demux* user;
+        uint32_t filterId;
+    };
+
+    /**
+     * Filter handlers to handle the data filtering.
+     * They are also responsible to write the filtered output into the filter FMQ
+     * and update the filterEvent bound with the same filterId.
+     */
+    Result startSectionFilterHandler(uint32_t filterId, vector<uint8_t> data);
+    Result startPesFilterHandler(uint32_t filterId);
+    Result startTsFilterHandler();
+    Result startMediaFilterHandler(uint32_t filterId);
+    Result startRecordFilterHandler(uint32_t filterId);
+    Result startPcrFilterHandler();
+    Result startFilterLoop(uint32_t filterId);
+
     /**
      * To create a FilterMQ with the the next available Filter ID.
      * Creating Event Flag at the same time.
@@ -77,60 +130,80 @@ class Demux : public IDemux {
      *
      * Return false is any of the above processes fails.
      */
-    bool createAndSaveMQ(uint32_t bufferSize, uint32_t filterId);
+    bool createFilterMQ(uint32_t bufferSize, uint32_t filterId);
+    bool createMQ(FilterMQ* queue, EventFlag* eventFlag, uint32_t bufferSize);
     void deleteEventFlag();
     bool writeDataToFilterMQ(const std::vector<uint8_t>& data, uint32_t filterId);
-    Result startSectionFilterHandler(DemuxFilterEvent event);
+    bool readDataFromMQ();
-    Result startPesFilterHandler(DemuxFilterEvent& event);
+    bool writeSectionsAndCreateEvent(uint32_t filterId, vector<uint8_t> data);
-    Result startTsFilterHandler();
+    /**
-    Result startMediaFilterHandler(DemuxFilterEvent& event);
+     * A dispatcher to read and dispatch input data to all the started filters.
-    Result startRecordFilterHandler(DemuxFilterEvent& event);
+     * Each filter handler handles the data filtering/output writing/filterEvent updating.
-    Result startPcrFilterHandler();
+     */
-    bool writeSectionsAndCreateEvent(DemuxFilterEvent& event, uint32_t sectionNum);
+    bool filterAndOutputData();
-    void filterThreadLoop(DemuxFilterEvent* event);
+    static void* __threadLoopFilter(void* data);
-    static void* __threadLoop(void* data);
+    static void* __threadLoopInput(void* user);
+    void filterThreadLoop(uint32_t filterId);
+    void inputThreadLoop();
 
     uint32_t mDemuxId;
     uint32_t mSourceFrontendId;
     /**
-     * Record the last used filer id. Initial value is -1.
+     * Record the last used filter id. Initial value is -1.
      * Filter Id starts with 0.
      */
     uint32_t mLastUsedFilterId = -1;
+    /**
+     * Record all the used filter Ids.
+     * Any removed filter id should be removed from this set.
+     */
+    set<uint32_t> mUsedFilterIds;
+    /**
+     * Record all the unused filter Ids within mLastUsedFilterId.
+     * Removed filter Id should be added into this set.
+     * When this set is not empty, ids here should be allocated first
+     * and added into usedFilterIds.
+     */
+    set<uint32_t> mUnusedFilterIds;
     /**
      * A list of created FilterMQ ptrs.
      * The array number is the filter ID.
      */
     vector<unique_ptr<FilterMQ>> mFilterMQs;
-    vector<DemuxFilterType> mFilterTypes;
     vector<EventFlag*> mFilterEventFlags;
+    vector<DemuxFilterEvent> mFilterEvents;
+    unique_ptr<FilterMQ> mInputMQ;
+    unique_ptr<FilterMQ> mOutputMQ;
+    EventFlag* mInputEventFlag;
+    EventFlag* mOutputEventFlag;
     /**
      * Demux callbacks used on filter events or IO buffer status
      */
     vector<sp<IDemuxCallback>> mDemuxCallbacks;
-    /**
+    sp<IDemuxCallback> mInputCallback;
-     * How many times a specific filter has written since started
+    sp<IDemuxCallback> mOutputCallback;
-     */
+    // Thread handlers
-    vector<uint16_t> mFilterWriteCount;
+    pthread_t mInputThread;
-    pthread_t mThreadId = 0;
+    pthread_t mOutputThread;
+    vector<pthread_t> mFilterThreads;
     /**
      * If a specific filter's writing loop is still running
      */
-    vector<bool> mThreadRunning;
+    vector<bool> mFilterThreadRunning;
+    bool mInputThreadRunning;
     /**
      * Lock to protect writes to the FMQs
      */
     std::mutex mWriteLock;
+    /**
+     * Lock to protect writes to the filter event
+     */
+    std::mutex mFilterEventLock;
     /**
      * How many times a filter should write
      * TODO make this dynamic/random/can take as a parameter
      */
     const uint16_t SECTION_WRITE_COUNT = 10;
-    // A struct that passes the arguments to a newly created filter thread
-    struct ThreadArgs {
-        Demux* user;
-        DemuxFilterEvent* event;
-    };
 };
 
 }  // namespace implementation

tv/tuner/1.0/types.hal

@@ -479,3 +479,99 @@ typedef uint32_t AvSyncHwId;
  *  framework and apps.
  */
 typedef vec<uint8_t> TunerKeyToken;
+
+/**
+ * A data format in demux's output or input according to ISO/IEC 13818-1.
+ */
+@export
+enum DemuxDataFormat : uint32_t {
+    /* Data is Transport Stream. */
+    TS,
+    /* Data is Packetized Elementary Stream. */
+    PES,
+    /* Data is Elementary Stream. */
+    ES,
+};
+
+/**
+ * A status of the demux's output.
+ */
+typedef DemuxFilterStatus DemuxOutputStatus;
+
+/**
+ *  The Settings for the demux's output.
+ */
+struct DemuxOutputSettings {
+    /**
+     * Register for interested status events so that the HAL can send these
+     * status events back to client.
+     */
+    bitfield<DemuxOutputStatus> statusMask;
+    /**
+     * Unconsumed data size in bytes in the output. The HAL uses it to trigger
+     * DemuxOutputStatus::LOW_WATER.
+     */
+    uint32_t lowThreshold;
+    /**
+     * Unconsumed data size in bytes in the output. The HAL uses it to trigger
+     * DemuxOutputStatus::High_WATER.
+     */
+    uint32_t highThreshold;
+    /**
+     * The data format in the output.
+     */
+    DemuxDataFormat dataFormat;
+    /**
+     * The packet size in bytes in the output.
+     */
+    uint8_t packetSize;
+};
+
+/**
+ * A status of the demux's input.
+ */
+@export
+enum DemuxInputStatus : uint32_t {
+    /**
+     * The space of the demux's input is empty.
+     */
+    SPACE_EMPTY        = 1 << 0,
+    /**
+     * The spece of the demux's input is almost empty.
+     */
+    SPACE_ALMOST_EMPTY = 1 << 1,
+    /**
+     * The space of the demux's input is almost full.
+     */
+    SPACE_ALMOST_FULL  = 1 << 2,
+    /**
+     * The space of the demux's input is full.
+     */
+    SPACE_FULL         = 1 << 3,
+};
+
+struct DemuxInputSettings {
+    /**
+     * Register for interested status events so that the HAL can send these
+     * status events back to client.
+     */
+    bitfield<DemuxInputStatus> statusMask;
+    /**
+     * Unused space size in bytes in the input. The HAL uses it to trigger
+     * DemuxInputStatus::SPACE_ALMOST_EMPTY.
+     */
+    uint32_t lowThreshold;
+    /**
+     * Unused space size in bytes in the input. The HAL uses it to trigger
+     * DemuxInputStatus::SPACE_ALMOST_FULL.
+     */
+    uint32_t highThreshold;
+    /**
+     * The data format in the input.
+     */
+    DemuxDataFormat dataFormat;
+    /**
+     * The packet size in bytes in the input.
+     */
+    uint8_t packetSize;
+};

tv/tuner/1.0/vts/functional/VtsHalTvTunerV1_0TargetTest.cpp

@@ -34,6 +34,8 @@
 #include <hidlmemory/FrameworkUtils.h>
 #include <utils/Condition.h>
 #include <utils/Mutex.h>
+#include <fstream>
+#include <iostream>
 
 #define WAIT_TIMEOUT 3000000000
 
@@ -53,11 +55,16 @@ using android::hardware::MessageQueue;
 using android::hardware::MQDescriptorSync;
 using android::hardware::Return;
 using android::hardware::Void;
+using android::hardware::tv::tuner::V1_0::DemuxDataFormat;
 using android::hardware::tv::tuner::V1_0::DemuxFilterEvent;
 using android::hardware::tv::tuner::V1_0::DemuxFilterPesEvent;
 using android::hardware::tv::tuner::V1_0::DemuxFilterSectionEvent;
+using android::hardware::tv::tuner::V1_0::DemuxFilterSettings;
 using android::hardware::tv::tuner::V1_0::DemuxFilterStatus;
 using android::hardware::tv::tuner::V1_0::DemuxFilterType;
+using android::hardware::tv::tuner::V1_0::DemuxInputSettings;
+using android::hardware::tv::tuner::V1_0::DemuxInputStatus;
+using android::hardware::tv::tuner::V1_0::DemuxOutputStatus;
 using android::hardware::tv::tuner::V1_0::DemuxQueueNotifyBits;
 using android::hardware::tv::tuner::V1_0::FrontendAtscModulation;
 using android::hardware::tv::tuner::V1_0::FrontendAtscSettings;
@@ -77,9 +84,9 @@ using android::hardware::tv::tuner::V1_0::Result;
 namespace {
 
 using FilterMQ = MessageQueue<uint8_t, kSynchronizedReadWrite>;
-using FilterMQDesc = MQDescriptorSync<uint8_t>;
+using MQDesc = MQDescriptorSync<uint8_t>;
 
-const std::vector<uint8_t> goldenDataInputBuffer{
+const std::vector<uint8_t> goldenDataOutputBuffer{
         0x00, 0x00, 0x00, 0x01, 0x09, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x67, 0x42, 0xc0, 0x1e, 0xdb,
         0x01, 0x40, 0x16, 0xec, 0x04, 0x40, 0x00, 0x00, 0x03, 0x00, 0x40, 0x00, 0x00, 0x0f, 0x03,
         0xc5, 0x8b, 0xb8, 0x00, 0x00, 0x00, 0x01, 0x68, 0xca, 0x8c, 0xb2, 0x00, 0x00, 0x01, 0x06,
@@ -119,10 +126,21 @@ const std::vector<uint8_t> goldenDataInputBuffer{
 };
 
 const uint16_t FMQ_SIZE_4K = 0x1000;
+const uint32_t FMQ_SIZE_1M = 0x100000;
 // Equal to SECTION_WRITE_COUNT on the HAL impl side
 // The HAL impl will repeatedly write to the FMQ the count times
 const uint16_t SECTION_READ_COUNT = 10;
 
+struct FilterConf {
+    DemuxFilterType type;
+    DemuxFilterSettings setting;
+};
+
+struct InputConf {
+    string inputDataFile;
+    DemuxInputSettings setting;
+};
+
 class FrontendCallback : public IFrontendCallback {
   public:
     virtual Return<void> onEvent(FrontendEventType frontendEventType) override {
@@ -184,8 +202,10 @@ void FrontendCallback::testOnDiseqcMessage(sp<IFrontend>& frontend, FrontendSett
 class DemuxCallback : public IDemuxCallback {
   public:
     virtual Return<void> onFilterEvent(const DemuxFilterEvent& filterEvent) override {
+        ALOGW("[VTS] FILTER EVENT %d", filterEvent.filterId);
         android::Mutex::Autolock autoLock(mMsgLock);
         mFilterEventReceived = true;
+        // maybe assemble here??
         mFilterEvent = filterEvent;
         mMsgCondition.signal();
         return Void();
@@ -196,24 +216,53 @@ class DemuxCallback : public IDemuxCallback {
         return Void();
     }
 
+    virtual Return<void> onOutputStatus(DemuxOutputStatus /*status*/) override { return Void(); }
+
+    virtual Return<void> onInputStatus(DemuxInputStatus status) override {
+        // android::Mutex::Autolock autoLock(mMsgLock);
+        switch (status) {
+            case DemuxInputStatus::SPACE_EMPTY:
+            case DemuxInputStatus::SPACE_ALMOST_EMPTY:
+                mKeepWritingInputFMQ = true;
+                break;
+            case DemuxInputStatus::SPACE_ALMOST_FULL:
+            case DemuxInputStatus::SPACE_FULL:
+                mKeepWritingInputFMQ = false;
+                break;
+        }
+        return Void();
+    }
+
     void testOnFilterEvent(uint32_t filterId);
-    void testOnSectionFilterEvent(sp<IDemux>& demux, uint32_t filterId,
+    void testOnSectionFilterEvent(sp<IDemux>& demux, uint32_t filterId, MQDesc& filterMQDescriptor,
-                                  FilterMQDesc& filterMQDescriptor);
+                                  MQDesc& inputMQDescriptor);
-    void testOnPesFilterEvent(sp<IDemux>& demux, uint32_t filterId,
+    void startPlaybackInputThread(InputConf inputConf, MQDesc& inputMQDescriptor);
-                              FilterMQDesc& filterMQDescriptor);
+    bool readAndCompareSectionEventData();
-    void readAndCompareSectionEventData();
+
-    void readAndComparePesEventData();
+    static void* __threadLoopInput(void* threadArgs);
+    void inputThreadLoop(InputConf inputConf, bool* keepWritingInputFMQ, MQDesc& inputMQDescriptor);
 
   private:
+    struct InputThreadArgs {
+        DemuxCallback* user;
+        InputConf inputConf;
+        bool* keepWritingInputFMQ;
+        MQDesc& inputMQDesc;
+    };
     bool mFilterEventReceived = false;
     std::vector<uint8_t> mDataOutputBuffer;
     std::unique_ptr<FilterMQ> mFilterMQ;
+    std::unique_ptr<FilterMQ> mInputMQ;
     uint16_t mDataLength = 0;
     DemuxFilterEvent mFilterEvent;
     android::Mutex mMsgLock;
     android::Mutex mReadLock;
     android::Condition mMsgCondition;
     EventFlag* mFilterMQEventFlag;
+    EventFlag* mInputMQEventFlag;
+    bool mKeepWritingInputFMQ;
+    bool mInputThreadRunning;
+    pthread_t mInputThread;
 };
 
 void DemuxCallback::testOnFilterEvent(uint32_t filterId) {
@@ -230,83 +279,138 @@ void DemuxCallback::testOnFilterEvent(uint32_t filterId) {
     EXPECT_TRUE(filterId == mFilterEvent.filterId) << "filter id match";
 }
 
+void DemuxCallback::startPlaybackInputThread(InputConf inputConf, MQDesc& inputMQDescriptor) {
+    struct InputThreadArgs* threadArgs =
+            (struct InputThreadArgs*)malloc(sizeof(struct InputThreadArgs));
+    threadArgs->user = this;
+    threadArgs->inputConf = inputConf;
+    threadArgs->keepWritingInputFMQ = &mKeepWritingInputFMQ;
+    threadArgs->inputMQDesc = inputMQDescriptor;
+
+    pthread_create(&mInputThread, NULL, __threadLoopInput, (void*)threadArgs);
+    pthread_setname_np(mInputThread, "test_playback_input_loop");
+}
+
+/*void DemuxCallback::testPlaybackDataFlow(bool* keepWritingInputFMQ) {
+    // timeout logic here
+
+    // assemble logic here
+
+
+}*/
+
 void DemuxCallback::testOnSectionFilterEvent(sp<IDemux>& demux, uint32_t filterId,
-                                             FilterMQDesc& filterMQDescriptor) {
+                                             MQDesc& filterMQDescriptor,
+                                             MQDesc& inputMQDescriptor) {
     Result status;
     // Create MQ to read the output into the local buffer
     mFilterMQ = std::make_unique<FilterMQ>(filterMQDescriptor, true /* resetPointers */);
     EXPECT_TRUE(mFilterMQ);
+    // Get the MQ to write the input to the HAL
+    mInputMQ = std::make_unique<FilterMQ>(inputMQDescriptor, true /* resetPointers */);
+    EXPECT_TRUE(mInputMQ);
     // Create the EventFlag that is used to signal the HAL impl that data have been
     // read the Filter FMQ
     EXPECT_TRUE(EventFlag::createEventFlag(mFilterMQ->getEventFlagWord(), &mFilterMQEventFlag) ==
                 android::OK);
+    // Create the EventFlag that is used to signal the HAL impl that data have been
+    // written into the Input FMQ
+    EXPECT_TRUE(EventFlag::createEventFlag(mInputMQ->getEventFlagWord(), &mInputMQEventFlag) ==
+                android::OK);
     // Start filter
     status = demux->startFilter(filterId);
+    status = demux->startInput();
+
     EXPECT_EQ(status, Result::SUCCESS);
     // Test start filter and receive callback event
     for (int i = 0; i < SECTION_READ_COUNT; i++) {
+        // Write input FMQ and notify the Tuner Implementation
+        EXPECT_TRUE(mInputMQ->write(goldenDataOutputBuffer.data(), goldenDataOutputBuffer.size()));
+        mInputMQEventFlag->wake(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_READY));
         testOnFilterEvent(filterId);
         // checksum of mDataOutputBuffer and Input golden input
-        readAndCompareSectionEventData();
+        if (readAndCompareSectionEventData() && i < SECTION_READ_COUNT - 1) {
+            mFilterMQEventFlag->wake(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_CONSUMED));
+        }
     }
 }
 
-void DemuxCallback::testOnPesFilterEvent(sp<IDemux>& demux, uint32_t filterId,
+bool DemuxCallback::readAndCompareSectionEventData() {
-                                         FilterMQDesc& filterMQDescriptor) {
-    Result status;
-    // Create MQ to read the output into the local buffer
-    mFilterMQ = std::make_unique<FilterMQ>(filterMQDescriptor, true /* resetPointers */);
-    EXPECT_TRUE(mFilterMQ);
-    // Create the EventFlag that is used to signal the HAL impl that data have been
-    // read the Filter FMQ
-    EXPECT_TRUE(EventFlag::createEventFlag(mFilterMQ->getEventFlagWord(), &mFilterMQEventFlag) ==
-                android::OK);
-    // Start filter
-    status = demux->startFilter(filterId);
-    EXPECT_EQ(status, Result::SUCCESS);
-    // Test start filter and receive callback event
-    testOnFilterEvent(filterId);
-    // checksum of mDataOutputBuffer and Input golden input
-    readAndComparePesEventData();
-}
-
-void DemuxCallback::readAndCompareSectionEventData() {
     bool result = false;
     for (int i = 0; i < mFilterEvent.events.size(); i++) {
         DemuxFilterSectionEvent event = mFilterEvent.events[i].section();
         mDataLength = event.dataLength;
-        EXPECT_TRUE(mDataLength == goldenDataInputBuffer.size()) << "buffer size does not match";
+        EXPECT_TRUE(mDataLength == goldenDataOutputBuffer.size()) << "buffer size does not match";
 
         mDataOutputBuffer.resize(mDataLength);
         result = mFilterMQ->read(mDataOutputBuffer.data(), mDataLength);
         EXPECT_TRUE(result) << "can't read from Filter MQ";
 
         for (int i = 0; i < mDataLength; i++) {
-            EXPECT_TRUE(goldenDataInputBuffer[i] == mDataOutputBuffer[i]) << "data does not match";
+            EXPECT_TRUE(goldenDataOutputBuffer[i] == mDataOutputBuffer[i]) << "data does not match";
         }
     }
-    if (result) {
+    return result;
-        mFilterMQEventFlag->wake(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_CONSUMED));
-    }
 }
 
-void DemuxCallback::readAndComparePesEventData() {
+void* DemuxCallback::__threadLoopInput(void* threadArgs) {
-    // TODO handle multiple events in one filter callback event
+    DemuxCallback* const self =
-    DemuxFilterPesEvent event = mFilterEvent.events[0].pes();
+            static_cast<DemuxCallback*>(((struct InputThreadArgs*)threadArgs)->user);
-    mDataLength = event.dataLength;
+    self->inputThreadLoop(((struct InputThreadArgs*)threadArgs)->inputConf,
-    EXPECT_TRUE(mDataLength == goldenDataInputBuffer.size()) << "buffer size does not match";
+                          ((struct InputThreadArgs*)threadArgs)->keepWritingInputFMQ,
+                          ((struct InputThreadArgs*)threadArgs)->inputMQDesc);
+    return 0;
+}
 
-    mDataOutputBuffer.resize(mDataLength);
+void DemuxCallback::inputThreadLoop(InputConf inputConf, bool* keepWritingInputFMQ,
-    bool result = mFilterMQ->read(mDataOutputBuffer.data(), mDataLength);
+                                    MQDesc& inputMQDescriptor) {
-    EXPECT_TRUE(result) << "can't read from Filter MQ";
+    mInputThreadRunning = true;
 
-    if (result) {
+    std::unique_ptr inputMQ =
-        mFilterMQEventFlag->wake(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_CONSUMED));
+            std::make_unique<FilterMQ>(inputMQDescriptor, true /* resetPointers */);
+    EXPECT_TRUE(inputMQ);
+
+    // Create the EventFlag that is used to signal the HAL impl that data have been
+    // written into the Input FMQ
+    EventFlag* inputMQEventFlag;
+    EXPECT_TRUE(EventFlag::createEventFlag(inputMQ->getEventFlagWord(), &inputMQEventFlag) ==
+                android::OK);
+
+    // open the stream and get its length
+    std::ifstream inputData(inputConf.inputDataFile /*"ts/test1.ts"*/, std::ifstream::binary);
+    int writeSize = inputConf.setting.packetSize * 6;
+    char* buffer = new char[writeSize];
+    if (!inputData) {
+        // log
+        mInputThreadRunning = false;
     }
 
-    for (int i = 0; i < mDataLength; i++) {
+    while (mInputThreadRunning) {
-        EXPECT_TRUE(goldenDataInputBuffer[i] == mDataOutputBuffer[i]) << "data does not match";
+        // move the stream pointer for packet size * 2k? every read until end
+        while (*keepWritingInputFMQ) {
+            inputData.read(buffer, writeSize);
+            if (!inputData) {
+                int leftSize = inputData.gcount();
+                inputData.clear();
+                inputData.read(buffer, leftSize);
+                // Write the left over of the input data and quit the thread
+                if (leftSize > 0) {
+                    EXPECT_TRUE(inputMQ->write((unsigned char*)&buffer[0],
+                                               leftSize / inputConf.setting.packetSize));
+                    inputMQEventFlag->wake(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_READY));
+                }
+                mInputThreadRunning = false;
+                break;
+            }
+            // Write input FMQ and notify the Tuner Implementation
+            EXPECT_TRUE(inputMQ->write((unsigned char*)&buffer[0], 6));
+            inputMQEventFlag->wake(static_cast<uint32_t>(DemuxQueueNotifyBits::DATA_READY));
+            inputData.seekg(writeSize, inputData.cur);
+        }
     }
+
+    delete[] buffer;
+    inputData.close();
 }
 
 // Test environment for Tuner HIDL HAL.
@@ -341,24 +445,33 @@ class TunerHidlTest : public ::testing::VtsHalHidlTargetTestBase {
     sp<IDescrambler> mDescrambler;
     sp<IDemux> mDemux;
     sp<DemuxCallback> mDemuxCallback;
-    FilterMQDesc mFilterMQDescriptor;
+    MQDesc mFilterMQDescriptor;
+    MQDesc mInputMQDescriptor;
+
     uint32_t mDemuxId;
     uint32_t mFilterId;
 
+    pthread_t mInputThread;
+    bool mInputThreadRunning;
+
     ::testing::AssertionResult createFrontend(int32_t frontendId);
     ::testing::AssertionResult tuneFrontend(int32_t frontendId);
     ::testing::AssertionResult stopTuneFrontend(int32_t frontendId);
     ::testing::AssertionResult closeFrontend(int32_t frontendId);
     ::testing::AssertionResult createDemux();
     ::testing::AssertionResult createDemuxWithFrontend(int32_t frontendId);
+    ::testing::AssertionResult getInputMQDescriptor();
+    ::testing::AssertionResult addInputToDemux(DemuxInputSettings setting);
     ::testing::AssertionResult addSectionFilterToDemux();
-    ::testing::AssertionResult addPesFilterToDemux();
+    ::testing::AssertionResult addFilterToDemux(DemuxFilterType type, DemuxFilterSettings setting);
-    ::testing::AssertionResult getFilterMQDescriptor(sp<IDemux>& demux, const uint32_t filterId);
+    ::testing::AssertionResult getFilterMQDescriptor(const uint32_t filterId);
-    ::testing::AssertionResult readSectionFilterDataOutput();
-    ::testing::AssertionResult readPesFilterDataOutput();
     ::testing::AssertionResult closeDemux();
     ::testing::AssertionResult createDescrambler();
     ::testing::AssertionResult closeDescrambler();
+
+    ::testing::AssertionResult readSectionFilterDataOutput();
+    ::testing::AssertionResult playbackDataFlowTest(vector<FilterConf> filterConf,
+                                                    InputConf inputConf, string goldenOutput);
 };
 
 ::testing::AssertionResult TunerHidlTest::createFrontend(int32_t frontendId) {
@@ -405,7 +518,7 @@ class TunerHidlTest : public ::testing::VtsHalHidlTargetTestBase {
 
 ::testing::AssertionResult TunerHidlTest::stopTuneFrontend(int32_t frontendId) {
     Result status;
-    if (createFrontend(frontendId) == ::testing::AssertionFailure()) {
+    if (!mFrontend && createFrontend(frontendId) == ::testing::AssertionFailure()) {
         return ::testing::AssertionFailure();
     }
 
@@ -415,11 +528,12 @@ class TunerHidlTest : public ::testing::VtsHalHidlTargetTestBase {
 
 ::testing::AssertionResult TunerHidlTest::closeFrontend(int32_t frontendId) {
     Result status;
-    if (createFrontend(frontendId) == ::testing::AssertionFailure()) {
+    if (!mFrontend && createFrontend(frontendId) == ::testing::AssertionFailure()) {
         return ::testing::AssertionFailure();
     }
 
     status = mFrontend->close();
+    mFrontend = nullptr;
     return ::testing::AssertionResult(status == Result::SUCCESS);
 }
 
@@ -437,11 +551,11 @@ class TunerHidlTest : public ::testing::VtsHalHidlTargetTestBase {
 ::testing::AssertionResult TunerHidlTest::createDemuxWithFrontend(int32_t frontendId) {
     Result status;
 
-    if (createDemux() == ::testing::AssertionFailure()) {
+    if (!mDemux && createDemux() == ::testing::AssertionFailure()) {
         return ::testing::AssertionFailure();
     }
 
-    if (createFrontend(frontendId) == ::testing::AssertionFailure()) {
+    if (!mFrontend && createFrontend(frontendId) == ::testing::AssertionFailure()) {
         return ::testing::AssertionFailure();
     }
 
@@ -450,111 +564,14 @@ class TunerHidlTest : public ::testing::VtsHalHidlTargetTestBase {
     return ::testing::AssertionResult(status == Result::SUCCESS);
 }
 
-::testing::AssertionResult TunerHidlTest::addSectionFilterToDemux() {
-    Result status;
-
-    if (createDemux() == ::testing::AssertionFailure()) {
-        return ::testing::AssertionFailure();
-    }
-
-    // Create demux callback
-    mDemuxCallback = new DemuxCallback();
-
-    // Add section filter to the local demux
-    mDemux->addFilter(DemuxFilterType::SECTION, FMQ_SIZE_4K, mDemuxCallback,
-                      [&](Result result, uint32_t filterId) {
-                          mFilterId = filterId;
-                          status = result;
-                      });
-
-    // Add another section filter to the local demux
-    mDemux->addFilter(DemuxFilterType::SECTION, FMQ_SIZE_4K, mDemuxCallback,
-                      [&](Result result, uint32_t filterId) {
-                          mFilterId = filterId;
-                          status = result;
-                      });
-
-    // TODO Test configure the filter
-
-    return ::testing::AssertionResult(status == Result::SUCCESS);
-}
-
-::testing::AssertionResult TunerHidlTest::addPesFilterToDemux() {
-    Result status;
-
-    if (createDemux() == ::testing::AssertionFailure()) {
-        return ::testing::AssertionFailure();
-    }
-
-    // Create demux callback
-    mDemuxCallback = new DemuxCallback();
-
-    // Add PES filter to the local demux
-    mDemux->addFilter(DemuxFilterType::PES, FMQ_SIZE_4K, mDemuxCallback,
-                      [&](Result result, uint32_t filterId) {
-                          mFilterId = filterId;
-                          status = result;
-                      });
-
-    // Add another PES filter to the local demux
-    mDemux->addFilter(DemuxFilterType::PES, FMQ_SIZE_4K, mDemuxCallback,
-                      [&](Result result, uint32_t filterId) {
-                          mFilterId = filterId;
-                          status = result;
-                      });
-
-    // TODO Test configure the filter
-
-    return ::testing::AssertionResult(status == Result::SUCCESS);
-}
-
-::testing::AssertionResult TunerHidlTest::getFilterMQDescriptor(sp<IDemux>& demux,
-                                                                const uint32_t filterId) {
-    Result status;
-
-    if (!demux) {
-        return ::testing::AssertionFailure();
-    }
-
-    mDemux->getFilterQueueDesc(filterId, [&](Result result, const FilterMQDesc& filterMQDesc) {
-        mFilterMQDescriptor = filterMQDesc;
-        status = result;
-    });
-
-    return ::testing::AssertionResult(status == Result::SUCCESS);
-}
-
-::testing::AssertionResult TunerHidlTest::readSectionFilterDataOutput() {
-    if (addSectionFilterToDemux() == ::testing::AssertionFailure() ||
-        getFilterMQDescriptor(mDemux, mFilterId) == ::testing::AssertionFailure()) {
-        return ::testing::AssertionFailure();
-    }
-
-    // Test start filter and read the output data
-    mDemuxCallback->testOnSectionFilterEvent(mDemux, mFilterId, mFilterMQDescriptor);
-
-    return ::testing::AssertionResult(true);
-}
-
-::testing::AssertionResult TunerHidlTest::readPesFilterDataOutput() {
-    if (addPesFilterToDemux() == ::testing::AssertionFailure() ||
-        getFilterMQDescriptor(mDemux, mFilterId) == ::testing::AssertionFailure()) {
-        return ::testing::AssertionFailure();
-    }
-
-    // Test start filter and read the output data
-    mDemuxCallback->testOnPesFilterEvent(mDemux, mFilterId, mFilterMQDescriptor);
-
-    return ::testing::AssertionResult(true);
-}
-
 ::testing::AssertionResult TunerHidlTest::closeDemux() {
     Result status;
-    if (createDemux() == ::testing::AssertionFailure()) {
+    if (!mDemux && createDemux() == ::testing::AssertionFailure()) {
         return ::testing::AssertionFailure();
     }
 
     status = mDemux->close();
+    mDemux = nullptr;
     return ::testing::AssertionResult(status == Result::SUCCESS);
 }
 
@@ -569,7 +586,7 @@ class TunerHidlTest : public ::testing::VtsHalHidlTargetTestBase {
         return ::testing::AssertionFailure();
     }
 
-    if (createDemux() == ::testing::AssertionFailure()) {
+    if (!mDemux && createDemux() == ::testing::AssertionFailure()) {
         return ::testing::AssertionFailure();
     }
 
@@ -585,14 +602,185 @@ class TunerHidlTest : public ::testing::VtsHalHidlTargetTestBase {
 
 ::testing::AssertionResult TunerHidlTest::closeDescrambler() {
     Result status;
-    if (createDescrambler() == ::testing::AssertionFailure()) {
+    if (!mDescrambler && createDescrambler() == ::testing::AssertionFailure()) {
         return ::testing::AssertionFailure();
     }
 
     status = mDescrambler->close();
+    mDescrambler = nullptr;
     return ::testing::AssertionResult(status == Result::SUCCESS);
 }
 
+::testing::AssertionResult TunerHidlTest::addInputToDemux(DemuxInputSettings setting) {
+    Result status;
+
+    if (!mDemux && createDemux() == ::testing::AssertionFailure()) {
+        return ::testing::AssertionFailure();
+    }
+
+    // Create demux callback
+    if (!mDemuxCallback) {
+        mDemuxCallback = new DemuxCallback();
+    }
+
+    // Add section filter to the local demux
+    status = mDemux->addInput(FMQ_SIZE_1M, mDemuxCallback);
+
+    if (status != Result::SUCCESS) {
+        return ::testing::AssertionFailure();
+    }
+
+    status = mDemux->configureInput(setting);
+
+    return ::testing::AssertionResult(status == Result::SUCCESS);
+}
+
+::testing::AssertionResult TunerHidlTest::getInputMQDescriptor() {
+    Result status;
+
+    if (!mDemux && createDemux() == ::testing::AssertionFailure()) {
+        return ::testing::AssertionFailure();
+    }
+
+    mDemux->getInputQueueDesc([&](Result result, const MQDesc& inputMQDesc) {
+        mInputMQDescriptor = inputMQDesc;
+        status = result;
+    });
+
+    return ::testing::AssertionResult(status == Result::SUCCESS);
+}
+
+::testing::AssertionResult TunerHidlTest::addSectionFilterToDemux() {
+    Result status;
+
+    if (!mDemux && createDemux() == ::testing::AssertionFailure()) {
+        return ::testing::AssertionFailure();
+    }
+
+    // Create demux callback
+    if (!mDemuxCallback) {
+        mDemuxCallback = new DemuxCallback();
+    }
+
+    // Add section filter to the local demux
+    mDemux->addFilter(DemuxFilterType::SECTION, FMQ_SIZE_4K, mDemuxCallback,
+                      [&](Result result, uint32_t filterId) {
+                          mFilterId = filterId;
+                          status = result;
+                      });
+
+    return ::testing::AssertionResult(status == Result::SUCCESS);
+}
+
+::testing::AssertionResult TunerHidlTest::addFilterToDemux(DemuxFilterType type,
+                                                           DemuxFilterSettings setting) {
+    Result status;
+
+    if (!mDemux && createDemux() == ::testing::AssertionFailure()) {
+        return ::testing::AssertionFailure();
+    }
+
+    // Create demux callback
+    if (!mDemuxCallback) {
+        mDemuxCallback = new DemuxCallback();
+    }
+
+    // Add filter to the local demux
+    mDemux->addFilter(type, FMQ_SIZE_4K, mDemuxCallback, [&](Result result, uint32_t filterId) {
+        // TODO use a map to save all the filter id and FMQ
+        mFilterId = filterId;
+        status = result;
+    });
+
+    if (status != Result::SUCCESS) {
+        return ::testing::AssertionFailure();
+    }
+
+    // Configure the filter
+    status = mDemux->configureFilter(mFilterId, setting);
+
+    return ::testing::AssertionResult(status == Result::SUCCESS);
+}
+
+::testing::AssertionResult TunerHidlTest::getFilterMQDescriptor(const uint32_t filterId) {
+    Result status;
+
+    if (!mDemux) {
+        return ::testing::AssertionFailure();
+    }
+
+    mDemux->getFilterQueueDesc(filterId, [&](Result result, const MQDesc& filterMQDesc) {
+        mFilterMQDescriptor = filterMQDesc;
+        status = result;
+    });
+
+    return ::testing::AssertionResult(status == Result::SUCCESS);
+}
+
+::testing::AssertionResult TunerHidlTest::readSectionFilterDataOutput() {
+    // Filter Configuration Module
+    DemuxInputSettings setting{
+            .statusMask = 0xf,
+            .lowThreshold = 0x1000,
+            .highThreshold = 0x100000,
+            .dataFormat = DemuxDataFormat::TS,
+            .packetSize = 188,
+    };
+    if (addSectionFilterToDemux() == ::testing::AssertionFailure() ||
+        getFilterMQDescriptor(mFilterId) == ::testing::AssertionFailure() ||
+        addInputToDemux(setting) == ::testing::AssertionFailure() ||
+        getInputMQDescriptor() == ::testing::AssertionFailure()) {
+        return ::testing::AssertionFailure();
+    }
+
+    // Data Verify Module
+    // Test start filter and read the output data
+    mDemuxCallback->testOnSectionFilterEvent(mDemux, mFilterId, mFilterMQDescriptor,
+                                             mInputMQDescriptor);
+
+    // Clean Up Module
+    return closeDemux();  //::testing::AssertionSuccess();
+}
+
+::testing::AssertionResult TunerHidlTest::playbackDataFlowTest(vector<FilterConf> filterConf,
+                                                               InputConf inputConf,
+                                                               string /*goldenOutput*/) {
+    Result status;
+    // Filter Configuration Module
+    for (int i = 0; i < filterConf.size(); i++) {
+        if (addFilterToDemux(filterConf[i].type, filterConf[i].setting) ==
+                    ::testing::AssertionFailure() ||
+            // TODO use a map to save the FMQs/EvenFlags and pass to callback
+            getFilterMQDescriptor(mFilterId) == ::testing::AssertionFailure()) {
+            return ::testing::AssertionFailure();
+        }
+    }
+
+    // Playback Input Module
+    DemuxInputSettings inputSetting = inputConf.setting;
+    if (addInputToDemux(inputSetting) == ::testing::AssertionFailure() ||
+        getInputMQDescriptor() == ::testing::AssertionFailure()) {
+        return ::testing::AssertionFailure();
+    }
+    mDemuxCallback->startPlaybackInputThread(inputConf, mInputMQDescriptor);
+    status = mDemux->startInput();
+    if (status != Result::SUCCESS) {
+        return ::testing::AssertionFailure();
+    }
+
+    // Data Verify Module
+    // golden output, created FMQ to read and EventFlags to DATA_CONSUMED
+    // Maintain each filter's real output (and how to assemble?????)
+    // mDemuxCallback->testPlaybackDataFlow();
+
+    // Clean Up Module
+    // TODO what about remove input, remove filters
+    return closeDemux();
+}
+
+/*
+ * API STATUS TESTS
+ */
 TEST_F(TunerHidlTest, CreateFrontend) {
     Result status;
     hidl_vec<FrontendId> feIds;
@@ -673,12 +861,6 @@ TEST_F(TunerHidlTest, CloseFrontend) {
     }
 }
 
-TEST_F(TunerHidlTest, CreateDemux) {
-    description("Create Demux");
-
-    ASSERT_TRUE(createDemux());
-}
-
 TEST_F(TunerHidlTest, CreateDemuxWithFrontend) {
     Result status;
     hidl_vec<FrontendId> feIds;
@@ -699,50 +881,34 @@ TEST_F(TunerHidlTest, CreateDemuxWithFrontend) {
     }
 }
 
-TEST_F(TunerHidlTest, AddSectionFilterToDemux) {
+TEST_F(TunerHidlTest, CreateDemux) {
-    description("Add a section filter to a created demux");
+    description("Create Demux");
-    ASSERT_TRUE(addSectionFilterToDemux());
+    ASSERT_TRUE(createDemux());
-}
-
-TEST_F(TunerHidlTest, AddPesFilterToDemux) {
-    description("Add a pes filter to a created demux");
-    ASSERT_TRUE(addPesFilterToDemux());
-}
-
-TEST_F(TunerHidlTest, GetFilterMQDescriptor) {
-    description("Get MQ Descriptor from a created filter");
-    ASSERT_TRUE(addSectionFilterToDemux());
-    ASSERT_TRUE(getFilterMQDescriptor(mDemux, mFilterId));
-}
-
-TEST_F(TunerHidlTest, ReadSectionFilterOutput) {
-    description("Read data output from FMQ of a Section Filter");
-    ASSERT_TRUE(readSectionFilterDataOutput());
-}
-
-TEST_F(TunerHidlTest, ReadPesFilterOutput) {
-    description("Read data output from FMQ of a PES Filter");
-    ASSERT_TRUE(readPesFilterDataOutput());
 }
 
 TEST_F(TunerHidlTest, CloseDemux) {
     description("Close Demux");
-
     ASSERT_TRUE(closeDemux());
 }
 
 TEST_F(TunerHidlTest, CreateDescrambler) {
     description("Create Descrambler");
-
     ASSERT_TRUE(createDescrambler());
 }
 
 TEST_F(TunerHidlTest, CloseDescrambler) {
     description("Close Descrambler");
-
     ASSERT_TRUE(closeDescrambler());
 }
 
+/*
+ * DATA FLOW TESTS
+ */
+TEST_F(TunerHidlTest, ReadSectionFilterOutput) {
+    description("Read data output from FMQ of a Section Filter");
+    ASSERT_TRUE(readSectionFilterDataOutput());
+}
+
 }  // namespace
 
 int main(int argc, char** argv) {