CJoin.cpp

Go to the documentation of this file.

#include "stdafx.h"

using AxPipe::Assert::OutputDebugStringF;

namespace AxPipe {
    void
    CJoin::CTSinkJoin::Out(CSeg *pSeg) {
        OutputDebugStringF(_T("CJoin::CTSinkJoin::Out()\n"));
        // Only send data if we're not empty - otherwise we may be locked up
        if (!m_fEmpty) {
            m_Sync.WorkStart();
            m_pNextSeg = pSeg;
            m_Sync.WorkSignal();
        } else {
            SetError(ERROR_CODE_GENERIC, _T("CJoin::CTSinkJoin::Out [Unexpected call]"));
            // If we're empty, just ignore extra data and report error.
            if (pSeg) {
                pSeg->Release();
            }
        }
    }

    bool
    CJoin::CTSinkJoin::OutClose() {
        OutputDebugString(_T("CJoin::CTSinkJoin::OutPlug()\n"));
        Out(NULL);
        return false;
    }

    bool
    CJoin::CTSinkJoin::OutFlush() {
        OutputDebugString(_T("CJoin::CTSinkJoin::OutFlush()\n"));
        Out(new CSeg);
        return false;
    }

    CJoin::CTSinkJoin::CTSinkJoin() {
        m_pNextSeg = NULL;
        m_fEmpty = false;
    }

    bool
    CJoin::CTSinkJoin::IsEmpty() {
        return m_fEmpty;
    }

    CSeg *
    CJoin::CTSinkJoin::GetSeg() {
        m_fEmpty = m_pNextSeg == NULL;
        return m_pNextSeg;
    }

    void
    CJoin::CTSinkJoin::SinkWorkWait() {
        m_Sync.WorkWait();
    }

    void
    CJoin::CTSinkJoin::SinkWorkEnd() {
        m_Sync.WorkEnd();
    }

    CJoin::CJoin() {
        m_nMaxStreams = 0;
        m_ppInSinks = NULL;
    }

    CJoin::~CJoin() {
        OutputDebugString(_T("CJoin::~CJoin()\n"));
        ASSCHK(m_fExit, _T("CJoin::~CJoin() with worker still active"));
        if (m_ppInSinks) {
            for (int i = 0; i < m_nMaxStreams; i++) {
                delete m_ppInSinks[i];
                m_ppInSinks[i] = NULL;
            }
            delete[] m_ppInSinks;
            m_ppInSinks = NULL;
        }
    }

    void
    CJoin::OutPlug() {
        if (GetErrorCode() == ERROR_CODE_SUCCESS) {
            for (int i = 0; i < m_nMaxStreams; i++) {
                if (m_ppInSinks[i] && m_ppInSinks[i]->GetErrorCode() != ERROR_CODE_SUCCESS) {
                    SetError(m_ppInSinks[i]->GetErrorCode(), m_ppInSinks[i]->GetErrorMsg());
                    return;
                }
            }
        }
    }
    
    CJoin *
    CJoin::Init(int nMaxStreams) {
        m_ppInSinks = new CTSinkJoin *[m_nMaxStreams = nMaxStreams];
        for (int i = 0; i < m_nMaxStreams; i++) {
            m_ppInSinks[i] = NULL;
        }
        return this;
    }

    CSink &
    CJoin::GetSink(int ix) {
        ASSCHK(ix < m_nMaxStreams, _T("CJoin::GetSink() [Invalid stream index]"));
        if (!m_ppInSinks[ix]) {
            return *(m_ppInSinks[ix] = new CTSinkJoin);
        }
        return *m_ppInSinks[ix];
    }

    CSeg *
    CJoin::StreamSeg(int ix) {
        ASSCHK(ix > m_nMaxStreams || m_ppInSinks[ix] != NULL, _T("CJoin::StreamSeg() [Invalid stream index]"));
        if (m_ppInSinks[ix]->IsEmpty()) {
            return NULL;
        }
        m_ppInSinks[ix]->SinkWorkWait();
        CSeg *pSeg = m_ppInSinks[ix]->GetSeg();
        m_ppInSinks[ix]->SinkWorkEnd();
        return pSeg;
    }

    int
    CJoin::StreamIx(int ix) {
        return ix % m_nMaxStreams;
    }
    
    int
    CJoin::StreamNum() {
        return m_nMaxStreams;
    }
    
    bool
    CJoin::StreamEmpty(int ix) {
        return m_ppInSinks[ix]->IsEmpty();
    }
};

---