pstreams.h

PTypes (C++ Portable Types Library) is a simple alternative to the STL that includes multithreading

    outstm* get_stm()  { return stm; }
    void set_stm(outstm* stm);
};


//
// inmemory - memory stream
//

class ptpublic inmemory: public instm 
{
protected:
    string mem;
    virtual void bufalloc();
    virtual void buffree();
    virtual void bufvalidate();
    virtual void doopen();
    virtual void doclose();
    virtual int dorawread(char* buf, int count);

public:
    inmemory(const string& imem);
    virtual ~inmemory();
    virtual int classid();
    virtual string get_streamname();
};


//
// outmemory - memory stream
//

class ptpublic outmemory: public outstm 
{
protected:
    string mem;
    int limit;
    int increment;
    virtual void doopen();
    virtual void doclose();
    virtual int dorawwrite(const char* buf, int count);

public:
    outmemory(int limit = -1, int increment = 1024);
    virtual ~outmemory();
    virtual int classid();
    virtual string get_streamname();
    const char* get_data()   { return pconst(mem); }
    int tell()               { return abspos; }
    string get_strdata();
};


// -------------------------------------------------------------------- //
// ---  file input/output --------------------------------------------- //
// -------------------------------------------------------------------- //


//
// infile - file input
//

class outfile;

class ptpublic infile: public instm
{
protected:
    string filename;
    int    syshandle;   // if not -1, assigned to handle in open() instead of opening a file by a name
    int    peerhandle;  // pipe peer handle, needed for closing the peer after fork() on unix

    virtual void doopen();
    virtual void doclose();

public:
    infile();
    infile(const char* ifn);
    infile(const string& ifn);
    virtual ~infile();
    virtual int classid();

    void pipe(outfile&);
    virtual string get_streamname();
    int get_syshandle()                     { return syshandle; }
    void set_syshandle(int ihandle)         { close(); syshandle = ihandle; }
    int get_peerhandle()                    { return peerhandle; }
    string get_filename()                   { return filename; }
    void set_filename(const string& ifn)    { close(); filename = ifn; }
    void set_filename(const char* ifn)      { close(); filename = ifn; }
};


//
// outfile - file output
//

class ptpublic outfile: public outstm
{
protected:
    friend class infile; // infile::pipe() needs access to peerhandle

    string filename;
    int    syshandle;   // if not -1, assigned to handle in open() instead of opening a file by a name
    int    peerhandle;  // pipe peer handle, needed for closing the peer after fork() on unix
    int    umode;       // unix file mode (unix only), default = 644
    bool   append;      // append (create new if needed), default = false

    virtual void doopen();
    virtual void doclose();

public:
    outfile();
    outfile(const char* ifn, bool iappend = false);
    outfile(const string& ifn, bool iappend = false);
    virtual ~outfile();
    virtual int classid();

    virtual void flush();
    virtual string get_streamname();

    int get_syshandle()                     { return syshandle; }
    void set_syshandle(int ihandle)         { close(); syshandle = ihandle; }
    int get_peerhandle()                    { return peerhandle; }
    string get_filename()                   { return filename; }
    void set_filename(const string& ifn)    { close(); filename = ifn; }
    void set_filename(const char* ifn)      { close(); filename = ifn; }
    bool get_append()                       { return append; }
    void set_append(bool iappend)           { close(); append = iappend; }
    int  get_umode()                        { return umode; }
    void set_umode(int iumode)              { close(); umode = iumode; }
};


//
// intee - UNIX tee-style utility class
//

class ptpublic intee: public infilter {
protected:
    outfile file;
    virtual void doopen();
    virtual void doclose();
    virtual void dofilter();
public:
    intee(instm* istm, const char* ifn, bool iappend = false);
    intee(instm* istm, const string& ifn, bool iappend = false);
    virtual ~intee();

    outfile* get_file()   { return &file; }
    virtual string get_streamname();
};


// -------------------------------------------------------------------- //
// ---  named pipes --------------------------------------------------- //
// -------------------------------------------------------------------- //


// on Unix this directory can be overridden by providing the
// full path, e.g. '/var/run/mypipe'. the path is ignored on 
// Windows and is always replaced with '\\<server>\pipe\'

#ifndef WIN32
#  define DEF_NAMED_PIPES_DIR "/tmp/"
#endif


#ifdef WIN32

const int DEF_PIPE_TIMEOUT = 20000;         // in milliseconds, for reading and writing
const int DEF_PIPE_OPEN_TIMEOUT = 1000;     // for connecting to the remote pipe:
const int DEF_PIPE_OPEN_RETRY = 5;          //    will double the timeout value for each retry,
                                            //    i.e. 1 second, then 2, then 4 etc.
const int DEF_PIPE_SYSTEM_BUF_SIZE = 4096;

#endif


class ptpublic namedpipe: public fdxstm
{
    friend class npserver;

protected:
    string pipename;
    int    svhandle;

#ifdef WIN32
    // we use overlapped IO in order to have timed waiting in serve()
    // and also to implement timeout error on the client side
    OVERLAPPED ovr;
    virtual int dorawread(char* buf, int count);
    virtual int dorawwrite(const char* buf, int count);
    static string namedpipe::realpipename(const string& pipename, const string& svrname = nullstring);
    void initovr();
#else
    static string namedpipe::realpipename(const string& pipename);
    static bool setupsockaddr(const string& pipename, void* sa);
    void initovr()  {}
#endif

    virtual void doopen();
    virtual void doclose();
    virtual int  doseek(int, ioseekmode);

public:
    namedpipe();
    namedpipe(const string& ipipename);
#ifdef WIN32
    namedpipe(const string& ipipename, const string& servername);
#endif
    virtual ~namedpipe();

    virtual void flush();
    virtual string get_streamname();

    string get_pipename()   { return pipename; }
    void set_pipename(const string&);
    void set_pipename(const char*);
};


class ptpublic npserver: public unknown
{
    string pipename;
    int    handle;
    bool   active;

    void error(int code, const char* defmsg);
    void open();
    void close();
#ifdef WIN32
    void openinst();
    void closeinst();
#endif

public:
    npserver(const string& ipipename);
    ~npserver();

    bool serve(namedpipe& client, int timeout = -1);
};


// -------------------------------------------------------------------- //
// ---  utility streams ----------------------------------------------- //
// -------------------------------------------------------------------- //

//
// MD5 -- message digest algorithm
// Derived from L. Peter Deutsch's work, please see src/pmd5.cxx
//


const int md5_digsize = 16;
typedef uchar md5_digest[md5_digsize];

// from md5.h

typedef unsigned char md5_byte_t; /* 8-bit byte */
typedef unsigned int md5_word_t; /* 32-bit word */


typedef struct md5_state_s {
    md5_word_t count[2];	/* message length in bits, lsw first */
    md5_word_t abcd[4];		/* digest buffer */
    md5_byte_t buf[64];		/* accumulate block */
} md5_state_t;


class ptpublic outmd5: public outfilter {
protected:
    md5_state_s ctx;
    md5_digest digest;

    virtual void doopen();
    virtual void doclose();
    virtual int dorawwrite(const char* buf, int count);

public:
    outmd5(outstm* istm = nil);
    virtual ~outmd5();
    
    virtual string get_streamname();

    const unsigned char* get_bindigest()  { close(); return digest; }
    string get_digest();
};


//
// null i/o stream
//


class ptpublic outnull: public outstm
{
protected:
    virtual int  dorawwrite(const char*, int);
    virtual void doopen();
    virtual void doclose();
public:
    outnull();
    virtual ~outnull();
    virtual string get_streamname();
};


// -------------------------------------------------------------------- //
// ---  unit ---------------------------------------------------------- //
// -------------------------------------------------------------------- //


#ifdef _MSC_VER
// disable "type name first seen using 'struct' now seen using 'class'" warning
#  pragma warning (disable: 4099)
// disable "class '...' needs to have dll-interface to be used by clients of class 
// '...'" warning, since the compiler may sometimes give this warning incorrectly.
#  pragma warning (disable: 4251)
#endif


class ptpublic unit: public component
{
protected:
    friend class unit_thread;

    unit*         pipe_next;    // next unit in the pipe chain, assigned by connect()
    unit_thread*  main_thread;  // async execution thread, started by run() if necessary
    int           running;      // running status, to protect from recursive calls to run() and waitfor()

    void do_main();

public:
    compref<instm> uin;
    compref<outstm> uout;

    unit();
    virtual ~unit();
    virtual int classid();

    // things that may be overridden in descendant classes
    virtual void main();        // main code, called from run()
    virtual void cleanup();     // main code cleanup, called from run()

    // service methods
    void connect(unit* next);
    void run(bool async = false);
    void waitfor();
};
typedef unit* punit;


typedef unit CUnit;         // send me a $10 check if you use this alias (not obligatory though,
                            // because the library is free, after all)


//
// standard input, output and error devices
//

ptpublic extern infile  pin;
ptpublic extern outfile pout;
ptpublic extern outfile perr;
ptpublic extern outnull pnull;


#ifdef _MSC_VER
#pragma pack(pop)
#endif


PTYPES_END

#endif // __PSTREAMS_H__