~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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

 *

 * Original author: Paul McCullagh (H&G2JCtL)

 * 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.

 *

 */

#include "CSConfig.h"

#include <string.h>

#include <inttypes.h>

#include "CSMemory.h"

#include "CSStream.h"

#include "CSGlobal.h"

using namespace std;

/*

 * ---------------------------------------------------------------

 * STREAM UTILITIES

 */

void CSStream::pipe(CSOutputStream *out, CSInputStream *in)

{

	char	*buffer;

	size_t	size;

	enter_();

	push_(out);

	push_(in);

	buffer = (char *) cs_malloc(DEFAULT_BUFFER_SIZE);

	push_ptr_(buffer);

	for (;;) {

		size = in->read(buffer, DEFAULT_BUFFER_SIZE);

		self->interrupted();

		if (!size)

			break;

		out->write(buffer, size);

		self->interrupted();

	}

	in->close();

	out->close();

	release_(buffer);

	release_(in);

	release_(out);

	exit_();

}

/*

 * ---------------------------------------------------------------

 * INPUT STREAMS

 */

CSStringBuffer *CSInputStream::readLine()

{

	int				ch;

	CSStringBuffer	*sb = NULL;

	enter_();

	ch = read();

	if (ch != -1) {

		new_(sb, CSStringBuffer(20));

		push_(sb);

		while (ch != '\n' && ch != '\r' && ch != -1) {

			sb->append((char) ch);

			ch = read();

		}

		if (ch == '\r') {

			if (peek() == '\n')

				ch = read();

		}

		pop_(sb);

	}

	return_(sb);

}

/*

 * ---------------------------------------------------------------

 * OUTPUT STREAMS

 */

void CSOutputStream::printLine(const char *cstr)

{

	enter_();

	print(cstr);

	print(getEOL());

	flush();

	exit_();

}

void CSOutputStream::print(const char *cstr)

{

	enter_();

	write(cstr, strlen(cstr));

	exit_();

}

void CSOutputStream::print(CSString *s)

{

	enter_();

	print(s->getCString());

	exit_();

}

void CSOutputStream::print(int value)

{

	char buffer[20];

	snprintf(buffer, 20, "%d", value);

	print(buffer);

}

void CSOutputStream::print(uint64_t value)

{

	char buffer[30];

	snprintf(buffer, 30, "%"PRIu64"", value);

	print(buffer);

}

/*

 * ---------------------------------------------------------------

 * FILE INPUT STREAMS

 */

CSFileInputStream::~CSFileInputStream()

{

	if (iFile)

		iFile->release();

}

size_t CSFileInputStream::read(char *b, size_t len)

{

	size_t size;

	enter_();

	size = iFile->read(b, iReadOffset, len, 0);

	iReadOffset += size;

	return_(size);

}

int CSFileInputStream::read()

{

	size_t	size;

	char	ch;

	enter_();

	size = iFile->read(&ch, iReadOffset, 1, 0);

	iReadOffset += size;

	return_(size == 0 ? -1 : (int) ch);

}

void CSFileInputStream::reset() {iReadOffset = 0;}

int CSFileInputStream::peek()

{

	size_t	size;

	char	ch;

	enter_();

	size = iFile->read(&ch, iReadOffset, 1, 0);

	return_(size == 0 ? -1 : (int) ch);

}

void CSFileInputStream::close()

{

	enter_();

	iFile->close();

	exit_();

}

CSFileInputStream *CSFileInputStream::newStream(CSFile *f)

{

	CSFileInputStream *s;

	if (!(s = new CSFileInputStream())) {

		f->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iFile = f;

	return s;

}

CSFileInputStream *CSFileInputStream::newStream(CSFile *f, off64_t offset)

{

	CSFileInputStream *s;

	if (!(s = new CSFileInputStream())) {

		f->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iFile = f;

	s->iReadOffset = offset;

	return s;

}

/*

 * ---------------------------------------------------------------

 * FILE OUTPUT STREAMS

 */

CSFileOutputStream::~CSFileOutputStream()

{

	if (iFile)

		iFile->release();

}

void CSFileOutputStream::write(const char *b, size_t len)

{

	enter_();

	iFile->write(b, iWriteOffset, len);

	iWriteOffset += len;

	exit_();

}

const char *CSFileOutputStream::getEOL()

{

	enter_();

	return_(iFile->getEOL());

}

void CSFileOutputStream::flush()

{

	enter_();

	iFile->flush();

	exit_();

}

void CSFileOutputStream::write(char b)

{

	enter_();

	iFile->write(&b, iWriteOffset, 1);

	iWriteOffset += 1;

	exit_();

}

void CSFileOutputStream::reset() {iWriteOffset = 0;}

void CSFileOutputStream::close()

{

	enter_();

	iFile->close();

	exit_();

}

CSFileOutputStream *CSFileOutputStream::newStream(CSFile *f)

{

	CSFileOutputStream *s;

	if (!(s = new CSFileOutputStream())) {

		f->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iFile = f;

	return  s;

}

CSFileOutputStream *CSFileOutputStream::newStream(CSFile *f, off64_t offset)

{

	CSFileOutputStream *s;

	if (!(s = new CSFileOutputStream())) {

		f->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iFile = f;

	s->iWriteOffset = offset;

	return  s;

}

/*

 * ---------------------------------------------------------------

 * SOCKET INPUT STREAMS

 */

CSSocketInputStream::~CSSocketInputStream()

{

	if (iSocket)

		iSocket->release();

}

void CSSocketInputStream::close()

{

	enter_();

	iSocket->close();

	exit_();

}

size_t CSSocketInputStream::read(char *b, size_t len)

{

	enter_();

	return_(iSocket->read(b, len));

}

int CSSocketInputStream::read()

{

	enter_();

	return_(iSocket->read());

}

int CSSocketInputStream::peek()

{

	enter_();

	return_(iSocket->peek());

}

void CSSocketInputStream::reset()

{

	enter_();

	CSException::throwException(CS_CONTEXT, CS_ERR_OPERATION_NOT_SUPPORTED, "CSSocketInputStream::reset() not supported");

	exit_();

}

CSSocketInputStream *CSSocketInputStream::newStream(CSSocket *s)

{

	CSSocketInputStream *str;

	if (!(str = new CSSocketInputStream())) {

		s->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	str->iSocket = s;

	return  str;

}

/*

 * ---------------------------------------------------------------

 * SOCKET OUTPUT STREAMS

 */

CSSocketOutputStream::~CSSocketOutputStream()

{

	if (iSocket)

		iSocket->release();

}

void CSSocketOutputStream::close()

{

	enter_();

	iSocket->close();

	exit_();

}

void CSSocketOutputStream::write(const char *b, size_t len)

{

	enter_();

	iSocket->write(b, len);

	exit_();

}

void CSSocketOutputStream::flush()

{

	enter_();

	iSocket->flush();

	exit_();

}

void CSSocketOutputStream::write(char b)

{

	enter_();

	iSocket->write(b);

	exit_();

}

void CSSocketOutputStream::reset()

{

	enter_();

	CSException::throwException(CS_CONTEXT, CS_ERR_OPERATION_NOT_SUPPORTED, "CSSocketOutputStream::reset() not supported");

	exit_();

}

CSSocketOutputStream *CSSocketOutputStream::newStream(CSSocket *s)

{

	CSSocketOutputStream *str;

	if (!(str = new CSSocketOutputStream())) {

		s->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	str->iSocket = s;

	return  str;

}

/*

 * ---------------------------------------------------------------

 * BUFFERED INPUT STREAMS

 */

CSBufferedInputStream::~CSBufferedInputStream()

{

	if (iStream)

		iStream->release();

}

void CSBufferedInputStream::close()

{

	enter_();

	iStream->close();

	exit_();

}

size_t CSBufferedInputStream::read(char *b, size_t len)

{

	size_t tfer;

	enter_();

	if (iBuffPos < iBuffTotal) {

		tfer = iBuffTotal - iBuffPos;

		if (tfer > len)

			tfer = len;

		memcpy(b, iBuffer + iBuffPos, tfer);

		iBuffPos += tfer;

	}

	else

		tfer = iStream->read(b, len);

	return_(tfer);

}

int CSBufferedInputStream::read()

{

	int ch;

	enter_();

	if (iBuffPos == iBuffTotal) {

		iBuffTotal = iStream->read((char *) iBuffer, CS_STREAM_BUFFER_SIZE);

		iBuffPos = 0;

	}

	if (iBuffPos < iBuffTotal) {

		ch = iBuffer[iBuffPos];

		iBuffPos++;

	}

	else

		ch = -1;

	return_(ch);

}

int CSBufferedInputStream::peek()

{

	int ch;

	enter_();

	if (iBuffPos == iBuffTotal) {

		iBuffTotal = iStream->read((char *) iBuffer, CS_STREAM_BUFFER_SIZE);

		iBuffPos = 0;

	}

	if (iBuffPos < iBuffTotal)

		ch = iBuffer[iBuffPos];

	else

		ch = -1;

	return_(ch);

}

void CSBufferedInputStream::reset()

{

	iBuffPos = iBuffTotal =0;

	iStream->reset();

}

CSBufferedInputStream *CSBufferedInputStream::newStream(CSInputStream* i)

{

	CSBufferedInputStream *s;

	if (!(s = new CSBufferedInputStream())) {

		i->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iStream = i;

	return  s;

}

/*

 * ---------------------------------------------------------------

 * BUFFERED OUTPUT STREAMS

 */

CSBufferedOutputStream::~CSBufferedOutputStream()

{

	if (iStream)

		iStream->release();

}

void CSBufferedOutputStream::close()

{

	enter_();

	iStream->close();

	exit_();

}

void CSBufferedOutputStream::write(const char *b, size_t len)

{

	size_t tfer;

	enter_();

	if (iBuffTotal < CS_STREAM_BUFFER_SIZE) {

		tfer = CS_STREAM_BUFFER_SIZE - iBuffTotal;

		if (tfer > len)

			tfer = len;

		memcpy(iBuffer + iBuffTotal, b, tfer);

		iBuffTotal += tfer;

		b += tfer;

		len -= tfer;

	}

	if (len > 0) {

		flush();

		if (len > CS_STREAM_BUFFER_SIZE)

			iStream->write(b, len);

		else {

			memcpy(iBuffer, b, len);

			iBuffTotal = len;

		}

	}

	exit_();

}

void CSBufferedOutputStream::flush()

{

	enter_();

	if (iBuffTotal > 0) {

		iStream->write((char *) iBuffer, iBuffTotal);

		iBuffTotal = 0;

	}

	exit_();

}

void CSBufferedOutputStream::write(char b)

{

	enter_();

	if (iBuffTotal == CS_STREAM_BUFFER_SIZE)

		flush();

	iBuffer[iBuffTotal] = b;

	iBuffTotal++;

	exit_();

}

void CSBufferedOutputStream::reset()

{

	iBuffTotal = 0;

	iStream->reset();

}

CSBufferedOutputStream *CSBufferedOutputStream::newStream(CSOutputStream* i)

{

	CSBufferedOutputStream *s;

	if (!(s = new CSBufferedOutputStream())) {

		i->release();

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iStream = i;

	return s;

}

/*

 * ---------------------------------------------------------------

 * MEMORY INPUT STREAMS

 */

CSMemoryInputStream *CSMemoryInputStream::newStream(const u_char* buffer, uint32_t length)

{

	CSMemoryInputStream *s;

	if (!(s = new CSMemoryInputStream())) {

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iMemory = buffer;

	s->iMemTotal = length;

	return s;

}

CSMemoryOutputStream *CSMemoryOutputStream::newStream(size_t init_length, size_t min_alloc)

{

	CSMemoryOutputStream *s;

	if (!(s = new CSMemoryOutputStream())) {

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->iMemory = (u_char *) cs_malloc(init_length);

	s->iMemTotal = init_length;

	s->iMemSpace = init_length;

	s->iMemPos = s->iMemory;

	s->iMemMin = min_alloc;

	return s;

}

CSMemoryOutputStream::~CSMemoryOutputStream()

{

	if (iMemory)

		cs_free(iMemory);

}

void CSMemoryOutputStream::write(const char *b, size_t len)

{

	if (iMemSpace < len) {

		size_t new_size = iMemTotal + ((len < iMemMin)? iMemMin:len);

		cs_realloc((void**) &iMemory, new_size);

		iMemPos = iMemory + (iMemTotal - iMemSpace);

		iMemSpace += (new_size - iMemTotal);

		iMemTotal = new_size;		

	}

	memcpy(iMemPos, b, len);

	iMemPos +=len;

	iMemSpace -= len;

}

void CSMemoryOutputStream::write(const char b)

{

	if (!iMemSpace) {

		cs_realloc((void**) &iMemory, iMemTotal + iMemMin);

		iMemPos = iMemory + iMemTotal;

		iMemSpace += iMemMin;

		iMemTotal += iMemMin;		

	}

	*iMemPos = b;

	iMemPos++;

	iMemSpace--;

}

void CSMemoryOutputStream::reset()

{

	iMemPos = iMemory;

	iMemSpace = iMemTotal;

}

/*

 * ---------------------------------------------------------------

 * STATIC (user) MEMORY OUTPUT STREAM

 */

void CSStaticMemoryOutputStream::write(const char *b, size_t len)

{

	if (iMemSpace < len) {

		enter_();

		CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "CSStaticMemoryOutputStream: overflow");

		exit_();

	}

	memcpy(iMemPos, b, len);

	iMemPos +=len;

	iMemSpace -= len;

}

void CSStaticMemoryOutputStream::write(const char b)

{

	if (!iMemSpace) {

		enter_();

		CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "CSStaticMemoryOutputStream: overflow");

		exit_();

	}

	*iMemPos = b;

	iMemPos++;

	iMemSpace--;

}

/*

 * ---------------------------------------------------------------

 * Callback InPUT STREAM

 */

CSCallbackInputStream *CSCallbackInputStream::newStream(CSStreamReadCallbackFunc callback, void *user_data)

{

	CSCallbackInputStream *s;

	if (!(s = new CSCallbackInputStream())) {

		CSException::throwOSError(CS_CONTEXT, ENOMEM);

	}

	s->callback = callback;

	s->cb_data = user_data;

	return s;

}