pipe.C

#include <unistd.h>
#include <string>
#include <iostream>

#include <archon/util/exception.H>
#include <archon/util/stream.H>
#include <archon/util/pipe.H>

using namespace std;
using namespace Archon::Utilities;

struct Producer: Thread
{
  Ref<Stream::Writer> w;
  const char *s;

  Producer(Ref<Stream::Writer> w, const char *s):w(w), s(s) {}

  virtual void main()
  {
    char buffer[64];

    for(int i=0;i<20000;++i)
    {
      sprintf(buffer, "%s(%d)\n", s, i);
      w->writeAll(buffer, strlen(buffer));
    }
    w.reset();
  }
};

struct Consumer: Thread
{
  Ref<Stream::Reader> r;

  Consumer(Ref<Stream::Reader> r):r(r) {}

  virtual void main()
  {
    for(;;)
    {
      char b[1025];
      int n = r->read(b, 1025);
      if(n == 0) break;
      write(STDOUT_FILENO, b, n);
    }
    r.reset();
  }
};

int main() throw()
{
  set_unexpected(Exception::terminal<exceptionCatchInfo>);
  set_terminate (Exception::terminal<exceptionCatchInfo>);

  Ref<Stream::Reader> r;
  Ref<Stream::Writer> w;
  Stream::Pipe::make(r, w, 253, true);

  RefAny a = r;

  cerr << r->getUseCount() << "\n";
  cerr << w->getUseCount() << "\n";

  Thread::sleep(Time(2.0));

  a.reset();

  Ref<Producer> producer1 = new Producer(w, "Alpha");
  Ref<Producer> producer2 = new Producer(w, "Beta");
  Ref<Producer> producer3 = new Producer(w, "Gamma");
  Ref<Consumer> consumer  = new Consumer(r);
  r.reset();
  w.reset();
  Thread::start(producer1);
  Thread::start(producer2);
  Thread::start(producer3);
  Thread::start(consumer);

  Thread::mainExitWait();

  return 0;
}

---