#ifndef DPMULTIPLEX_HEADER
#define DPMULTIPLEX_HEADER 1
////////////////////////////////////////////////////////
//! rcsid="$Id: Multiplex.hh,v 1.12 2000/12/13 12:17:15 ees1cg Exp $"
//! example=exDPMultiplex.cc
//! docentry="Data Processing"
//! file="amma/StdType/DataProc/Multiplex.hh"
//! lib=DataProc
//! author="Charles Galambos"
//! date="04/07/98"

#include "amma/DP/Port.hh"
#include "amma/SArray1d.hh"

#define DPMDEBUG 0
#if DPMDEBUG 
#define ONDEBUG(x) x
#else
#define ONDEBUG(x)
#endif

///////////////////////////////
//! userlevel=Develop
//: Multiplex an operation body.
// This is indended to be used with DPThreadC to
// do multiprocessing. <p>
// This will preserve the order of the data in the 
// stream.

template<class InT,class OutT>
class DPMultiplexBodyC : public DPIOPortBodyC<InT,OutT>
{
public:
  DPMultiplexBodyC(IntT num,const DPIOPortC<InT,OutT> & nproc);
  //: Constructor.
  // NB. This only uses nproc as a template, it is not actually
  // used for processing. This avoids possible problems of copying
  // data out of a running thread. <p>
  // ** nproc MUST not have a running thread **.
  
  DPMultiplexBodyC(const SArray1dC<DPIOPortC<InT,OutT> > &nprocs);
  //: Constructor.
  
  DPMultiplexBodyC(istream &in);
  //: Stream constructor.
  
  DPMultiplexBodyC(const DPMultiplexBodyC &oth);
  //: Copy constructor.
  // Makes a deep copy of procs array.
  
  virtual BooleanT IsPutReady() const;
  //: Is some data ready ?
  // TRUE = yes.
  
  virtual BooleanT IsGetReady() const;
  //: Is some data ready ?
  // TRUE = yes.

  virtual void PutEOS();
  //: Put End Of Stream marker.
  
  virtual BooleanT Put(const OutT &dat);
  //: Put data.
  
  virtual IntT PutArray(const SArray1dC<OutT> &data);
  //: Put an array of data to stream.
  // returns the number of elements processed.
  
  virtual InT Get();
  //: Get next piece of data.
  // May block if not ready, or it will return a constructed
  // with the default constructor.
  
  inline IntT GetArray(SArray1dC<InT> &data);
  //: Get an array of data from stream.
  // returns the number of elements processed.
  
  virtual BooleanT Get(InT &buff);
  //: Get next piece of data.
  // May block if not ready, or it will return a constructed
  // with the default constructor.

  virtual BooleanT IsGetEOS() const;
  //: Is get EOS ?
  
  virtual BodyRefCounterVC &Copy() const;
  //: Make a deep copy of object.
  
  virtual BooleanT Save(ostream &out) const;
  //: Save to ostream.
  
private:
  IntT in,out;
  SArray1dC<DPIOPortC<InT,OutT> > procs;
};

//////////////////////////////////
//! userlevel=Normal
//: Multiplex an operation.
// This is indended to be used with DPThreadC to
// do multiprocessing.

template<class InT,class OutT>
class DPMultiplexC : public DPIOPortC<InT,OutT> 
{
public:
  DPMultiplexC();
  //: Default constructor.
  
  DPMultiplexC(IntT num,const DPIOPortC<InT,OutT> & nproc);
  //: Constructor.
  
  DPMultiplexC(const SArray1dC<DPIOPortC<InT,OutT> > &procs);
  //: Constructor.
  
  DPMultiplexC(const DPMultiplexC<InT,OutT> &oth);
  //: Copy constructor.
};

///////////////////////////////

template<class InT,class OutT>
DPIOPortC<InT,OutT> DPMultiplex(IntT num,const DPIOPortC<InT,OutT> & nproc)
{ return DPMultiplexC<InT,OutT>(num,nproc); }

template<class InT,class OutT>
DPIOPortC<InT,OutT> DPMultiplex(const SArray1dC<DPIOPortC<InT,OutT> > &procs)
{ return DPMultiplexC<InT,OutT>(procs); }

////////////////////////////////////

template<class InT,class OutT>
DPMultiplexBodyC<InT,OutT>::DPMultiplexBodyC(IntT num,const DPIOPortC<InT,OutT> & nproc)
  : in(0),
    out(0),
    procs(num)
{
  assert(nproc.IsValid());
  for(IntT i = 0;i < num;i++)
    procs[i] = nproc.Copy();
}

template<class InT,class OutT>
DPMultiplexBodyC<InT,OutT>::DPMultiplexBodyC(const SArray1dC<DPIOPortC<InT,OutT> > &nprocs)
  : in(0),
    out(0),
    procs(nprocs)
{}

template<class InT,class OutT>
DPMultiplexBodyC<InT,OutT>::DPMultiplexBodyC(istream &strmin)
  : DPIOPortBodyC<InT,OutT>(strmin)
{ strmin >> in >> out >> procs; }

template<class InT,class OutT>
DPMultiplexBodyC<InT,OutT>::DPMultiplexBodyC(const DPMultiplexBodyC &oth) 
  : in(oth.in),
    out(oth.out),
    procs(oth.procs.Size())
{
  for(IntT i = 0;i < (IntT) oth.procs.Size();i++)
    procs[i] = oth.procs[i].Copy();
}

template<class InT,class OutT>
BooleanT 
DPMultiplexBodyC<InT,OutT>::IsPutReady() const
{ return procs[in].IsPutReady(); }

template<class InT,class OutT>
BooleanT 
DPMultiplexBodyC<InT,OutT>::IsGetReady() const
{ return procs[out].IsGetReady(); }


template<class InT,class OutT>
void 
DPMultiplexBodyC<InT,OutT>::PutEOS()
{ 
  IntT i;
  // Put EOS's in proper sequence. 
  for(i = in;i < (IntT) procs.Size();i++)
    procs[i].PutEOS();
  for(i = 0;i < in;i++)
    procs[i].PutEOS();
}

template<class InT,class OutT>
BooleanT 
DPMultiplexBodyC<InT,OutT>::Put(const OutT &dat) 
{ 
  ONDEBUG(cerr << "DPMultiplexBodyC<InT,OutT>::Put(),  in:" << in << " Started\n");
  if(!procs[in].Put(dat)) {
    cerr << "WARNING: DPMultiplexBodyC<InT,OutT>::Put(), Failed. \n";
    return FALSE;
  }
  ONDEBUG(cerr << "DPMultiplexBodyC<InT,OutT>::Put(),  in:" << in << " Done\n");
  in++;
  if(in >= (IntT) procs.Size())
    in = 0;
  return TRUE; 
}

template<class InT,class OutT>
IntT DPMultiplexBodyC<InT,OutT>::PutArray(const SArray1dC<OutT> &data) {
  for(SArray1dIterC<OutT> it(data);it;it++) {
    if(!procs[in].Put(*it)) {
      cerr << "WARNING: DPMultiplexBodyC<InT,OutT>::PutArray(), Failed. \n";
      return it.Index().V();
    }
    in++;
    if(in >= (IntT) procs.Size())
      in = 0;
  }
  return data.Size();
}

template<class InT,class OutT>
InT 
DPMultiplexBodyC<InT,OutT>::Get() 
{ 
  ONDEBUG(cerr << "DPMultiplexBodyC<InT,OutT>::Get(),  out:" << out << " Started\n");
  InT ret = procs[out].Get(); 
  ONDEBUG(cerr << "DPMultiplexBodyC<InT,OutT>::Get(),  out:" << out << " Done.\n");
  out++;
  if(out >= (IntT) procs.Size())
    out = 0;
  return ret;
}

template<class InT,class OutT>
BooleanT DPMultiplexBodyC<InT,OutT>::Get(InT &buff)
{
  ONDEBUG(cerr << "DPMultiplexBodyC<InT,OutT>::Get(),  out:" << out << " Started\n");
  if(!procs[out].Get(buff)) {
    cerr << "WARNING: DPMultiplexBodyC<InT,OutT>::Get(), Failed. \n";
    return FALSE;
  }
  ONDEBUG(cerr << "DPMultiplexBodyC<InT,OutT>::Get(),  out:" << out << " Done.\n");
  out++;
  if(out >= (IntT) procs.Size())
    out = 0;
  return TRUE;
}


template<class InT,class OutT>
IntT DPMultiplexBodyC<InT,OutT>::GetArray(SArray1dC<InT> &data) { 
  for(SArray1dIterC<InT> it(data);it;it++) {
    if(!procs[out].Get(*it)) {
      cerr << "WARNING: DPMultiplexBodyC<InT,OutT>::PutArray(), Failed. \n";
      return it.Index().V();
    }
    out++;
    if(out >= (IntT) procs.Size())
      out = 0;
  }
  return data.Size();
}

template<class InT,class OutT>
BooleanT DPMultiplexBodyC<InT,OutT>::IsGetEOS() const
{ 
  // No point in advancing the stream ptr.
  return procs[out].IsGetEOS(); 
}

template<class InT,class OutT>
BodyRefCounterVC &
DPMultiplexBodyC<InT,OutT>::Copy() const 
{ return *new DPMultiplexBodyC<InT,OutT>(*this); }

template<class InT,class OutT>
BooleanT DPMultiplexBodyC<InT,OutT>::Save(ostream &out) const 
{
  DPIOPortBodyC<InT,OutT>::Save(out);
  out << in << " " << out << " " << procs;
  return TRUE;
}

//////////////////////////////////////////////

template<class InT,class OutT>
DPMultiplexC<InT,OutT>::DPMultiplexC()
{}

template<class InT,class OutT>
DPMultiplexC<InT,OutT>::DPMultiplexC(IntT num,const DPIOPortC<InT,OutT> & nproc)
  : DPEntityC(*new DPMultiplexBodyC<InT,OutT>(num,nproc))
{}
  
template<class InT,class OutT>
DPMultiplexC<InT,OutT>::DPMultiplexC(const SArray1dC<DPIOPortC<InT,OutT> > &procs)
  : DPEntityC(*new DPMultiplexBodyC<InT,OutT>(procs))
{}

template<class InT,class OutT>
DPMultiplexC<InT,OutT>::DPMultiplexC(const DPMultiplexC<InT,OutT> &oth)
  : DPEntityC(oth)
{}

#ifdef ONDEBUG
#undef ONDEBUG
#endif

#endif