pipe.H

/*
 * This file is part of the "Archon" framework.
 * (http://files3d.sourceforge.net)
 *
 * Copyright © 2002 by Kristian Spangsege and Brian Kristiansen.
 *
 * Permission to use, copy, modify, and distribute this software and
 * its documentation under the terms of the GNU General Public License is
 * hereby granted. No representations are made about the suitability of
 * this software for any purpose. It is provided "as is" without express
 * or implied warranty. See the GNU General Public License
 * (http://www.gnu.org/copyleft/gpl.html) for more details.
 *
 * The characters in this file are ISO8859-1 encoded.
 *
 * The documentation in this file is in "Doxygen" style
 * (http://www.doxygen.org).
 */

#ifndef ARCHON_UTILITIES_PIPE_H
#define ARCHON_UTILITIES_PIPE_H

#include <string.h>

#include <archon/util/mutex.H>
#include <archon/util/stream.H>

namespace Archon
{
  namespace Utilities
  {
    namespace Stream
    {
      using namespace std;

      template<class C>
      struct BasicPipe: virtual ReaderBase<C>, virtual WriterBase<C>
      {
        static void make(Ref<ReaderBase<C> > &r,
                         Ref<WriterBase<C> > &w,
                         int bufferSize = 4096, bool expand = false)
        {
          BasicPipe<C> *p = new BasicPipe<C>(bufferSize, expand);
          r.reset(p);
          w.reset(p);
        }

        void raiseReadError(string message)
        {
          Mutex::Lock l(*this);
          readError = message;
        }

        void raiseWriteError(string message)
        {
          Mutex::Lock l(*this);
          writeError = message;
        }

        int read(C *b, int n)
          throw(ReadException, UnexpectedException)
        {
          if(n == 0) return 0;
          Mutex::Lock l(*this);

          for(;;)
          {
            if(readError.size()) ARCHON_THROW1(ReadException, readError);
            if(buffers.size() > 1 || readPos != writePos) break;
            if(!WriterBase<C>::getWriteCountNoLock()) return 0;
            nonEmpty.wait();
          }

          int rel = writePos - readPos;
          int capasity = buffers.size() > 1 || rel < 0 ? bufferSize - readPos : rel;
          if(capasity < n) n = capasity;

          memcpy(b, buffers.front() + readPos, n*sizeof(C));

          readPos += n;
          if(readPos == bufferSize)
          {
            if(expand)
            {
              delete[] buffers.front();
              buffers.pop_front();
            }
            readPos = 0;
          }

          l.release();
          nonFull.notifyAll();
          return n;
        }

        int write(const C *b, int n)
          throw(WriteException, UnexpectedException)
        {
          if(n == 0) return 0;
          Mutex::Lock l(*this);

          int capasity;
          if(expand)
          {
            if(writeError.size()) ARCHON_THROW1(WriteException, writeError);
            if(!ReaderBase<C>::getReadCountNoLock())
              ARCHON_THROW1(WriteException, "Pipe has no more readers");
            capasity = bufferSize - writePos;
            if(!writePos) buffers.push_back(new C[bufferSize]);
          }
          else for(;;)
          {
            if(writeError.size()) ARCHON_THROW1(WriteException, writeError);
            if(!ReaderBase<C>::getReadCountNoLock())
              ARCHON_THROW1(WriteException, "Pipe has no more readers");
            if(writePos+1 != (readPos ? readPos : bufferSize))
            {
              if(readPos)
              {
                int rel = readPos - writePos;
                capasity = rel > 0 ? rel - 1 : bufferSize - writePos;
              }
              else capasity = bufferSize - writePos - 1;
              if(!buffers.size()) buffers.push_back(new C[bufferSize]);
              break;
            }
            nonFull.wait();
          }

          if(capasity < n) n = capasity;
          
          memcpy(buffers.back() + writePos, b, n*sizeof(C));

          writePos += n;
          if(writePos == bufferSize) writePos = 0;

          l.release();
          nonEmpty.notifyAll();
          return n;
        }

      protected:
        BasicPipe(int bufferSize, bool expand):
          bufferSize(bufferSize), expand(expand),
          readPos(0), writePos(0),
          nonEmpty(*this), nonFull(*this) {}

        virtual ~BasicPipe()
        {
          list<char *>::iterator i = buffers.begin();
          while(i != buffers.end()) delete[] *i++;
        }

        // We know at least one of readCount(), writeCount() have
        // gone to zero.
        void refDispose(Mutex::Lock &l)
        {
          if(ReaderBase<C>::getReadCountNoLock()) nonEmpty.notifyAll();
          else if(WriterBase<C>::getWriteCountNoLock()) nonFull.notifyAll();
          else
          {
            l.release();
            delete this;
          }
        }

      private:
        int bufferSize;

        bool expand;

        list<C *> buffers;

        int readPos;

        int writePos;

        string readError;
        string writeError;

        Condition nonEmpty, nonFull;
      };

      typedef BasicPipe<char> Pipe;
    }
  }
}

#endif // ARCHON_UTILITIES_PIPE_H

---