splitter.cpp

T264是中国的视频编码自由组织合力开发的264编解码程序

/*  -- Return a given pin -- */
CBasePin *CBaseSplitterFilter::GetPin(int iPin)
{
    CAutoLock lck(&m_csPins);
    if (iPin == 0 && m_pInput != NULL) {
        return m_pInput;
    }
    if (m_pInput != NULL) {
        iPin--;
    }
    if (iPin < 0 || iPin >= m_OutputPins.GetCount()) {
        return NULL;
    }
    POSITION pos = m_OutputPins.GetHeadPosition();
    while (iPin-- > 0) {
        m_OutputPins.GetNext(pos);
    }
    return m_OutputPins.GetNext(pos);
}

/*  -- Add an output pin -- */
BOOL CBaseSplitterFilter::AddOutputPin(CSplitterOutputPin *pOutputPin)
{
    CAutoLock lckFilter(&m_csFilter);
    CAutoLock lckPins(&m_csPins);
    ASSERT(m_State == State_Stopped);
    IncrementPinVersion();
    pOutputPin->AddRef();
    if (m_OutputPins.AddTail(pOutputPin) == NULL) {
        delete pOutputPin;
        return FALSE;
    }
    return TRUE;
}

/*  -- Destroy the input pin -- */
void CBaseSplitterFilter::DestroyInputPin()
{
    delete m_pInput;
    m_pInput = NULL;
}

/*  -- Delete the output pins -- */
void CBaseSplitterFilter::DestroyOutputPins()
{
    CAutoLock lckFilter(&m_csFilter);
    CAutoLock lckPins(&m_csPins);
    ASSERT(m_State == State_Stopped);
    for (;;) {
        CSplitterOutputPin *pPin = m_OutputPins.RemoveHead();
        if (pPin == NULL) {
            break;
        }
        IncrementPinVersion();

        //  Disconnect if necessary
        IPin *pPeer = pPin->GetConnected();
        if (pPeer != NULL) {
            pPeer->Disconnect();
            pPin->Disconnect();
        }
        pPin->Release();
    }
}

/*  Notify to create a new stream
    Called back through CParseNotify
*/
HRESULT CBaseSplitterFilter::CreateStream(
    LPCWSTR         pszName,
    CStreamNotify **ppStreamNotify
)
{
    HRESULT hr = S_OK;
    CSplitterOutputPin *pPin =
        new CSplitterOutputPin(this, &hr, pszName);
    if (FAILED(hr)) {
        delete pPin;
        return hr;
    }
    if (pPin == NULL) {
        return E_OUTOFMEMORY;
    }

    if (!AddOutputPin(pPin)) {
        delete pPin;
        return E_OUTOFMEMORY;
    }
    *ppStreamNotify = pPin->GetNotify();
    return S_OK;
}

HRESULT CBaseSplitterFilter::CompleteConnect(IPin *pReceivePin)
{
    m_pParser = CreateParser(&m_Notify, InputPin()->MediaType());
    if (m_pParser == NULL) {
        return E_OUTOFMEMORY;
    }
    /*  Create a reader to initialize the parser */
    CParseReaderFromAsync rdr(InputPin()->Reader());

    HRESULT hr = m_pParser->Init(&rdr);
    if (SUCCEEDED(hr)) {
        ALLOCATOR_PROPERTIES Props, Actual;

        /*  Our allocator doesn't accumulate alignment correctly
            so use the alignment currently set
        */
        Allocator()->GetProperties(&Props);
        m_pParser->GetSizeAndCount(&Props.cbBuffer, &Props.cBuffers);
        ((IMemAllocator *)Allocator())->SetProperties(&Props, &Actual);
    }
    return hr;
}

/*  Process some data */
HRESULT CBaseSplitterFilter::Receive(IMediaSample *pSample)
{
    /*  Notify our allocator about how much data we've got,
        Call the parser to eat the data,
        eat the data that was parsed
    */

    Allocator()->AddBuffer((CMediaSample *)pSample);

    /*  Now eat the data */
    LONG lValid;
    PBYTE pbValid = Allocator()->GetValid(&lValid);
    LONG lProcessed;
    HRESULT hr = m_pParser->Process(pbValid, lValid, &lProcessed);
    if (S_OK != hr) {
        /*  If something goes wrong do error processing */
        return hr;
    }

    Allocator()->Advance(lProcessed);

    /*  Flush the output queues */
    POSITION pos = m_OutputPins.GetHeadPosition();
    while (pos) {
        CSplitterOutputPin *pPin = m_OutputPins.GetNext(pos);
        pPin->SendAnyway();
    }
    return S_OK;
}

/*  Send EndOfStream */
void CBaseSplitterFilter::EndOfStream()
{
    ResetAllocatorAndParser();
    POSITION pos = m_OutputPins.GetHeadPosition();
    while (pos) {
        CSplitterOutputPin *pPin = m_OutputPins.GetNext(pos);
        if (pPin->IsConnected()) {
            pPin->DeliverEndOfStream();
        }
    }
}

/*  Send BeginFlush() */
HRESULT CBaseSplitterFilter::BeginFlush()
{
    POSITION pos = m_OutputPins.GetHeadPosition();

    while (pos) {
        CSplitterOutputPin *pPin = m_OutputPins.GetNext(pos);
        if (pPin->IsConnected()) {
            pPin->DeliverBeginFlush();
        }
    }
    return S_OK;
}

/*  Send EndFlush() */
HRESULT CBaseSplitterFilter::EndFlush()
{
    ResetAllocatorAndParser();
    POSITION pos = m_OutputPins.GetHeadPosition();
    while (pos) {
        CSplitterOutputPin *pPin = m_OutputPins.GetNext(pos);
        if (pPin->IsConnected()) {
            pPin->DeliverEndFlush();
        }
    }
    return S_OK;
}

/*  Break connection on the input pin */
void CBaseSplitterFilter::BreakConnect()
{
    delete m_pParser;
    m_pParser = NULL;
    DestroyOutputPins();
}

/*  -- Input pin methods -- */

/*  Constructor */
CSplitterInputPin::CSplitterInputPin(
    CBaseSplitterFilter *pFilter,
    HRESULT *phr
) : CBaseInputPin(NAME("CSplitterInputPin"),
             pFilter,
             &pFilter->m_csFilter,
             phr,
             L"Input"),
    m_puller(this)
{
}

/*  Connection stuff */
HRESULT CSplitterInputPin::CheckConnect(IPin *pPin)
{
    HRESULT hr = CBaseInputPin::CheckConnect(pPin);
    if (FAILED(hr)) {
        return hr;
    }

    /*  Create our allocator */
    ASSERT(m_pAllocator == NULL);

    m_pAllocator = new CSequentialAllocator(
                           NULL, // No owner - allocators are separate objects
                           &hr);
    if (m_pAllocator == NULL) {
        return E_OUTOFMEMORY;
    }
    if (FAILED(hr)) {
        return hr;
    }

    /*  The base classes expect the allocator to be AddRef'd */
    m_pAllocator->AddRef();

    // BUGBUG CPullPin might change the allocator and not tell us!
    hr = m_puller.Connect(pPin, m_pAllocator, TRUE);
    if (FAILED(hr)) {
        return hr;
    }
    return hr;
}

/*  Break connection - filter already locked by caller */
HRESULT CSplitterInputPin::BreakConnect()
{
    Filter()->BreakConnect();
    m_puller.Disconnect();
    if (m_pAllocator) {
        m_pAllocator->Release();
        m_pAllocator = NULL;
    }
    return CBaseInputPin::BreakConnect();
}

/*  End of stream */
STDMETHODIMP CSplitterInputPin::EndOfStream()
{
    CAutoLock lck(&Filter()->m_csStream);
    HRESULT hr = CheckStreaming();
    if (FAILED(hr)) {
        return hr;
    }

    /*  Forward to the output pins */
    Filter()->EndOfStream();
    return S_OK;
}

/*  Forward BeginFlush() and EndFlush() to filter */
STDMETHODIMP CSplitterInputPin::BeginFlush()
{
    CAutoLock lck(m_pLock);
    CBaseInputPin::BeginFlush();
    Filter()->BeginFlush();
    return S_OK;
}
STDMETHODIMP CSplitterInputPin::EndFlush()
{
    CAutoLock lck(&Filter()->m_csStream);
    CBaseInputPin::EndFlush();
    Filter()->EndFlush();
    return S_OK;
}

/*  Receive - forward to filter */
STDMETHODIMP CSplitterInputPin::Receive(IMediaSample *pSample)
{
    CAutoLock lck(&Filter()->m_csStream);
    HRESULT hr = CheckStreaming();
    if (FAILED(hr)) {
        return hr;
    }
    hr = Filter()->Receive(pSample);
    if (S_OK != hr) {
        NotifyError(hr);
    }
    return hr;
}

HRESULT CSplitterInputPin::Active()
{
    /*  If we're not seekable then just seek to start for now */
    REFERENCE_TIME rtDuration;
    m_puller.Duration(&rtDuration);
    m_puller.Seek(0, rtDuration);
    HRESULT hr = m_puller.Active();
    if (FAILED(hr)) {
        return hr;
    }
    return CBaseInputPin::Active();
}

HRESULT CSplitterInputPin::Inactive()
{
    m_puller.Inactive();
    return CBaseInputPin::Inactive();
}


/*  forward connection stuff to filter */
HRESULT CSplitterInputPin::CheckMediaType(const CMediaType *pmt)
{
    return Filter()->CheckInputType(pmt);
}
HRESULT CSplitterInputPin::CompleteConnect(IPin *pReceivePin)
{
    return Filter()->CompleteConnect(pReceivePin);
}