pstream.h

Go to the documentation of this file.

/* $Id: pstream.h,v 1.52 2002/11/08 02:51:51 redi Exp $
PStreams - POSIX Process I/O for C++
Copyright (C) 2001,2002 Jonathan Wakely

This file is part of PStreams.

PStreams is free software; you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as
published by the Free Software Foundation; either version 2.1 of
the License, or (at your option) any later version.

PStreams is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License
along with PStreams; if not, write to the Free Software Foundation, Inc.,
59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#ifndef REDI_PSTREAM_H
#define REDI_PSTREAM_H

#include <ios>
#include <streambuf>
#include <istream>
#include <ostream>
#include <string>
#include <vector>
#include <algorithm>
#include <cstring>
#include <cerrno>
#include <sys/types.h>  // for pid_t
#include <sys/wait.h>   // for waitpid()
#include <unistd.h>     // for pipe() fork() exec() and filedes functions
#include <signal.h>     // for kill()


// TODO   add input buffering to pstreambuf

// TODO   abstract process creation and control to a process class.

// TODO   capitalise class names ?
// basic_pstreambuf -> BasicPStreamBuf
// basic_opstream   -> BasicOPStream
// basic_ipstream   -> BasicIPStream
// basic_pstream    -> BasicPStream
// basic_rpstream   -> BasicRPStream



#define PSTREAMS_VERSION 0x002a   // 0.42


namespace redi
{
  struct pstreams
  {
    typedef std::ios_base::openmode           pmode;

    static const pmode pstdin  = std::ios_base::out; 
    static const pmode pstdout = std::ios_base::in;  
    static const pmode pstderr = std::ios_base::app; 

  protected:
    static const size_t bufsz = 32;
    static const size_t pbsz  = 2;
  };

  template <typename CharT, typename Traits>
    class basic_pstreambuf
    : public std::basic_streambuf<CharT, Traits>
    , public pstreams
    {
    public:
      // Type definitions for dependent types
      typedef CharT                             char_type;
      typedef Traits                            traits_type;
      typedef typename traits_type::int_type    int_type;
      typedef typename traits_type::off_type    off_type;
      typedef typename traits_type::pos_type    pos_type;
      typedef int                               fd_t;

      basic_pstreambuf();

      basic_pstreambuf(const std::string& command, pmode mode);

      basic_pstreambuf(const std::string& file, const std::vector<std::string>& argv, pmode mode);

      ~basic_pstreambuf();

      basic_pstreambuf*
      open(const std::string& command, pmode mode);

      basic_pstreambuf*
      open(const std::string& file, const std::vector<std::string>& argv, pmode mode);

      basic_pstreambuf*
      close();

      basic_pstreambuf*
      kill(int signal = SIGTERM);

      void
      peof();

      bool
      read_err(bool readerr = true);
      
      bool
      is_open() const;

      bool
      exited();

#if REDI_EVISCERATE_PSTREAMS

      size_t
      fopen(FILE*& in, FILE*& out, FILE*& err);
#endif

      int
      status() const;

      int
      error() const;

    protected:
      int_type
      overflow(int_type c);

      int_type
      underflow();

      int_type
      pbackfail(int_type c = traits_type::eof());

      int
      sync();

      std::streamsize
      xsputn(const char_type* s, std::streamsize n);

      bool
      write(char_type c);

      bool
      read(char_type& c);

      std::streamsize
      write(char_type* s, std::streamsize n);

      std::streamsize
      read(char_type* s, std::streamsize n);

    protected:
      enum buf_read_src { rsrc_out = 0, rsrc_err = 1 };

      pid_t
      fork(pmode mode);

      int
      wait(bool nohang = false);

      fd_t&
      wpipe();

      fd_t&
      rpipe();

      fd_t&
      rpipe(buf_read_src which);

      void
      create_buffers(pmode mode);

      void
      destroy_buffers(pmode mode);

      bool
      empty_buffer();

      bool
      fill_buffer();

      char_type*
      rbuffer();

      buf_read_src
      switch_read_buffer(buf_read_src);

    private:
      basic_pstreambuf(const basic_pstreambuf&);
      basic_pstreambuf& operator=(const basic_pstreambuf&);

      void
      init_rbuffers();

      pid_t         ppid_;        // pid of process
      fd_t          wpipe_;       // pipe used to write to process' stdin
      fd_t          rpipe_[2];    // two pipes to read from, stdout and stderr
      char_type*                wbuffer_;
      char_type*                rbuffer_[2];
      char_type*                rbufstate_[3];
      buf_read_src  rsrc_;
      int           status_;      // hold exit status of child process
      int           error_;       // hold errno if fork() or exec() fails
    };

  template <typename CharT, typename Traits = std::char_traits<CharT> >
    class pstream_common
    : virtual public std::basic_ios<CharT, Traits>
    , public pstreams
    {
    protected:
      typedef basic_pstreambuf<CharT, Traits>       streambuf_type;

    public:
      typedef typename streambuf_type::pmode        pmode;

      pstream_common();

      pstream_common(const std::string& command, pmode mode);

      pstream_common(const std::string& file, const std::vector<std::string>& argv, pmode mode);

      virtual void
      open(const std::string& command, pmode mode);

      virtual void
      open(const std::string& file, const std::vector<std::string>& argv, pmode mode);

      void
      close();

      bool
      is_open() const;

      const std::string&
      command() const;

      streambuf_type*
      rdbuf() const;

#if REDI_EVISCERATE_PSTREAMS

      size_t
      fopen(FILE*& in, FILE*& out, FILE*& err);
#endif

    protected:
      virtual
      ~pstream_common() = 0;

      std::string       command_; 
      streambuf_type    buf_;     
    };


  template <typename CharT, typename Traits = std::char_traits<CharT> >
    class basic_ipstream
    : public std::basic_istream<CharT, Traits>
    , public pstream_common<CharT, Traits>
    {
      typedef std::basic_istream<CharT, Traits>     istream_type;
      typedef pstream_common<CharT, Traits>         pbase_type;
      typedef typename pbase_type::streambuf_type   streambuf_type;

    public:
      typedef typename pbase_type::pmode            pmode;

      basic_ipstream()
      : istream_type(NULL), pbase_type()
      {}

      basic_ipstream(const std::string& command, pmode mode = std::ios_base::in)
      : istream_type(NULL), pbase_type(command, mode)
      {}

      basic_ipstream(const std::string& file, const std::vector<std::string>& argv, pmode mode = std::ios_base::in)
      : istream_type(NULL), pbase_type(file, argv, mode)
      {}

      ~basic_ipstream()
      {}

      void
      open(const std::string& command, pmode mode = std::ios_base::in)
      { pbase_type::open(command, mode); }

      void
      open(const std::string& file, const std::vector<std::string>& argv, pmode mode = std::ios_base::in)
      { pbase_type::open(file, argv, mode); }

      basic_ipstream&
      out()
      {
        buf_.read_err(false);
        return *this;
      }

      basic_ipstream&
      err()
      {
        buf_.read_err(true);
        return *this;
      }
    };


  template <typename CharT, typename Traits = std::char_traits<CharT> >
    class basic_opstream
    : public std::basic_ostream<CharT, Traits>
    , public pstream_common<CharT, Traits>
    {
      typedef std::basic_ostream<CharT, Traits>     ostream_type;
      typedef pstream_common<CharT, Traits>         pbase_type;
      typedef typename pbase_type::streambuf_type   streambuf_type;

    public:
      typedef typename pbase_type::pmode            pmode;

      basic_opstream()
      : ostream_type(NULL), pbase_type()
      {}

      basic_opstream(const std::string& command, pmode mode = std::ios_base::out)
      : ostream_type(NULL), pbase_type(command, mode)
      {}

      basic_opstream(const std::string& file, const std::vector<std::string>& argv, pmode mode = std::ios_base::out)
      : ostream_type(NULL), pbase_type(file, argv, mode)
      {}

      ~basic_opstream() { }

      void
      open(const std::string& command, pmode mode = std::ios_base::out)
      { pbase_type::open(command, mode); }

      void
      open(const std::string& file, const std::vector<std::string>& argv, pmode mode = std::ios_base::out)
      { pbase_type::open(file, argv, mode); }
    };


  template <typename CharT, typename Traits = std::char_traits<CharT> >
    class basic_pstream
    : public std::basic_iostream<CharT, Traits>
    , public pstream_common<CharT, Traits>
    {
      typedef std::basic_iostream<CharT, Traits>    iostream_type;
      typedef pstream_common<CharT, Traits>         pbase_type;
      typedef typename pbase_type::streambuf_type   streambuf_type;

    public:
      typedef typename pbase_type::pmode            pmode;

      basic_pstream()
      : iostream_type(NULL), pbase_type()
      {}

      basic_pstream(const std::string& command, pmode mode = std::ios_base::in|std::ios_base::out)
      : iostream_type(NULL), pbase_type(command, mode)
      {}

      basic_pstream(const std::string& file, const std::vector<std::string>& argv, pmode mode = std::ios_base::in|std::ios_base::out)
      : iostream_type(NULL), pbase_type(file, argv, mode)
      {}

      ~basic_pstream() { }

      void
      open(const std::string& command, pmode mode = std::ios_base::in|std::ios_base::out)
      { pbase_type::open(command, mode); }

      void
      open(const std::string& file, const std::vector<std::string>& argv, pmode mode = std::ios_base::in|std::ios_base::out)
      { pbase_type::open(file, argv, mode); }

      basic_pstream&
      out()
      {
        buf_.read_err(false);
        return *this;
      }

      basic_pstream&
      err()
      {
        buf_.read_err(true);
        return *this;
      }
    };


  typedef basic_pstreambuf<char, std::char_traits<char> > pstreambuf;
  typedef basic_ipstream<char> ipstream;
  typedef basic_opstream<char> opstream;
  typedef basic_pstream<char> pstream;


  template <typename C, typename T>
    std::basic_ostream<C,T>&
    peof(std::basic_ostream<C,T>& s);

  template <typename C, typename T>
    std::basic_ostream<C,T>&
    peof(std::basic_ostream<C,T>& s)
    {
      static_cast<basic_pstreambuf<C,T>*>(s.rdbuf())->peof();
      //dynamic_cast<basic_pstreambuf<C,T>*>(s.rdbuf())->peof();
      return s;
    }


  /*
   * member definitions for pstreambuf
   */


  template <typename C, typename T>
    inline
    basic_pstreambuf<C,T>::basic_pstreambuf()
    : ppid_(0)
    , wpipe_(-1)
    , wbuffer_(0)
    , rsrc_(rsrc_out)
    , status_(-1)
    , error_(0)
    {
      init_rbuffers();
    }


  template <typename C, typename T>
    inline
    basic_pstreambuf<C,T>::basic_pstreambuf(const std::string& command, pmode mode)
    : ppid_(0)
    , wpipe_(-1)
    , wbuffer_(0)
    , rsrc_(rsrc_out)
    , status_(-1)
    , error_(0)
    {
      init_rbuffers();
      open(command, mode);
    }

  template <typename C, typename T>
    inline
    basic_pstreambuf<C,T>::basic_pstreambuf(const std::string& file, const std::vector<std::string>& argv, pmode mode)
    : ppid_(0)
    , wpipe_(-1)
    , wbuffer_(0)
    , rsrc_(rsrc_out)
    , status_(-1)
    , error_(0)
    {
      init_rbuffers();
      open(file, argv, mode);
    }

  template <typename C, typename T>
    inline
    basic_pstreambuf<C,T>::~basic_pstreambuf()
    {
      close();
    }

  template <typename C, typename T>
    inline basic_pstreambuf<C,T>*
    basic_pstreambuf<C,T>::open(const std::string& command, pmode mode)
    {
      basic_pstreambuf<C,T>* ret = NULL;

      if (!is_open())
      {
        switch(fork(mode))
        {
          case 0 :
          {
            // this is the new process, exec command
            ::execlp("sh", "sh", "-c", command.c_str(), 0);

            // can only reach this point if exec() failed

            // parent can get exit code from waitpid()
            std::exit(errno);
          }
          case -1 :
          {
            // couldn't fork, error already handled in pstreambuf::fork()
            break;
          }
          default :
          {
            // this is the parent process
#if 0
            // check process not exited already
            // very unlikely, since not enough time for shell to parse cmd!

            switch (wait(true))
            {
              case 0 :
                // activate buffers
                create_buffers(mode);
                ret = this;
                break;
              case 1:
                // child exited already
                sync();
                close_fd_array(&wpipe_, 1);
                close_fd_array(rpipe_, 2);
                break;
              default :
                break;
            }
#else
            // activate buffers
            create_buffers(mode);
            ret = this;
#endif
          }
        }
      }
      return ret;
    }


  template <typename C, typename T>
    inline basic_pstreambuf<C,T>*
    basic_pstreambuf<C,T>::open(const std::string& file, const std::vector<std::string>& argv, pmode mode)
    {
      basic_pstreambuf<C,T>* ret = NULL;

      if (!is_open())
      {
        switch(fork(mode))
        {
          case 0 :
          {
            // this is the new process, exec command

            char** arg_v = new char*[argv.size()+1];
            for (size_t i = 0; i < argv.size(); ++i)
            {
              const std::string& src = argv[i];
              char*& dest = arg_v[i];
              dest = new char[src.size()+1];
              dest[ src.copy(dest, src.size()) ] = 0;
            }
            arg_v[argv.size()] = 0;

            ::execvp(file.c_str(), arg_v);

            // can only reach this point if exec() failed

            // parent can get exit code from waitpid()
            std::exit(errno);
          }
          case -1 :
          {
            // couldn't fork, error already handled in pstreambuf::fork()
            break;
          }
          default :
          {
            // this is the parent process
#if 0
            // check process not exited already
            // very unlikely, since not enough time for shell to parse cmd!

            switch (wait(true))
            {
              case 0 :
                // activate buffers
                create_buffers(mode);
                ret = this;
                break;
              case 1:
                // child exited already
                sync();
                close_fd_array(&wpipe_, 1);
                close_fd_array(rpipe_, 2);
                break;
              default :
                break;
            }
#else
            // activate buffers
            create_buffers(mode);
            ret = this;
#endif
          }
        }
      }
      return ret;
    }

  inline void
  close_fd_array(int* filedes, size_t count)
  {
    for (size_t i = 0; i < count; ++i)
      if (filedes[i] >= 0)
        if (::close(filedes[i]) == 0)
          filedes[i] = -1;
  }

  template <typename C, typename T>
    pid_t
    basic_pstreambuf<C,T>::fork(pmode mode)
    {
      pid_t pid = -1;

      // three pairs of file descriptors, for pipes connected to the
      // process' stdin, stdout and stderr
      // (stored in a single array so close_fd_array() can close all at once)
      fd_t fd[6] =  {-1, -1, -1, -1, -1, -1};
      fd_t* pin = fd;
      fd_t* pout = fd+2;
      fd_t* perr = fd+4;

      // constants for read/write ends of pipe
      const int RD = 0;
      const int WR = 1;

      // N.B.
      // For the pstreambuf pin is an output stream and
      // pout and perr are input streams.

      if (!error_ && mode&pstdin && ::pipe(pin))
        error_ = errno;

      if (!error_ && mode&pstdout && ::pipe(pout))
        error_ = errno;

      if (!error_ && mode&pstderr && ::pipe(perr))
        error_ = errno;

      if (!error_)
      {
        pid = ::fork();
        switch (pid)
        {
          case 0 :
          {
            // this is the new process

            // for each open pipe close one end and redirect the
            // respective standard stream to the other end

            if (*pin >= 0)
            {
              ::close(pin[WR]);
              ::dup2(pin[RD], STDIN_FILENO);
              ::close(pin[RD]);
            }
            if (*pout >= 0)
            {
              ::close(pout[RD]);
              ::dup2(pout[WR], STDOUT_FILENO);
              ::close(pout[WR]);
            }
            if (*perr >= 0)
            {
              ::close(perr[RD]);
              ::dup2(perr[WR], STDERR_FILENO);
              ::close(perr[WR]);
            }
            break;
          }
          case -1 :
          {
            // couldn't fork for some reason
            error_ = errno;
            // close any open pipes
            close_fd_array(fd, 6);
            break;
          }
          default :
          {
            // this is the parent process, store process' pid
            ppid_ = pid;

            // store one end of open pipes and close other end
            if (*pin >= 0)
            {
              wpipe_ = pin[WR];
              ::close(pin[RD]);
            }
            if (*pout >= 0)
            {
              rpipe_[rsrc_out] = pout[RD];
              ::close(pout[WR]);
            }
            if (*perr >= 0)
            {
              rpipe_[rsrc_err] = perr[RD];
              ::close(perr[WR]);
            }

            if (rpipe_[rsrc_out] == -1 && rpipe_[rsrc_err] >= 0)
            {
              // reading stderr but not stdout, so use stderr for all reads
              read_err(true);
            }
          }
        }
      }
      else
      {
        // close any pipes we opened before failure
        close_fd_array(fd, 6);
      }
      return pid;
    }


  template <typename C, typename T>
    inline basic_pstreambuf<C,T>*
    basic_pstreambuf<C,T>::close()
    {
      basic_pstreambuf<C,T>* ret = NULL;
      if (is_open())
      {
        sync();
        destroy_buffers(pstdin|pstdout|pstderr);

        close_fd_array(&wpipe_, 1);
        close_fd_array(rpipe_, 2);

        if (wait() == 1)
        {
          ret = this;
        }
      }
      return ret;
    }


  template <typename C, typename T>
    inline void
    basic_pstreambuf<C,T>::init_rbuffers()
    {
      rpipe_[rsrc_out] = rpipe_[rsrc_err] = -1;
      rbuffer_[rsrc_out] = rbuffer_[rsrc_err] = 0;
      rbufstate_[0] = rbufstate_[1] = rbufstate_[2] = 0;
    }


  template <typename C, typename T>
    void
    basic_pstreambuf<C,T>::create_buffers(pmode mode)
    {
      if (mode & pstdin)
      {
        delete[] wbuffer_;
        wbuffer_ = new char_type[bufsz];
        setp(wbuffer_, wbuffer_ + bufsz);
      }
      if (mode & pstdout)
      {
        delete[] rbuffer_[rsrc_out];
        rbuffer_[rsrc_out] = new char_type[bufsz];
        if (rsrc_ == rsrc_out)
          setg(rbuffer_[rsrc_out] + pbsz, rbuffer_[rsrc_out] + pbsz,
              rbuffer_[rsrc_out] + pbsz);
      }
      if (mode & pstderr)
      {
        delete[] rbuffer_[rsrc_err];
        rbuffer_[rsrc_err] = new char_type[bufsz];
        if (rsrc_ == rsrc_err)
          setg(rbuffer_[rsrc_err] + pbsz, rbuffer_[rsrc_err] + pbsz,
              rbuffer_[rsrc_err] + pbsz);
      }
    }


  template <typename C, typename T>
    void
    basic_pstreambuf<C,T>::destroy_buffers(pmode mode)
    {
      if (mode & pstdin)
      {
        setp(0, 0);
        delete[] wbuffer_;
        wbuffer_ = 0;
      }
      if (mode & pstdout)
      {
        if (rsrc_ == rsrc_out)
          setg(0, 0, 0);
        delete[] rbuffer_[rsrc_out];
        rbuffer_[rsrc_out] = 0;
      }
      if (mode & pstderr)
      {
        if (rsrc_ == rsrc_err)
          setg(0, 0, 0);
        delete[] rbuffer_[rsrc_err];
        rbuffer_[rsrc_err] = 0;
      }
    }

  template <typename C, typename T>
    typename basic_pstreambuf<C,T>::buf_read_src
    basic_pstreambuf<C,T>::switch_read_buffer(buf_read_src src)
    {
      if (rsrc_ != src)
      {
        char_type* tmpbufstate[3];
        tmpbufstate[0] = eback();
        tmpbufstate[1] = gptr();
        tmpbufstate[2] = egptr();
        setg(rbufstate_[0], rbufstate_[1], rbufstate_[2]);
        for (size_t i = 0; i < 3; ++i)
          rbufstate_[i] = tmpbufstate[i];
      }
      return rsrc_;
    }


  template <typename C, typename T>
    inline int
    basic_pstreambuf<C,T>::wait(bool nohang)
    {
      int exited = -1;
      if (is_open())
      {
        int status;
        switch(::waitpid(ppid_, &status, nohang ? WNOHANG : 0))
        {
          case 0 :
            // nohang was true and process has not exited
            exited = 0;
            break;
          case -1 :
            error_ = errno;
            break;
          default :
            // process has exited
            ppid_ = 0;
            status_ = status;
            exited = 1;
            // TODO  close pipes with close_fd_array() ?
            break;
        }
      }
      return exited;
    }


  template <typename C, typename T>
    inline basic_pstreambuf<C,T>*
    basic_pstreambuf<C,T>::kill(int signal)
    {
      basic_pstreambuf<C,T>* ret = NULL;
      if (is_open())
      {
        if (::kill(ppid_, signal))
          error_ = errno;
        else
          ret = this;
      }
      return ret;
    }


  template <typename C, typename T>
    inline bool
    basic_pstreambuf<C,T>::exited()
    {
      // TODO  should close() if is_open() and has exited
      return wait(true)==1;
    }


  template <typename C, typename T>
    inline int
    basic_pstreambuf<C,T>::status() const
    {
      return status_;
    }

  template <typename C, typename T>
    inline int
    basic_pstreambuf<C,T>::error() const
    {
      return error_;
    }

  template <typename C, typename T>
    inline void
    basic_pstreambuf<C,T>::peof()
    {
      sync();
      destroy_buffers(pstdin);
      close_fd_array(&wpipe_, 1);
    }

  template <typename C, typename T>
    inline bool
    basic_pstreambuf<C,T>::is_open() const
    {
      return bool(ppid_>0);
    }

  template <typename C, typename T>
    inline bool
    basic_pstreambuf<C,T>::read_err(bool readerr)
    {
      buf_read_src src = readerr ? rsrc_err : rsrc_out;
      if (rpipe_[src]>=0)
      {
        rsrc_ = src;
        return true;
      }
      return false;
    }

  template <typename C, typename T>
    typename basic_pstreambuf<C,T>::int_type
    basic_pstreambuf<C,T>::overflow(int_type c)
    {
      if (!empty_buffer())
        return traits_type::eof();
      else if (!traits_type::eq_int_type(c, traits_type::eof()))
        return sputc(c);
      else
        return traits_type::not_eof(c);
    }


  template <typename C, typename T>
    int
    basic_pstreambuf<C,T>::sync()
    {
      return (empty_buffer() ? 0 : -1);
    }


  template <typename C, typename T>
    std::streamsize
    basic_pstreambuf<C,T>::xsputn(const char_type* s, std::streamsize n)
    {
      if (n < epptr() - pptr())
      {
        memcpy(pptr(), s, n * sizeof(char_type));
        pbump(n);
        return n;
      }
      else
      {
        for (std::streamsize i = 0; i < n; ++i)
        {
          if (traits_type::eq_int_type(sputc(s[i]), traits_type::eof()))
            return i;
        }
        return n;
      }
    }


  template <typename C, typename T>
    bool
    basic_pstreambuf<C,T>::empty_buffer()
    {
      int count = pptr() - pbase();
      std::streamsize written = write(wbuffer_, count);
      if (count > 0 && written == count)
      {
        pbump(-written);
        return true;
      }
      return false;
    }


  template <typename C, typename T>
    typename basic_pstreambuf<C,T>::int_type
    basic_pstreambuf<C,T>::underflow()
    {
      if (gptr() < egptr() || fill_buffer())
        return traits_type::to_int_type(*gptr());
      else
        return traits_type::eof();
    }


  template <typename C, typename T>
    typename basic_pstreambuf<C,T>::int_type
    basic_pstreambuf<C,T>::pbackfail(int_type c)
    {
      if (gptr() != eback())
      {
        gbump(-1);
        if (!traits_type::eq_int_type(c, traits_type::eof()))
          *gptr() = traits_type::to_char_type(c);
        return traits_type::not_eof(c);
      }
      else
         return traits_type::eof();
    }

  template <typename C, typename T>
    bool
    basic_pstreambuf<C,T>::fill_buffer()
    {
      int npb = std::min(gptr()-eback(), static_cast<int>(pbsz));

      std::memmove(rbuffer()+pbsz-npb, gptr()-npb, npb*sizeof(char_type));

      std::streamsize rc = read(rbuffer() + pbsz, bufsz - pbsz);

      if (rc > 0)
      {
        setg(rbuffer()+pbsz-npb, rbuffer()+pbsz, rbuffer()+pbsz+rc);
        return true;
      }
      else
      {
        setg(0, 0, 0);
        return false;
      }
    }


  template <typename C, typename T>
    inline bool
    basic_pstreambuf<C,T>::write(char_type c)
    {
      return (write(&c, 1) == 1);
    }

  template <typename C, typename T>
    inline bool
    basic_pstreambuf<C,T>::read(char_type& c)
    {
      return (read(&c, 1) == 1);
    }

  template <typename C, typename T>
    inline std::streamsize
    basic_pstreambuf<C,T>::write(char_type* s, std::streamsize n)
    {
      return (wpipe() >= 0 ? ::write(wpipe(), s, n * sizeof(char_type)) : 0);
    }

  template <typename C, typename T>
    inline std::streamsize
    basic_pstreambuf<C,T>::read(char_type* s, std::streamsize n)
    {
      return (rpipe() >= 0 ? ::read(rpipe(), s, n * sizeof(char_type)) : 0);
    }


  template <typename C, typename T>
    inline typename basic_pstreambuf<C,T>::fd_t&
    basic_pstreambuf<C,T>::wpipe()
    {
      return wpipe_;
    }

  template <typename C, typename T>
    inline typename basic_pstreambuf<C,T>::fd_t&
    basic_pstreambuf<C,T>::rpipe()
    {
      return rpipe_[rsrc_];
    }

  template <typename C, typename T>
    inline typename basic_pstreambuf<C,T>::fd_t&
    basic_pstreambuf<C,T>::rpipe(buf_read_src which)
    {
      return rpipe_[which];
    }

  template <typename C, typename T>
    inline typename basic_pstreambuf<C,T>::char_type*
    basic_pstreambuf<C,T>::rbuffer()
    {
      return rbuffer_[rsrc_];
    }


  /*
   * member definitions for pstream_common
   */

  template <typename C, typename T>
    inline
    pstream_common<C,T>::pstream_common()
    : std::basic_ios<C,T>(NULL)
    , command_()
    , buf_()
    {
      init(&buf_);
    }

  template <typename C, typename T>
    inline
    pstream_common<C,T>::pstream_common(const std::string& command, pmode mode)
    : std::basic_ios<C,T>(NULL)
    , command_(command)
    , buf_()
    {
      init(&buf_);
      open(command, mode);
    }
   
  template <typename C, typename T>
    inline
    pstream_common<C,T>::pstream_common(const std::string& file, const std::vector<std::string>& argv, pmode mode)
    : std::basic_ios<C,T>(NULL)
    , command_(file)
    , buf_()
    {
      init(&buf_);
      open(file, argv, mode);
    }

  template <typename C, typename T>
    inline
    pstream_common<C,T>::~pstream_common()
    {
    }

  template <typename C, typename T>
    inline void
    pstream_common<C,T>::open(const std::string& command, pmode mode)
    {
      if (!buf_.open((command_=command), mode))
        setstate(std::ios_base::failbit);
    }

  template <typename C, typename T>
    inline void
    pstream_common<C,T>::open(const std::string& file, const std::vector<std::string>& argv, pmode mode)
    {
      if (!buf_.open((command_=file), argv, mode))
        setstate(std::ios_base::failbit);
    }

  template <typename C, typename T>
    inline void
    pstream_common<C,T>::close()
    {
      if (!buf_.close())
        setstate(std::ios_base::failbit);
    }

  template <typename C, typename T>
    inline bool
    pstream_common<C,T>::is_open() const
    {
      return buf_.is_open();
    }

  template <typename C, typename T>
    inline const std::string&
    pstream_common<C,T>::command() const
    {
      return command_;
    }

  // TODO  document behaviour if buffer replaced.
  template <typename C, typename T>
    inline typename pstream_common<C,T>::streambuf_type*
    pstream_common<C,T>::rdbuf() const
    {
      return const_cast<streambuf_type*>(&buf_);
    }

#if REDI_EVISCERATE_PSTREAMS

  template <typename C, typename T>
    inline size_t
    basic_pstreambuf<C,T>::fopen(FILE*& in, FILE*& out, FILE*& err)
    {
      in = out = err = NULL;
      size_t open_files = 0;
      if (wpipe() > -1)
      {
        if (in = ::fdopen(wpipe(), "w"))
        {
            open_files |= pstdin;
        }
      }
      if (rpipe(rsrc_out) > -1) 
      {
        if (out = ::fdopen(rpipe(rsrc_out), "r"))
        {
            open_files |= pstdout;
        }
      }
      if (rpipe(rsrc_err) > -1)
      {
        if (err = ::fdopen(rpipe(rsrc_err), "r"))
        {
            open_files |= pstderr;
        }
      }
      return open_files;
    }

  template <typename C, typename T>
    inline size_t
    pstream_common<C,T>::fopen(FILE*& in, FILE*& out, FILE*& err)
    {
      return buf_.fopen(in, out, err);
    }

#endif // REDI_EVISCERATE_PSTREAMS


} // namespace redi

#endif  // REDI_PSTREAM_H

// vim: ts=2 sw=2 expandtab

---