//------------------------------------------------------------------------------
// File: PullPin.cpp
//
// Desc: DirectShow base classes - implements CPullPin class that pulls data
//       from IAsyncReader.
//
// Copyright (c) 1992-2001 Microsoft Corporation.  All rights reserved.
//------------------------------------------------------------------------------


#include <streams.h>
#include "pullpin.h"



CPullPin::CPullPin()
  : m_pReader(NULL),
    m_pAlloc(NULL),
    m_State(TM_Exit) {

}

CPullPin::~CPullPin() {
    Disconnect();
}

// returns S_OK if successfully connected to an IAsyncReader interface
// from this object
// Optional allocator should be proposed as a preferred allocator if
// necessary
HRESULT
CPullPin::Connect(IUnknown* pUnk, IMemAllocator* pAlloc, BOOL bSync) {
    CAutoLock lock(&m_AccessLock);

    if(m_pReader) {
        return VFW_E_ALREADY_CONNECTED;
    }

    HRESULT hr = pUnk->QueryInterface(IID_IAsyncReader, (void**)&m_pReader);
    if(FAILED(hr)) {
        return(hr);
    }

    hr = DecideAllocator(pAlloc, NULL);
    if(FAILED(hr)) {
        Disconnect();
        return hr;
    }

    LONGLONG llTotal, llAvail;
    hr = m_pReader->Length(&llTotal, &llAvail);
    if(FAILED(hr)) {
        Disconnect();
        return hr;
    }

    // convert from file position to reference time
    m_tDuration = llTotal * UNITS;
    m_tStop = m_tDuration;
    m_tStart = 0;
    m_bSync = bSync;

    return S_OK;
}

// disconnect any connection made in Connect
HRESULT
CPullPin::Disconnect() {
    CAutoLock lock(&m_AccessLock);

    StopThread();

    if(m_pReader) {
        m_pReader->Release();
        m_pReader = NULL;
    }

    if(m_pAlloc) {
        m_pAlloc->Release();
        m_pAlloc = NULL;
    }

    return S_OK;
}

// agree an allocator using RequestAllocator - optional
// props param specifies your requirements (non-zero fields).
// returns an error code if fail to match requirements.
// optional IMemAllocator interface is offered as a preferred allocator
// but no error occurs if it can't be met.
HRESULT
CPullPin::DecideAllocator(
    IMemAllocator * pAlloc,
    ALLOCATOR_PROPERTIES * pProps) {
    ALLOCATOR_PROPERTIES *pRequest;
    ALLOCATOR_PROPERTIES Request;
    if(pProps == NULL) {
        Request.cBuffers = 3;
        Request.cbBuffer = 64*1024;
        Request.cbAlign = 0;
        Request.cbPrefix = 0;
        pRequest = &Request;
    }
    else {
        pRequest = pProps;
    }
    HRESULT hr = m_pReader->RequestAllocator(pAlloc,
        pRequest,
        &m_pAlloc);
    return hr;
}

// start pulling data
HRESULT
CPullPin::Active(void) {
    ASSERT(!ThreadExists());
    return StartThread();
}

// stop pulling data
HRESULT
CPullPin::Inactive(void) {
    StopThread();

    return S_OK;
}

HRESULT
CPullPin::Seek(REFERENCE_TIME tStart, REFERENCE_TIME tStop) {
    CAutoLock lock(&m_AccessLock);

    ThreadMsg AtStart = m_State;

    if(AtStart == TM_Start) {
        BeginFlush();
        PauseThread();
        EndFlush();
    }

    m_tStart = tStart;
    m_tStop = tStop;

    HRESULT hr = S_OK;
    if(AtStart == TM_Start) {
        hr = StartThread();
    }

    return hr;
}

HRESULT
CPullPin::Duration(REFERENCE_TIME* ptDuration) {
    *ptDuration = m_tDuration;
    return S_OK;
}


HRESULT
CPullPin::StartThread() {
    CAutoLock lock(&m_AccessLock);

    if(!m_pAlloc || !m_pReader) {
        return E_UNEXPECTED;
    }

    HRESULT hr;
    if(!ThreadExists()) {

        // commit allocator
        hr = m_pAlloc->Commit();
        if(FAILED(hr)) {
            return hr;
        }

        // start thread
        if(!Create()) {
            return E_FAIL;
        }
    }

    m_State = TM_Start;
    hr = (HRESULT) CallWorker(m_State);
    return hr;
}

HRESULT
CPullPin::PauseThread() {
    CAutoLock lock(&m_AccessLock);

    if(!ThreadExists()) {
        return E_UNEXPECTED;
    }

    // need to flush to ensure the thread is not blocked
    // in WaitForNext
    HRESULT hr = m_pReader->BeginFlush();
    if(FAILED(hr)) {
        return hr;
    }

    m_State = TM_Pause;
    hr = CallWorker(TM_Pause);

    m_pReader->EndFlush();
    return hr;
}

HRESULT
CPullPin::StopThread() {
    CAutoLock lock(&m_AccessLock);

    if(!ThreadExists()) {
        return S_FALSE;
    }

    // need to flush to ensure the thread is not blocked
    // in WaitForNext
    HRESULT hr = m_pReader->BeginFlush();
    if(FAILED(hr)) {
        return hr;
    }

    m_State = TM_Exit;
    hr = CallWorker(TM_Exit);

    m_pReader->EndFlush();

    // wait for thread to completely exit
    Close();

    // decommit allocator
    if(m_pAlloc) {
        m_pAlloc->Decommit();
    }

    return S_OK;
}


DWORD
CPullPin::ThreadProc(void) {
    while(1) {
        DWORD cmd = GetRequest();
        switch(cmd) {
            case TM_Exit:
                Reply(S_OK);
                return 0;

            case TM_Pause:
                // we are paused already
                Reply(S_OK);
                break;

            case TM_Start:
                Reply(S_OK);
                Process();
                break;
        }

        // at this point, there should be no outstanding requests on the
        // upstream filter.
        // We should force begin/endflush to ensure that this is true.
        // !!!Note that we may currently be inside a BeginFlush/EndFlush pair
        // on another thread, but the premature EndFlush will do no harm now
        // that we are idle.
        m_pReader->BeginFlush();
        CleanupCancelled();
        m_pReader->EndFlush();
    }
}

HRESULT
CPullPin::QueueSample(
    REFERENCE_TIME& tCurrent,
    REFERENCE_TIME tAlignStop,
    BOOL bDiscontinuity
    ) {
    IMediaSample* pSample;

    HRESULT hr = m_pAlloc->GetBuffer(&pSample, NULL, NULL, 0);
    if(FAILED(hr)) {
        return hr;
    }

    LONGLONG tStopThis = tCurrent + (pSample->GetSize() * UNITS);
    if(tStopThis > tAlignStop) {
        tStopThis = tAlignStop;
    }
    pSample->SetTime(&tCurrent, &tStopThis);
    tCurrent = tStopThis;

    pSample->SetDiscontinuity(bDiscontinuity);

    hr = m_pReader->Request(pSample,
        0);
    if(FAILED(hr)) {
        pSample->Release();

        CleanupCancelled();
        OnError(hr);
    }
    return hr;
}

HRESULT
CPullPin::CollectAndDeliver(
    REFERENCE_TIME tStart,
    REFERENCE_TIME tStop) {
    IMediaSample* pSample = NULL;   // better be sure pSample is set
    DWORD_PTR dwUnused;
    HRESULT hr = m_pReader->WaitForNext(INFINITE,
        &pSample,
        &dwUnused);
    if(FAILED(hr)) {
        if(pSample) {
            pSample->Release();
        }
    }
    else {
        hr = DeliverSample(pSample, tStart, tStop);
    }
    if(FAILED(hr)) {
        CleanupCancelled();
        OnError(hr);
    }
    return hr;

}

HRESULT
CPullPin::DeliverSample(
    IMediaSample* pSample,
    REFERENCE_TIME tStart,
    REFERENCE_TIME tStop
    ) {
    // fix up sample if past actual stop (for sector alignment)
    REFERENCE_TIME t1, t2;
    pSample->GetTime(&t1, &t2);
    if(t2 > tStop) {
        t2 = tStop;
    }

    // adjust times to be relative to (aligned) start time
    t1 -= tStart;
    t2 -= tStart;
    pSample->SetTime(&t1, &t2);


    HRESULT hr = Receive(pSample);
    pSample->Release();
    return hr;
}

void
CPullPin::Process(void) {
    // is there anything to do?
    if(m_tStop <= m_tStart) {
        EndOfStream();
        return;
    }

    BOOL bDiscontinuity = TRUE;

    // if there is more than one sample at the allocator,
    // then try to queue 2 at once in order to overlap.
    // -- get buffer count and required alignment
    ALLOCATOR_PROPERTIES Actual;
    HRESULT hr = m_pAlloc->GetProperties(&Actual);

    // align the start position downwards
    REFERENCE_TIME tStart = AlignDown(m_tStart / UNITS, Actual.cbAlign) * UNITS;
    REFERENCE_TIME tCurrent = tStart;

    REFERENCE_TIME tStop = m_tStop;
    if(tStop > m_tDuration) {
        tStop = m_tDuration;
    }

    // align the stop position - may be past stop, but that
    // doesn't matter
    REFERENCE_TIME tAlignStop = AlignUp(tStop / UNITS, Actual.cbAlign) * UNITS;


    DWORD dwRequest;

    if(!m_bSync) {

        //  Break out of the loop either if we get to the end or we're asked
        //  to do something else
        while(tCurrent < tAlignStop) {

            // Break out without calling EndOfStream if we're asked to
            // do something different
            if(CheckRequest(&dwRequest)) {
                return;
            }

            // queue a first sample
            if(Actual.cBuffers > 1) {

                hr = QueueSample(tCurrent, tAlignStop, TRUE);
                bDiscontinuity = FALSE;

                if(FAILED(hr)) {
                    return;
                }
            }



            // loop queueing second and waiting for first..
            while(tCurrent < tAlignStop) {

                hr = QueueSample(tCurrent, tAlignStop, bDiscontinuity);
                bDiscontinuity = FALSE;

                if(FAILED(hr)) {
                    return;
                }

                hr = CollectAndDeliver(tStart, tStop);
                if(S_OK != hr) {

                    // stop if error, or if downstream filter said
                    // to stop.
                    return;
                }
            }

            if(Actual.cBuffers > 1) {
                hr = CollectAndDeliver(tStart, tStop);
                if(FAILED(hr)) {
                    return;
                }
            }
        }
    }
    else {

        // sync version of above loop
        while(tCurrent < tAlignStop) {

            // Break out without calling EndOfStream if we're asked to
            // do something different
            if(CheckRequest(&dwRequest)) {
                return;
            }

            IMediaSample* pSample;

            hr = m_pAlloc->GetBuffer(&pSample, NULL, NULL, 0);
            if(FAILED(hr)) {
                OnError(hr);
                return;
            }

            LONGLONG tStopThis = tCurrent + (pSample->GetSize() * UNITS);
            if(tStopThis > tAlignStop) {
                tStopThis = tAlignStop;
            }
            pSample->SetTime(&tCurrent, &tStopThis);
            tCurrent = tStopThis;

            if(bDiscontinuity) {
                pSample->SetDiscontinuity(TRUE);
                bDiscontinuity = FALSE;
            }

            hr = m_pReader->SyncReadAligned(pSample);

            if(FAILED(hr)) {
                pSample->Release();
                OnError(hr);
                return;
            }

            hr = DeliverSample(pSample, tStart, tStop);
            if(hr != S_OK) {
                if(FAILED(hr)) {
                    OnError(hr);
                }
                return;
            }
        }
    }

    EndOfStream();
}

// after a flush, cancelled i/o will be waiting for collection
// and release
void
CPullPin::CleanupCancelled(void) {
    while(1) {
        IMediaSample * pSample;
        DWORD_PTR dwUnused;

        HRESULT hr = m_pReader->WaitForNext(0,          // no wait
            &pSample,
            &dwUnused);
        if(pSample) {
            pSample->Release();
        }
        else {
            // no more samples
            return;
        }
    }
}