~drizzle-trunk/drizzle/development

/* Copyright (c) 2008 PrimeBase Technologies GmbH, Germany
 *
 * PrimeBase Media Stream for MySQL
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
 *
 * Original Author: Paul McCullagh
 * Continued development: Barry Leslie
 *
 * 2007-06-07
 *
 * CORE SYSTEM:
 * Basic input and output streams.
 *
 * These objects wrap the system streams, and simplify things.
 * I also want to standardize exceptions and implement
 * socket based streams.
 *
 */

#ifndef __CSSTREAM_H__
#define __CSSTREAM_H__

#define DEFAULT_BUFFER_SIZE		64000

#include "CSDefs.h"
#include "CSFile.h"
#include "CSSocket.h"

class CSInputStream : public CSRefObject {
public:
	CSInputStream() { }
	virtual ~CSInputStream() { }

	/*
	 * Closes this input stream and releases any system
	 * resources associated  with the stream.
	 */
	virtual void close() = 0;

	/*
	 * Reads up to len bytes of data from the input stream into an
	 * array of bytes. Return the number of bytes read.
	 *
	 * Zero will be returned in the case of EOF.
	 *
	 * This call blocks until at least one byte is
	 * returned.
	 */
	virtual size_t read(char *b, size_t len) = 0;

	/* Reads the next byte of data from the input stream.
	 * Returns -1 if EOF.
	 */
	virtual int read() = 0;

	/* Read one character ahead. */
	virtual int peek() = 0;

	/*
	 *  Reset this output stream to the start inorder to restart the write.
	 */
	virtual void reset() = 0; 

	/* Return the name of the file, or whatever: */
	virtual const char *identify() = 0;

	/*
	 * Read a line from the input stream. This function
	 * handles all types of line endings. The function
	 * return NULL on EOF.
	 */
	CSStringBuffer *readLine();
};

class CSOutputStream : public CSRefObject {
public:
 	/*
	 * Closes this input stream and releases any system
	 * resources associated  with the stream.
	 */
	virtual void close() = 0;

	/*
	 * Writes len bytes from the specified byte array starting at
	 * offset off to this output stream.
	 */
	virtual void write(const char *b, size_t len) = 0;

	/*
	 * Returns the default EOL indicator.
	 * Will be \n, \r or \r\n.
	 */
	virtual const char *getEOL() = 0;

	/*
	 *  Flushes this output stream and forces any buffered output
	 * bytes to be written out.
	 */
	virtual void flush() = 0;

	/*
	 * Writes the specified byte to this output stream.
	 */
	virtual void write(char b) = 0; 

	/*
	 * Reset this output stream to the start inorder to restart the write.
	 */
	virtual void reset() = 0; 

	virtual const char *identify() = 0;

	/*
	 * Write a line. Terminator is specific to the
	 * type of output stream and may depend on the
	 * current platform.
	 */
	void printLine(const char *cstr);

	/*
	 * Various write types:
	 */
	virtual void print(const char *cstr);
	virtual void print(CSString *s);
	virtual void print(int value);
	virtual void print(uint64_t value);
};

class CSStream : public CSObject {
public:
	static void pipe(CSOutputStream *out, CSInputStream *in);
};

/* File Stream: */

class CSFileInputStream : public CSInputStream {
public:
	CSFileInputStream(): iFile(NULL), iReadOffset(0) { }
	virtual ~CSFileInputStream();

	virtual void close();

	virtual size_t read(char *b, size_t len);

	virtual int read();

	virtual int peek();

	/*
	 *  Reset this output stream to the start inorder to restart the read.
	 */
	virtual void reset(); 

	virtual const char *identify();

	static CSFileInputStream *newStream(CSFile* f);
	static CSFileInputStream *newStream(CSFile* f, off64_t offset);

private:
	CSFile	*iFile;
	off64_t	iReadOffset;
};

class CSFileOutputStream : public CSOutputStream {
public:
	CSFileOutputStream(): iFile(NULL), iWriteOffset(0) { }
	virtual ~CSFileOutputStream();

	virtual void close();

	virtual void write(const char *b, size_t len);

	virtual const char *getEOL();

	virtual void flush();

	virtual void write(char b);

	virtual void reset(); 

	virtual const char *identify();

	static CSFileOutputStream *newStream(CSFile* f);
	static CSFileOutputStream *newStream(CSFile* f, off64_t offset);

private:
	CSFile	*iFile;
	off64_t	iWriteOffset;
};

/* Socket Stream */

class CSSocketInputStream : public CSInputStream {
public:
	CSSocketInputStream(): iSocket(NULL) { }
	virtual ~CSSocketInputStream();

	virtual void close();

	virtual size_t read(char *b, size_t len);

	virtual int read();

	virtual int peek();

	virtual void reset(); 

	virtual const char *identify();

	static CSSocketInputStream *newStream(CSSocket *s);

private:
	CSSocket* iSocket;
};

class CSSocketOutputStream : public CSOutputStream {
public:
	CSSocketOutputStream(): iSocket(NULL) { }
	virtual ~CSSocketOutputStream();

	virtual void close();

	virtual void write(const char *b, size_t len);

	virtual const char *getEOL() { return "\n"; };

	virtual void flush();

	virtual void write(char b);

	virtual void reset(); 

	virtual const char *identify();

	static CSSocketOutputStream *newStream(CSSocket *s);

private:
	CSSocket* iSocket;
};

/* Buffered Stream: */
#ifdef DEBUG_disabled
#define CS_STREAM_BUFFER_SIZE			80
#else
#define CS_STREAM_BUFFER_SIZE			(32 * 1024)
#endif

class CSBufferedInputStream : public CSInputStream {
public:
	CSBufferedInputStream(): iStream(NULL), iBuffTotal(0), iBuffPos(0) { }
	virtual ~CSBufferedInputStream();

	virtual void close();

	virtual size_t read(char *b, size_t len);

	virtual int read();

	virtual int peek();

	virtual void reset(); 

	virtual const char *identify();

	static CSBufferedInputStream *newStream(CSInputStream* i);

private:
	CSInputStream* iStream;
	u_char iBuffer[CS_STREAM_BUFFER_SIZE];
	uint32_t iBuffTotal;
	uint32_t iBuffPos;
};

class CSBufferedOutputStream : public CSOutputStream {
public:
	CSBufferedOutputStream(): iStream(NULL), iBuffTotal(0) { }
	virtual ~CSBufferedOutputStream();

	virtual void close();

	virtual void write(const char *b, size_t len);

	virtual const char *getEOL() { return "\n"; };

	virtual void flush();

	virtual void write(char b);

	virtual void reset(); 

	virtual const char *identify();

	static CSBufferedOutputStream *newStream(CSOutputStream* i);

private:
	CSOutputStream* iStream;
	u_char iBuffer[CS_STREAM_BUFFER_SIZE];
	uint32_t iBuffTotal;
};

/* memory Stream */
class CSMemoryInputStream : public CSInputStream {
public:
	CSMemoryInputStream(): iMemory(NULL), iMemTotal(0), iMemPos(0) { }
	~CSMemoryInputStream(){}

	virtual void close() {}

	virtual size_t read(char *b, size_t len)
	{
		if (len > (iMemTotal - iMemPos))
			len = iMemTotal - iMemPos;
		
		memcpy(b, iMemory + iMemPos, len);
		iMemPos += len;	
		return len;
	}

	virtual int read()
	{
		int b = -1;
		if (iMemPos < iMemTotal) 
			b = iMemory[iMemPos++];
		return b;
	}

	virtual int peek()
	{
		int b = -1;
		if (iMemPos < iMemTotal) 
			b = iMemory[iMemPos];
		return b;
	}

	virtual void reset() {iMemPos = 0;}
	
	virtual const char *identify() 
	{
		return "memory stream";
	}

	static CSMemoryInputStream *newStream(const u_char* buffer, uint32_t length);

private:
	const u_char *iMemory;
	uint32_t iMemTotal;
	uint32_t iMemPos;
};


class CSMemoryOutputStream : public CSOutputStream {
public:
	CSMemoryOutputStream(): iMemory(NULL), iMemTotal(0), iMemSpace(0), iMemMin(0), iMemPos(NULL){ }
	virtual ~CSMemoryOutputStream();

	virtual void close() {}

	virtual void write(const char *b, size_t len);
	virtual const char *getEOL() { return "\n"; };

	virtual void flush() {}

	virtual void write(char b);

	const u_char *getMemory(size_t *len)
	{
		*len = iMemPos - iMemory;
		return iMemory;
	}
	
	virtual void reset();
	
	virtual const char *identify();
	
	static CSMemoryOutputStream *newStream(size_t init_length, size_t min_alloc);

private:
	u_char *iMemory;
	uint32_t iMemTotal;
	uint32_t iMemSpace;
	uint32_t iMemMin;
	u_char *iMemPos;
};

class CSStaticMemoryOutputStream : public CSOutputStream {
public:
	CSStaticMemoryOutputStream(u_char *mem, off64_t size): iMemory(mem), iMemSpace(size), iMemSize(size), iMemPos(mem){ }
	virtual ~CSStaticMemoryOutputStream() {}

	virtual void close() {}

	virtual void write(const char *b, size_t len);
	virtual const char *getEOL() { return "\n"; };

	virtual void flush() {}

	virtual void write(char b);
	
	virtual void reset() 
	{
		iMemPos = iMemory;
		iMemSpace = iMemSize;
	}
	
	virtual const char *identify() 
	{
		return "memory stream";
	}
	
	off64_t getSize() { return iMemPos - iMemory; }

private:
	u_char *iMemory;
	off64_t iMemSpace;
	off64_t iMemSize;
	u_char *iMemPos;
};

typedef size_t (* CSStreamReadCallbackFunc) (void *caller_data, char *buffer, size_t buffer_size, u_char reset);

class CSCallbackInputStream : public CSInputStream {
public:
	CSCallbackInputStream(): callback(NULL), cb_data(NULL), havePeek(false), doReset(false) { }
	~CSCallbackInputStream(){}

	virtual void close() {}

	virtual size_t read(char *b, size_t len)
	{
		size_t size = 0;
		
		if (havePeek) {
			havePeek = false;
			*b =  peek_char;
			b++; len--;
			if (len) {
				size = callback(cb_data, b, len, doReset);
			}
				
			size++;			
		} else
			size = callback(cb_data, b, len, doReset);
			
		if (doReset)
			doReset = false;
			
		return size;
	}

	virtual int read()
	{
		char c;
		
		if (havePeek) {
			havePeek = false;
			return peek_char;
		}
		if (!callback(cb_data, &c, 1, doReset))
			c = -1;
			
		if (doReset)
			doReset = false;
		
		return c;
	}

	virtual int peek()
	{
		if (!havePeek) {
			if (callback(cb_data, &peek_char, 1, doReset))
				havePeek = true;
			else
				return -1;
		}
		return peek_char;
	}

	virtual void reset() 
	{
		havePeek = false;
		doReset = false;
	}

	virtual const char *identify() 
	{
		return "callback stream";
	}

	static CSCallbackInputStream *newStream(CSStreamReadCallbackFunc callback, void *user_data);

private:
	CSStreamReadCallbackFunc callback;
	void *cb_data;
	char peek_char;
	bool havePeek;
	bool doReset;
};

#endif