#ifndef DPBUFFER_HEADER
#define DPBUFFER_HEADER 1
////////////////////////////////////////////////////////
//! rcsid="$Id: Buffer.hh,v 1.13 2000/12/13 12:21:03 ees1cg Exp $"
//! lib=MTDataProc
//! docentry="Data Processing"
//! file="amma/StdType/MTDataProc/Buffer.hh"
//! author="Charles Galambos"
//! date="02/10/98"

#include "amma/PThread/Mutex.hh"
#include "amma/PThread/Semaphore.hh"
#include "amma/DList.hh"
#include "amma/DP/Port.hh"

///////////////////////////////
//! userlevel=Develop
//: MT safe queue body

template<class DataT>
class DPBufferBodyC : public DPIOPortBodyC<DataT,DataT>
{
public:
  DPBufferBodyC(BooleanT nblock = TRUE) 
    : ready(0),
      block(nblock)
  {}
  //: Default constructor.
  
  virtual BooleanT IsAsync() const;
  //: Does port work asynchronously ??
  
  virtual void PutEOS();
  //: Put End Of Stream marker.
  
  virtual BooleanT IsPutReady() const;
  //: Is port ready for data ?

  virtual BooleanT Put(const DataT &dat);
  //: Put data.
  
  virtual IntT PutArray(const SArray1dC<DataT> &data);
  //: Put an array of data to stream.
  // returns number of elements processed.
  
  virtual BooleanT IsGetReady() const;
  //: Is some data ready ?
  // TRUE = yes.
  // Defaults to !IsGetEOS().
  
  virtual BooleanT IsGetEOS() const;
  //: Has the End Of Stream been reached ?
  // TRUE = yes.
  
  virtual DataT Get();
  //: Get next piece of data.
  // May block if not ready, or it will return a constructed
  // with the default constructor.
  
  virtual IntT GetArray(SArray1dC<DataT> &data);
  //: Get an array of data from buffer.
  // returns number of elements processed.
  
  virtual BooleanT Get(DataT &buff);
  //: Try and get next piece of data.
  
  //  virtual BodyRefCounterVC &Copy() const; 
private:
  DListC<DataT> list; // Data buffer
  PThread::MutexC access;     // Mutual exclusion.
  PThread::SemaphoreC ready; // Inc'd and Dec'd with items in list.
  BooleanT block;   // Should be block if no data is available ?
  BooleanT gotEOS;  // Have we got an EOS ?
};

//////////////////////////////////
//! userlevel=Normal
//: MT safe queue handle.

template<class DataT>
class DPBufferC : public DPIOPortC<DataT,DataT> 
{
public:
  DPBufferC()
    : DPEntityC(aTRUE)
  {}
  //:Default Constructor.
  // Creates an invalid handle.
  
  DPBufferC(BooleanT nblock)
    : DPEntityC(*new DPBufferBodyC<DataT>(nblock))
  {}
  //: Constructor.
  
};

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

template<class DataT>
BooleanT DPBufferBodyC<DataT>::IsAsync() const 
{ return TRUE; }

template<class DataT>
void DPBufferBodyC<DataT>::PutEOS()
{ gotEOS = TRUE; }

template<class DataT>
BooleanT DPBufferBodyC<DataT>::IsPutReady() const
{ return !gotEOS; }

template<class DataT>
BooleanT DPBufferBodyC<DataT>::Put(const DataT &dat) 
{ 
  access.Lock();
  list.InsLast(dat);
  access.Unlock();
  ready.Post();
  return TRUE; 
}

template<class DataT>
IntT DPBufferBodyC<DataT>::PutArray(const SArray1dC<DataT> &data) {
  access.Lock();
  for(SArray1dIterC<DataT> it(data);it;it++) {
    list.InsLast(*it);
    ready.Post();
  }
  access.Unlock();
  return data.Size(); 
}

template<class DataT>
BooleanT DPBufferBodyC<DataT>::IsGetReady() const
{ 
  if(!block)
    return (ready.Count() > 0); 
  return true;
}

template<class DataT>
BooleanT DPBufferBodyC<DataT>::IsGetEOS() const
{ 
  if(!gotEOS)
    return FALSE;
  return (ready.Count() > 0);
}
  
template<class DataT>
DataT DPBufferBodyC<DataT>::Get() 
{ 
  DataT ret;
  if(!block) {
    if(!ready.TryWait())
      throw DataNotReadyC();
  } else
    ready.Wait();
  access.Lock();
#ifdef AMMA_CHECK
  if(list.IsEmpty()) {
    access.Unlock();
    cerr << "DPBufferBodyC<DataT>::Get(), WARNING: Sync lost!\n";
    throw DataNotReadyC("Unexpected data not ready.");
  }
#endif
  ret = list.GetFirst();
  access.Unlock();
  return ret; 
}

template<class DataT>
BooleanT DPBufferBodyC<DataT>::Get(DataT &buff)
{
  if(!block) {
    if(!ready.TryWait())
      return FALSE;
  } else
    ready.Wait();
  access.Lock();
#ifdef AMMA_CHECK
  if(list.IsEmpty()) {
    access.Unlock();
    cerr << "DPBufferBodyC<DataT>::Get(DataT &), WARNING: Sync lost!\n";
    return FALSE;
  }
#endif
  buff = list.GetFirst();
  access.Unlock();
  return TRUE;
}

template<class DataT>
IntT DPBufferBodyC<DataT>::GetArray(SArray1dC<DataT> &data) {
  for(SArray1dIterC<DataT> it(data);it;it++) {
    if(!block) {
      if(!ready.TryWait())
	return it.Index().V();
    } else
      ready.Wait();
    access.Lock();
#ifdef AMMA_CHECK
    if(list.IsEmpty()) {
      access.Unlock();
      cerr << "DPBufferBodyC<DataT>::GetArray(SArray1dC<DataT> &), WARNING: Sync lost!\n";
      return FALSE;
    }
#endif
    // We've got at least 1 item.
    *it = list.GetFirst();
    it++;
    // Read anything else we can.
    // don't read a list item before its semaphore has been posted.
    for(;it && !list.IsEmpty();it++) {
      if(!ready.TryWait())
	break;
      *it = list.GetFirst();
    }
    access.Unlock();
  }
  return data.Size();
}

#endif