CFilter.cpp

Go to the documentation of this file.

#include "stdafx.h"

namespace AxPipe {
    void
    CFilter::CoFilter(void *pvThis) {
        ((CFilter *)pvThis)->CoStartFilter(pvThis);
        // Should never get here!
        ASSCHK(false, _T("CFilter::CoFilter unexpected fiber exit"));
    }

    CFilter::CFilter() {
        m_ctxFilter.Init(this, CoFilter, this);
        m_ctxWork.Init(this, NULL, NULL);   // The Work context starts as the 'current'
        m_fFirstWork = true;                // Ensure that we know when Work() is called first time.
    }

    CFilter::~CFilter() {
        m_ctxFilter.Stop();
    }

    void
    CFilter::Out(CSeg *pSeg) {
        if (!m_fIsOpen) {
            SetError(ERROR_CODE_NOTOPEN, ERROR_MSG_NOTOPEN);
            return;
        }

        m_pSeg = pSeg;
        // Switch to Read() and InFilter()
        //OutputDebugString(_T("CFilter::Out(CSeg *pSeg) m_ctxFilter.Go()\n"));
        m_ctxFilter.Go();
    }

    bool
    CFilter::OutOpen() {
        OutputDebugString(_T("CFilter::OutOpen()\n"));
        if (m_fFirstWork) {
            // If first time, ensure we are executing in Work coroutine context now
            OutputDebugString(_T("CFilter::OutOpen() m_ctxWork.Go()\n"));
            m_ctxWork.Go();                 // This only initializes the context...
            m_fFirstWork = false;           // ...we get here immediately
        }
        // Default for filters is to not propagate
        return false;
    }
    
    bool
    CFilter::OutClose() {
        Out(NULL);
        // Default for filters is to not propagate
        return false;
    }
    
    bool
    CFilter::OutFlush() {
        Out(new CSeg);
        return true;
    }

    void
    CFilter::Work() {
        CPipe::Work();
    }

    void
    CFilter::CoStartFilter(void *pvThis) {
        // The filter may be called multiple times. It should exit when receiving
        // a eSegTypeClose segment, ready to be called again when more arrives.
        do {
            OutputDebugStringF(_T("CFilter::CoStartFilter(void *pvThis) InFilter(), this=%p\n"), this);
            InFilter();                     // Shold return when eof/empty is signalled.

            // Drive the sender until we either get a valid data segment, or we're killed
            do {
                OutputDebugStringF(_T("CFilter::CoStartFilter waiting for data, this=%p\n"), this);
                m_ctxWork.Go();
            } while (m_pSeg == NULL || m_pSeg->Len() == 0);
            OutputDebugStringF(_T("CFilter::CoStartFilter found data, this=%p\n"), this);
        } while (true);
        // Never get here!
    }

    CSeg *
    CFilter::Read() {
        // We may already have a segment waiting, at first call.
        if (!m_pSeg) {
            m_ctxWork.Go();
        }
        // m_pSeg can be valid, zero-length or NULL here. Nothing else.
        CSeg *pSeg = m_pSeg;
        m_pSeg = NULL;
        return pSeg;
    }
    bool
    CFilterByte::GetNextSeg() {
        // Release if empty.
        if (m_pSeg && !m_pSeg->Len()) {
            m_pSeg->Release();
            m_pSeg = NULL;
        }
        if (!m_pSeg) {
            m_ctxWork.Go();
        }
        return m_pSeg != NULL;              // Success as long as we get a segment, but it might be zero-len
    }
    int
    CFilterByte::ReadByte() {
        do {
            if (!GetNextSeg()) {
                return -1;
            }
        } while (!m_pSeg->Len());       // Ignore flush requests, just wait for data.

        // Now we now we have at least one byte.
        unsigned char c = *m_pSeg->PtrRd();
        m_pSeg->Drop(1);
        return c;
    }

    CSeg *
    CFilterByte::Read() {
        SetError(ERROR_CODE_GENERIC, ERROR_MSG_GENERIC, _T("Attempt to call CFilterByte::Read()"));
        return NULL;
    }

    size_t
    CFilterByte::Skip(size_t cb) {
        while (cb) {
            if (!GetNextSeg()) {
                break;
            }
            size_t cbChunk = m_pSeg->Len();
            if (cbChunk > cb) {
                cbChunk = cb;
            }
            m_pSeg->Drop(cbChunk);
            cb -= cbChunk;
        }
        return cb;
    }

    CSeg *
    CFilterBlock::ReadBlock(size_t cb) {
        // Zero means take what we get
        if (!cb) {
            while (GetNextSeg()) {
                if (m_pSeg->Len()) {
                    CSeg *pSeg = m_pSeg;
                    m_pSeg = NULL;
                    return pSeg;
                }
                m_pSeg->Release();
                m_pSeg = NULL;
            }
            return NULL;                    // No data - this is shown with NULL
        }
        // If no buffered data - we must get more in any case, so let's do it.
        if (!GetNextSeg()) {
            return NULL;                    // No data to get.
        }
        // This is a slight optimization to try to keep chunks in the original
        // segment as much as possible.
        // If the buffer contains the proper number of bytes already, clone it,
        // drop the bytes off the original, and set the length of the copy
        // returned.
        if (m_pSeg->Len() >= cb) {
            CSeg *pSeg = m_pSeg->Clone();
            m_pSeg->Drop(cb);
            pSeg->Len(cb);
            return pSeg;
        }

        // Now we know we must merge two or more buffers. Let's allocate a segment to return.
        CSeg *pSeg = new CSeg(cb);
        if (!pSeg) {
            SetError(ERROR_CODE_GENERIC, ERROR_MSG_GENERIC, _T("Out of memory"));
            return NULL;
        }

        size_t cbSeg;
        pSeg->Len(cbSeg = 0);           // Set the length of valid data in the segment to zero.
        // We also know at this point that we have valid data in the buffer.
        do {
            size_t cbChunk = m_pSeg->Len();
            if (cbChunk > cb - cbSeg) {
                cbChunk = cb - cbSeg;
            }
            memcpy(&pSeg->PtrWr()[cbSeg], m_pSeg->PtrRd(), cbChunk);
            m_pSeg->Drop(cbChunk);
            pSeg->Len(cbSeg += cbChunk);// Update the length of valid data in the segment.

            // If we've gotten all we need...
            if (cbSeg == cb) {
                return pSeg;
            }

            if (!GetNextSeg()) {
                return pSeg;        // Return what data we have.
            }
        } while (true);
        // Can't get here!
    }
};

---