AxPipe.h

Go to the documentation of this file.

#pragma once
#ifndef AXPIPE_H
#define AXPIPE_H

// Common base classes
#include "CError.h"
#include "CSeg.h"
#include "CSync.h"
#include "CThread.h"
#include "CCoContext.h"
#include "AxPipeAssert.h"

//
//  Use the Windows convention for Unicode/Ansi switcheable code, but if nothing
//  is defined, we default to Ansi here. Change as appropriate if necessary.
//
#ifndef _T
#define _T(s) s                             
#define _TCHAR char                         
#endif

namespace AxPipe {
    namespace Stock {
        // Dummy, just to get the description of the namespace documented.
    }

class CSeg;
extern DWORD dwTlsIndex;
extern volatile long nGlobalInit;
//
//  The error class is a virtual base class to all the others,
//  and is used to record errors. Normally, errors are not passed
//  by return value, nor are any exceptions thrown.
//
//  Use SetError() to indicate an error, and GetError(), to check for one.

static const _TCHAR *ERROR_MSG_GENERIC = _T("AxPipe:: %s");
static const _TCHAR *ERROR_MSG_INTERNAL = _T("AxPipe:: Internal error %s");
static const _TCHAR *ERROR_MSG_NOTOPEN = _T("AxPipe:: Pipe not Open");

enum ERROR_CODE {
    ERROR_CODE_SUCCESS = 0,                 
    ERROR_CODE_GENERIC,                     
    ERROR_CODE_INTERNAL,                    
    ERROR_CODE_NOTOPEN,                     
    ERROR_CODE_STOCK,                       
    ERROR_CODE_DERIVED = 100,               
};

// Modify according to compiler version. Include appropriate #ifdef's.
typedef __int64 longlong;                   
typedef unsigned __int64 ulonglong;         

enum eSegType {
    eSegTypeOpen = 1,                       
    eSegTypeFlush,                          
    eSegTypeClose,                          
    eSegTypePlug,                           
    eSegTypeDerived = 100,                  
};

class CGlobalInit {
public:
    CGlobalInit() {
        if (InterlockedIncrement(&nGlobalInit) == 1) {
            dwTlsIndex = TlsAlloc();
        }
    }
    ~CGlobalInit() {
        if (InterlockedDecrement(&nGlobalInit) == 0 && dwTlsIndex != TLS_OUT_OF_INDEXES) {
            TlsFree(dwTlsIndex);
            dwTlsIndex = TLS_OUT_OF_INDEXES;
        }
    }
};

class CSink : public CNoThread, public CError {
    friend class CPipe;                     
protected:
    bool m_fIsOpen;                         
    CSeg *m_pSeg;                           

private:
    bool DoSegWork(CSeg *pSeg);             

protected:
    void Work();                            
    CSeg *GetSeg(size_t cb);                
    virtual void Signal(void *vId, void *p);
    longlong SizeMax();                     

public:
    virtual void OutPump(CSeg *pSeg);       

protected:
    virtual longlong OutSizeMax();          
    virtual CSeg *OutGetSeg(size_t cb);     
    virtual bool OutSignal(void *vId, void *p); 
    virtual bool OutOpen();                 
    virtual bool OutFlush();                
    virtual bool OutClose();                
    virtual void OutPlug();                 
    virtual void OutSpecial(CSeg *pSeg);    

    virtual void Out(CSeg *pSeg) = 0;

public:
    CSink();                                
    virtual ~CSink();                       
    virtual void AppendSink(CSink *pSink, bool fAutoDelete); 
    virtual void DestructSink();            
    virtual void Sync();                    
};
class CPipe : public CSink {
    friend class CSplit;                    
    void DestructSink();                    

protected:
    CSink *m_pSink;                         

    void Work();                            
    void AppendSink(CSink *pSink, bool fAutoDelete); 
    CSeg *GetSeg(size_t cb);                
    void Signal(void *vId, void *p);        
    longlong OutSizeMax();                  
    CSeg *OutGetSeg(size_t cb);             
    bool OutSignal(void *vId, void *p);     
    bool OutOpen();                         
    bool OutClose();                        
    void OutSpecial(CSeg *pSeg);            

    void Out(CSeg *pSeg) = 0;

public:
    CPipe();                                
    ~CPipe();                               
    CPipe *Append(CSink *pSink);            
    CPipe *Append(CSink& sink);             
    void Sync();                            
    void Open();                            
    void Pump(CSeg *pSeg);                  
    void Flush();                           
    void Close();                           
};

class CSinkNull : public CSink {
public:
    inline void Out(CSeg *pSeg) {
        pSeg->Release();
    }
};

class CSplit : public CPipe {
    CPipe *m_pLeft;                         
    CPipe *m_pRight;                        
    
    void DestructSink();                    
    void PumpSplit(CSeg *pSeg);             
public:
    CSplit();                               
    void AppendSink(CSink *pSink, bool fAutoDelete); 
    void Sync();                            
    CSplit *Init(CPipe *pLeft, CPipe *pRight); 
    void Out(CSeg *pSeg);                   
    void OutSpecial(CSeg *pSeg);            
    bool OutFlush();                        
    bool OutClose();                        
    bool OutOpen();                         
};

class CPipeBlock : public CPipe {
    CSeg *m_pBlockPart;                     
    size_t m_cbBlockSize;                   

public:
    CPipeBlock();                           
    virtual ~CPipeBlock();                  
    CPipeBlock *Init(size_t cbBlockSize);   
    void OutPump(CSeg *pSeg);               
    CSeg *PartialBlock();                   
};

class CSource : public CPipe {
public:
    virtual ~CSource();                     
    CSource *Append(CSink *pSink);          
    CSource *Append(CSink& sink);           
    CSource *Open();                        
    CSource *Close();                       
    CSource *Drain();                       
    CSource *Plug();                        
    void Out(CSeg *pSeg);                   

protected:
    virtual CSeg *In() = 0;
};

class CSourceNull : public CSource {
protected:
    CSeg *In() {
        return new CSeg;
    }
};

class CSourceMem : public CSource {
    CSeg *m_pSegSave;                       
protected:
    CSeg *In() {
        if (m_pSegSave) {
            m_pSeg = m_pSegSave;
            m_pSegSave = NULL;
            return m_pSeg;
        } else {
            return new CSeg;
        }
    }

public:
    CSourceMem() {
        m_pSegSave = NULL;
    }
    CSourceMem *Init(size_t cb, void *p) {
        m_pSegSave = new CSeg(cb, p);
        return this;
    }
};
class CFilter : public CPipe {
private:
    bool m_fFirstWork;                      
    CCoContext m_ctxFilter;                 

    static void CoFilter(void *pvThis);     
    void CoStartFilter(void *pvThis);       
    void Out(CSeg *pSeg);                   

public:
    CFilter();                              
    ~CFilter();

protected:
    CCoContext m_ctxWork;                   

    bool OutOpen();                         
    bool OutClose();                        
    bool OutFlush();                        
    void Work();                            
    CSeg *Read();                           

    virtual void InFilter() = 0;
};

class CFilterByte : public CFilter {
protected:
    bool GetNextSeg();                      
protected:
    int ReadByte();                         
    CSeg *Read();                           
    size_t Skip(size_t cb);                 
};

class CFilterBlock : public CFilterByte {
protected:
    CSeg *ReadBlock(size_t cb);             
};
class CJoin : public CSource {
    class CTSinkJoin : public CThread<CSink> {
        CThreadSync m_Sync;                 
        CSeg *m_pNextSeg;                   
        bool m_fEmpty;                      
    protected:
        void Out(CSeg *pSeg);               
        bool OutClose();                    
        bool OutFlush();                    
    public:
        CTSinkJoin();                       
        CSeg *GetSeg();                     
        bool IsEmpty();                     
        void SinkWorkWait();                
        void SinkWorkEnd();                 
    };

    CTSinkJoin **m_ppInSinks;               
    int m_nMaxStreams;                      

public:
    CJoin();                                
    virtual ~CJoin();                       
    void OutPlug();
    CSink &GetSink(int ix);
    CJoin *Init(int nMaxStreams = 2);       

    //
    // Utility routines for In()
    //
    CSeg *StreamSeg(int ix);                
    int StreamIx(int ix);                   
    int StreamNum();                        
    bool StreamEmpty(int ix);               
};

} // namespace AxPipe
#endif

---