#ifndef __ASYNC_STREAM_H_INCLUDED__
#define __ASYNC_STREAM_H_INCLUDED__

#include <iostream>
#include <vector>
#include <stdexcept>
#include <pthread.h>

// The io_direction type is used to distinguish between input and
// output operations
enum io_direction {input=0, output=1};

class astream;

class critical_section {
  private:
	void* handle;

  public:
	critical_section();
	~critical_section();
	void enter();
	void leave();
};

// The iotarget class specifies a single constituent I/O operation
class iotarget
{
  public:
	virtual void invoke() = 0;
};

// ***********************************************************
// The following subclasses of io target represent individual
// I/O functions
// ***********************************************************

// The input_target class represents the >> operator
template<typename STREAM, typename DATATYPE>
class input_target : public iotarget {
  private:
	STREAM *str;
	DATATYPE *d;
  public:
	input_target(STREAM *isp, DATATYPE *data) : str(isp), d(data) {}
	virtual void invoke() { *str >> *d; }
};


// The get1_target class represents the get(char_type&) method
template<typename STREAM>
class get1_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type *cp;
  public:
	get1_target(STREAM *isp, typename STREAM::char_type *data) : str(isp), cp(data) {}
	virtual void invoke() { str->get(*cp); }
};

// The get2_target class represents the get(char_type*,streamsize) method
template<typename STREAM>
class get2_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type *cp;
	std::streamsize siz;
  public:
	get2_target(STREAM *isp, typename STREAM::char_type *data, std::streamsize n)
		: str(isp), cp(data), siz(n) {}
	virtual void invoke() { str->get(cp,siz); }
};

// The get3_target class represents the get(char_type*,streamsize,char_type) method
template<typename STREAM>
class get3_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type *cp;
	std::streamsize siz;
	typename STREAM::char_type del;
  public:
	get3_target(STREAM *isp, typename STREAM::char_type *data, std::streamsize n,
				typename STREAM::char_type delim)
		: str(isp), cp(data), siz(n), del(delim) {}
	virtual void invoke() { str->get(cp,siz,del); }
};

// The get4_target class represents the get(basic_streambuf<...>&) method
template<typename STREAM>
class get4_target : public iotarget {
  private:
	STREAM *str;
	std::basic_streambuf<typename STREAM::char_type, typename STREAM::traits_type> *sbuf;
  public:
	get4_target(STREAM *isp,
				std::basic_streambuf<typename STREAM::char_type, typename STREAM::traits_type> *sb)
		: str(isp), sbuf(sb) {}
	virtual void invoke() { str->get(*this->sb); }
};

// The get5_target class represents the get(basic_streambuf<...>&,char_type) method
template<typename STREAM>
class get5_target : public iotarget {
  private:
	STREAM *str;
	std::basic_streambuf<typename STREAM::char_type, typename STREAM::traits_type> *sbuf;
	typename STREAM::char_type del;
  public:
	get5_target(STREAM *isp,
				std::basic_streambuf<typename STREAM::char_type, typename STREAM::traits_type> *sb,
				typename STREAM::char_type delim)
		: str(isp), sbuf(sb), del(delim) {}
	virtual void invoke() { str->get(*this->sb,del); }
};

// The getline1_target class represents the getline(char_type*,streamsize) method
template<typename STREAM>
class getline1_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type *cp;
	std::streamsize siz;
  public:
	getline1_target(STREAM *isp, typename STREAM::char_type *s, std::streamsize n)
		: str(isp), cp(s), siz(n) {}
	virtual void invoke() { str->getline(cp,siz); }
};

// The getline2_target class represents the getline(char_type*,streamsize,char_type) method
template<typename STREAM>
class getline2_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type *cp;
	std::streamsize siz;
	typename STREAM::char_type del;
  public:
	getline2_target(STREAM *isp, typename STREAM::char_type *s, std::streamsize n,
					typename STREAM::char_type delim)
		: str(isp), cp(s), siz(n), del(delim) {}
	virtual void invoke() { str->getline(cp,siz,del); }
};

// The ignore_target class represents the ignore method
template<typename STREAM>
class ignore_target : public iotarget {
  private:
	STREAM *str;
	std::streamsize siz;
	typename STREAM::int_type del;
  public:
	ignore_target(STREAM *isp, std::streamsize n, typename STREAM::int_type delim)
		: str(isp), siz(n), del(delim) {}
	virtual void invoke() { str->ignore(siz,del); }
};

// The peek_target class represents the peek method
template<typename STREAM>
class peek_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type *cp;
  public:
	peek_target(STREAM *isp, typename STREAM::char_type *data) : str(isp), cp(data) {}
	virtual void invoke() { *cp = str->peek(); }
};

// The read_target class represents the read method
template<typename STREAM>
class read_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type *cp;
	std::streamsize siz;
  public:
	read_target(STREAM *isp, typename STREAM::char_type *s, std::streamsize n)
		: str(isp), cp(s), siz(n) {}
	virtual void invoke() { str->read(cp,siz); }
};


// The output_target class represents the << operator
template<typename STREAM, typename DATATYPE>
class output_target : public iotarget {
  private:
	STREAM *str;
	DATATYPE d;
  public:
	output_target(STREAM *osp, DATATYPE& data) : str(osp), d(data) {}
	virtual void invoke() { *str << d; }
};

// The put_target class represents the put() method
template<typename STREAM>
class put_target : public iotarget {
  private:
	STREAM *str;
	typename STREAM::char_type c;
  public:
	put_target(STREAM *osp, typename STREAM::char_type ch) : str(osp), c(ch) {}
	virtual void invoke() { str->put(c); }
};

// The write_target class represents the write() method
template<typename STREAM>
class write_target : public iotarget {
  private:
	STREAM *str;
	const typename STREAM::char_type *cp;
	std::streamsize siz;
  public:
	write_target(STREAM *osp, const typename STREAM::char_type *s, std::streamsize n)
		: str(osp), cp(s), siz(n) {}
	virtual void invoke() { str->write(cp,siz); }
};


// The flush_target class represents the flush() method
template<typename STREAM>
class flush_target : public iotarget {
  private:
	STREAM *str;
  public:
	flush_target(STREAM *osp) : str(osp) {}
	virtual void invoke() { str->flush(); }
};

// ***********************************************************
// End of iotarget subclasses
// ***********************************************************

// The event is a handle on which a thread waits for its
// subthreads to complete operation

typedef bool* event;

class super_io_operation;

// The half_io class represents either the input or the output
// half of the super_io_operation class

class half_io {
	friend class astream;
	friend class super_io_operation;

  private:
	super_io_operation *container;	// super_io_operation containing this half_io
	io_direction dir;			  	// Is this half_io for input or output?
	bool is_running;				// True if the operation is being executed
	event iostart();				// Start pending asynchronous I/O operations
	void cleanup();					// Clean up after a completed I/O operation
	void set_running() { is_running=true; }
	void clear_running() { is_running=false; }

  public:
	half_io();						// Constructor
	virtual ~half_io();				// Destructor
	void test_running();			// Test if half_io is doing I/O, and throw exception if yes
	std::vector<iotarget*> vec;		// Vectors of pending I/O operations
	pthread_t threadid;				// Thread identifier
	critical_section abort_reserve;	// Critical section mutex
	bool may_abort, do_abort;
	event event_id;					// Handle to asyncronous I/O events
	bool hasexcep;					// Did an exception occur asynchronously
	std::string excepstr;			// What-string of exception that occurred asynchronously
};


// The super_io_operation class is a superclass for all
// io_operation<...> classes

class super_io_operation {
	friend class astream;

  protected:
	super_io_operation();					// Constructor

  public:
	virtual ~super_io_operation() {}

	half_io hio[2];							// Input and output parts of this I/O operation
	virtual std::ios_base* get_mystr() = 0; // Fetch the underlying stream
};


// The io_operation class is the target for all asynchronous I/O.
// An io_operation object is used as the left side of the << and
// >> operators. An object of this type is created by using the ()
// operator on an astream object.

template <typename STREAM>
class io_operation : public super_io_operation {
  private:
	typedef typename STREAM::char_type char_type;	// Character type of I/O stream
	typedef typename STREAM::int_type int_type;		// Integer type of I/O stream
	STREAM *mystr;									// I/O stream

  public:
	io_operation(STREAM *str) : mystr(str) {}		// Constructor

	virtual std::ios_base* get_mystr() { return mystr; }	// Access function for get_mystr

	// Asynchronous version of put method
	io_operation<STREAM>& put(char_type c) {
		hio[output].test_running();
		hio[output].vec.push_back(new put_target<STREAM>(mystr,c));
		return *this;
	}

	// Asynchronous version of write method
	io_operation<STREAM>& write(const char_type* s, std::streamsize n) {
		hio[output].test_running();
		hio[output].vec.push_back(new write_target<STREAM>(mystr,s,n));
		return *this;
	}

	// Asynchronous version of flush method
	io_operation<STREAM>& flush() {
		hio[output].test_running();
		hio[output].vec.push_back(new flush_target<STREAM>(mystr));
		return *this;
	}


	// Asynchronous version of get(char_type&) method
	io_operation<STREAM>& get(char_type& c) {
		hio[input].test_running();
		hio[input].vec.push_back(new get1_target<STREAM>(mystr,&c));
		return *this;
	}


	// Asynchronous version of get(char_type*,streamsize) method
	io_operation<STREAM>& get(char_type* s, std::streamsize n) {
		hio[input].test_running();
		hio[input].vec.push_back(new get2_target<STREAM>(mystr,s,n));
		return *this;
	}

	// Asynchronous version of get(char_type*,streamsize,char_type) method
	io_operation<STREAM>& get(char_type* s, std::streamsize n, char_type delim) {
		hio[input].test_running();
		hio[input].vec.push_back(new get3_target<STREAM>(mystr,s,n,delim));
		return *this;
	}

	// Asynchronous version of get(basic_streambuf<...>&) method
	io_operation<STREAM>& get(
		std::basic_streambuf<typename STREAM::char_type, typename STREAM::traits_type>& sb) {
		hio[input].test_running();
		hio[input].vec.push_back(new get4_target<STREAM>(mystr,&sb));
		return *this;
	}

	// Asynchronous version of get(basic_streambuf<...>&,char_type) method
	io_operation<STREAM>& get(
		std::basic_streambuf<typename STREAM::char_type, typename STREAM::traits_type>& sb,
		char_type delim) {
		hio[input].test_running();
		hio[input].vec.push_back(new get5_target<STREAM>(mystr,&sb,delim));
		return *this;
	}

	// Asynchronous version of getline(char_type*,streamsize) method
	io_operation<STREAM>& getline(char_type* s, std::streamsize n) {
		hio[input].test_running();
		hio[input].vec.push_back(new getline1_target<STREAM>(mystr,s,n));
		return *this;
	}

	// Asynchronous version of getline(char_type*,streamsize,char_type) method
	io_operation<STREAM>& getline(char_type* s, std::streamsize n, char_type delim) {
		hio[input].test_running();
		hio[input].vec.push_back(new getline2_target<STREAM>(mystr,s,n,delim));
		return *this;
	}

	// Asynchronous version of ignore method
	io_operation<STREAM>& ignore(std::streamsize n, int_type delim) {
		hio[input].test_running();
		hio[input].vec.push_back(new ignore_target<STREAM>(mystr,n,delim));
		return *this;
	}

	// Asynchronous version of peek method
	io_operation<STREAM>& peek(int_type& c) {
		hio[input].test_running();
		hio[input].vec.push_back(new peek_target<STREAM>(mystr,&c));
		return *this;
	}

	// Asynchronous version of read method
	io_operation<STREAM>& read(char_type* s, std::streamsize n) {
		hio[input].test_running();
		hio[input].vec.push_back(new read_target<STREAM>(mystr,s,n));
		return *this;
	}


	// Asynchronous version of >> operator
	template<typename DATATYPE>
	io_operation<STREAM>& operator>>(DATATYPE& data) {
		hio[input].test_running();
		hio[input].vec.push_back(new input_target<STREAM,DATATYPE>(mystr,&data));
		return *this;
	}

	// Asynchronous version of << operator
	template<typename DATATYPE>
	io_operation<STREAM>& operator<<(DATATYPE data) {
		hio[output].test_running();
		hio[output].vec.push_back(new output_target<STREAM,DATATYPE>(mystr,data));
		return *this;
	}

	// The following additional definition of operator<< is really
	// superfluous because we already have the one above. However,
	// Microsoft Visual C++ has a bug that causes it to fail to match
	// the function above when doing something like '<< std::endl'.

  private:
	// my_ostream is a shorthand notation for the underlying
	// ostream used by the following << operator
	typedef std::basic_ostream<typename STREAM::char_type,typename STREAM::traits_type> my_ostream;

  public:
	// The following additional definition of operator<< is required to match
	// the function above when doing something like '<< std::endl'.
	io_operation<STREAM>& operator<<(my_ostream& (*data)(my_ostream&)) {
		hio[output].test_running();
		hio[output].vec.push_back(new output_target<STREAM, my_ostream& (*)(my_ostream&)>(mystr,data));
		return *this;
	}
};

// The async_error exception is used to propagate exceptions from an
// asynchronous function to the main thread.

class async_error : public std::runtime_error {
	friend class astream;

  private:
	std::ios_base *bad_ios;		// Stream causing the exception
	async_error();			// Default constructor used only by astream

  public:
	async_error(const std::string& arg, std::ios_base *badios); // Constructor
	// GCC complains if this destructor does not exist
	virtual ~async_error() throw() {}
	// Return the failing I/O stream
	std::ios_base *whatios() const;
};



// Objects of the astream class identify sets of I/O operations that
// must be performed concurrently. Consider this sample code:
//		astream ast;
//		ast(cout) << "Hello" << " world";
//		ast(cerr) << "Hallo";
//		ast(cerr) << " Welt";
//		ast.wait();
// In this code "Hello" and " world" will be written concurrenly with
// "Hallo" and " Welt". However, "Hello" will be written before " world",
// and "Hallo" will be written before " Welt".

class astream {
  private:
	std::vector<super_io_operation*> iop; // I/O commands to be performed concurrently
	bool hasexcep;			// Did an exception occur asynchronously
	async_error excep;		// Exception which occurred asynchronously

  public:
	// The () operator creates an appropriate io_operation object for
	// the astream. If an I/O stream already has an io_operation
	// object associated with it, then that object is returned,
	// otherwise a new object is created. In this manner, if ast is an
	// astream, the first occurrence of ast(cout) will create a new
	// io_operation object, but subsequent occurrences of ast(cout)
	// will merely return the same io_operation object

	template <typename STREAM>
	io_operation<STREAM>& operator()(STREAM& i) {
		for (std::vector<super_io_operation*>::iterator it=iop.begin(); it!=iop.end(); ++it) {
			if ((*it)->get_mystr() == &i)
				return *(static_cast<io_operation<STREAM>*>(*it));
		}

		io_operation<STREAM> *inp = new io_operation<STREAM>(&i);
		iop.push_back(inp);
		return *inp;
	}

	// The start() method starts all pending I/O operations
	void start();

	// The wait() method waits for one or all pending I/O operations
	// to finish. Asynchronous I/O operations that are not running,
	// will be started by this method. If its wait_type parameter is
	// wait_for_one, the method returns a pointer to a single I/O
	// stream that has completed its operation. If the parameter is
	// wait_for_all, wait() waits for all pending I/O operations to
	// complete and returns 0. If an exception has occurred in an
	// asynchronous I/O opertions, it is rethrown by wait().

	enum wait_type {wait_for_one, wait_for_all};

	std::ios_base* wait(long timeout_ms=0, wait_type wait_for=wait_for_one,
						io_direction *dir=0);

	// The poll() method checks if a pending I/O operation has
	// finished. Asynchronous I/O operations that are not running,
	// will be started by this method. The method returns a pointer to
	// an I/O stream that has completed its operation. If no stream
	// has completed, 0 is returned. If an exception has occurred in
	// an asynchronous I/O opertions, it is rethrown by poll().

	std::ios_base* poll(io_direction *dir=0);

	// The pendio() method checks if pending (i.e., not-waited-for) I/O
	// operations exist. It returns the number of pending I/O operations.

	int pendio();

	// The abort() method cancels all pending I/O requests. Note that
	// the state of I/O streams may be undefined after this function
	// has been called.

	void abort();
};

#endif  // of __ASYNC_STREAM_H_INCLUDED__