CPipe.cpp

Go to the documentation of this file.

#include "stdafx.h"

namespace AxPipe {
    using Assert::OutputDebugStringF;

    void
    CPipe::DestructSink() {
        OutputDebugStringF(_T("CPipe::DestructSink() m_fAutoDeleteSink=%d, m_pSink=%p\n"), m_fAutoDeleteSink, m_pSink);
        if (m_pSink) {
            m_pSink->WaitForIdle();
            ASSCHK(m_pSink->m_fExit, _T("CPipe::DestructSink() without proper Plug()"));
            if (m_fAutoDeleteSink) {
                delete m_pSink;
            } else {
                m_pSink->DestructSink();
            }
            m_pSink = NULL;
        }
    }

    void
    CPipe::Work() {
        CSeg *pSeg = m_pSeg;
        m_pSeg = NULL;

        if (DoSegWork(pSeg->AddRef()) && m_pSink) {
            // Propagate
            m_pSink->OutPump(pSeg);
            // If we're about to exit, wait for the the sink to exit too.
            // That way, we know that downstream has all exited before we
            // do at the return of this call.
            if (m_fExit) {
                m_pSink->WorkExitWait();
            }
        } else {
            pSeg->Release();
        }
    }

    void
    CPipe::AppendSink(CSink *pSink, bool fAutoDeleteSink) {
        if (m_pSink) {
            m_pSink->AppendSink(pSink, fAutoDeleteSink);
        } else {
            m_fAutoDeleteSink = fAutoDeleteSink;
            (m_pSink = pSink)->CError::Init(this);
        }
    }
    CPipe *
    CPipe::Append(CSink *pSink) {
        AppendSink(pSink, true);
        return this;
    }

    CPipe *
    CPipe::Append(CSink& sink) {
        AppendSink(&sink, false);
        return this;
    }
 
    void
    CPipe::Sync() {
        WaitForIdle();
        m_pSink->Sync();                    // Wait for the rest of the pipe too.
    }

    void
    CPipe::Open() {
        if (m_pSink) {
            m_pSink->OutPump((new CSeg)->SetType(eSegTypeOpen));
        }
    }

    void
    CPipe::Pump(CSeg *pSeg) {
        if (m_pSink) {
            m_pSink->OutPump(pSeg);
        } else {
            pSeg->Release();
        }
    }

    void
    CPipe::Flush() {
        if (m_pSink) {
            m_pSink->OutPump((new CSeg)->SetType(eSegTypeFlush));
        }
    }

    void
    CPipe::Close() {
        if (m_pSink) {
            m_pSink->OutPump((new CSeg)->SetType(eSegTypeClose));
        }
    }

    // See the CSink comments for more details, for a non CSink object.
    // \see CSink
    CSeg *
    CPipe::GetSeg(size_t cb)  {
        return m_pSink ? m_pSink->OutGetSeg(cb) : new CSeg(cb);
    }

    void
    CPipe::Signal(void *vId, void *p) {
        if (m_pSink) {
            m_pSink->WaitForIdle();         // Ensure sync with next
            if (m_pSink->OutSignal(vId, p)) {
                m_pSink->Signal(vId, p);
            }
        }
    }

    longlong
    CPipe::OutSizeMax() {
        return m_pSink ? m_pSink->OutSizeMax() : -1;
    }
    
    CSeg *
    CPipe::OutGetSeg(size_t cb) {
        return new CSeg(cb);
    }
    
    bool
    CPipe::OutSignal(void *vId, void *p) {
        return true;
    }
    
    bool
    CPipe::OutOpen() {
        return true;
    };
    
    bool
    CPipe::OutClose() {
        return true;
    };
    
    void
    CPipe::OutSpecial(CSeg *pSeg) {
        Pump(pSeg);
    }

    CPipe::CPipe() {
        m_pSink = NULL;
    }

    CPipe::~CPipe() {
        DestructSink();                     // Destruct sinks, if any.
    }
};

---