streams.mx

一个内存数据库的源代码这是服务器端还有客户端

@' The contents of this file are subject to the MonetDB Public
@' Version 1.1 (the "License"); you may not use this file except in
@' compliance with the License. You may obtain a copy of the License at
@' http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
@'
@' Software distributed under the License is distributed on an "AS IS"
@' basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
@' License for the specific language governing rights and limitations
@' under the License.
@'
@' The Original Code is the MonetDB Database System.
@'
@' The Initial Developer of the Original Code is CWI.
@' Portions created by CWI are Copyright (C) 1997-2007 CWI.
@' All Rights Reserved.

@f streams
@a Niels Nes
@* A simple interface to IO streams
All file IO is funneled through the stream library, which
guarantees cross-platform capabilities.
Several protocols are provided, e.g. it can be used to 
open 'non compressed, gzipped, bzipped' data files. 
It encapsulates the corresponding library managed in
.../stream.

@mal
module streams;

atom streams:ptr;

command openReadBytes{unsafe}( filename:str ) :streams
address stream_open_rstreamwrap
comment "open a file stream for reading";
command openWriteBytes{unsafe}( filename:str ) :streams
address stream_open_wstreamwrap
comment "open a file stream for writing";

command openRead{unsafe}( filename:str ) :streams
address stream_open_rastreamwrap
comment "open ascii file stream for reading";
command openWrite{unsafe}( filename:str ) :streams
address stream_open_wastreamwrap
comment "open ascii file stream for writing";

command openRead{unsafe}(s:streams):streams
address stream_stream_rstreamwrap
comment "convert an ascii stream to binary";
command openWrite{unsafe}(s:streams):streams
address stream_stream_wstreamwrap
comment "convert an ascii stream to binary";

command socketReadBytes{unsafe}(skt:int,name:str):streams
address stream_socket_rstreamwrap
comment "open a socket stream for reading";
command socketWriteBytes{unsafe}(skt:int,name:str):streams
address stream_socket_wstreamwrap
comment "open a socket stream for writing";
command socketRead{unsafe}(skt:int,name:str):streams
address stream_socket_rastreamwrap
comment "open ascii socket stream for reading";
command socketWrite{unsafe}(skt:int,name:str):streams
address stream_socket_wastreamwrap
comment "open ascii socket stream for writing";

command blocked{unsafe}( s:streams ) :streams
address open_block_streamwrap
comment "open a block based stream";

command writeStr{unsafe}( s:streams, data:str ):void
address stream_write_stringwrap
comment "write data on the stream";
command writeInt{unsafe}( s:streams, data:int ):void
address stream_writeIntwrap
comment "write data on the stream";
command readStr{unsafe}( s:streams):str
address stream_read_stringwrap
comment "read string data from the stream";
command readInt{unsafe}( s:streams):int
address stream_readIntwrap
comment "read integer data from the stream";

command flush{unsafe}( s:streams )
address stream_flush_streamwrap
comment "flush the stream";

command close{unsafe}( s:streams )
address stream_close_streamwrap
comment "close and destroy the stream s";

atom bstream:ptr;
command create{unsafe}(s:streams, bufsize:int):bstream
address bstream_create_wrapwrap
comment "create a buffered stream";

command destroy{unsafe}(s:bstream)
address bstream_destroy_wrapwrap
comment "destroy bstream";

command read{unsafe}(s:bstream,size:int):int
address bstream_read_wrapwrap
comment "read at least size bytes into the buffer of s";

@-
@{
The implementation follows and should be self-documenting
@h
#ifndef _STREAMS_H_
#define _STREAMS_H_

#include <gdk.h>
#include <blob.h>

typedef ptr Stream;
typedef ptr Bstream;

#endif /*_STREAMS_H_*/
@c
#include "mal_config.h"
#include <mal.h>
#include "streams.h"
#include <stdio.h>

/*
int io_stdin (Stream *ret) { *(stream**)ret = GDKin;  return GDK_SUCCEED; }
int io_stdout(Stream *ret) { *(stream**)ret = GDKout; return GDK_SUCCEED; }
int io_stderr(Stream *ret) { *(stream**)ret = GDKerr; return GDK_SUCCEED; }
*/

#ifdef WIN32
#ifndef LIBSTREAMS
#define streams_export extern __declspec(dllimport)
#else
#define streams_export extern __declspec(dllexport)
#endif
#else
#define streams_export extern
#endif

streams_export int stream_write_string(Stream *S, str data);
streams_export int stream_writeInt_wrap(Stream *S, int *data);
streams_export int stream_readInt_wrap(int *data, Stream *S);
streams_export int stream_read_string(str *res, Stream *S);
streams_export int stream_flush_stream(Stream *S);
streams_export int stream_close_stream(Stream *S);
streams_export int open_block_stream(Stream *S, Stream *is);
streams_export int bstream_create_wrap(Bstream *BS, Stream *S, int *bufsize);
streams_export int bstream_destroy_wrap(Bstream *BS);
streams_export int bstream_read_wrap(int *res, Bstream *BS, int *size);
streams_export str stream_write_stringwrap(int *ret, Stream *S, str *data);
streams_export str stream_writeIntwrap(int *ret, Stream *S, int *data);
streams_export str stream_readIntwrap(int *ret, Stream *S);
streams_export str stream_read_stringwrap(str *res, Stream *s);
streams_export str stream_flush_streamwrap(int *ret, Stream *s);
streams_export str stream_close_streamwrap(int *ret, Stream *s);
streams_export str open_block_streamwrap(Stream *S, Stream *is);
streams_export str bstream_create_wrapwrap(Bstream *Bs, Stream *S, int *bufsize);
streams_export str bstream_destroy_wrapwrap(int *ret, Bstream *BS);
streams_export str bstream_read_wrapwrap(int *res, Bstream *BS, int *size);
streams_export str stream_readIntwrap(int *ret, Stream *S);
streams_export str stream_read_stringwrap(str *res, Stream *s);

@= open_stream
streams_export int stream_open_@1(Stream *S, str filename);
int
stream_open_@1(Stream *S, str filename)
{
	stream *s;

	if ((s = open_@1( filename )) == NULL || stream_errnr(s)) {
		if (s)
			stream_destroy(s);
		GDKerror("stream_open: could not open file '%s'\n", filename);
		return GDK_FAIL;
	} else {
		*(stream**)S = s;
		return GDK_SUCCEED;
	}
}
@c
@:open_stream(rstream)@
@:open_stream(wstream)@
@:open_stream(rastream)@
@:open_stream(wastream)@

@= stream_stream
streams_export int stream_stream_@1(Stream *Sout, Stream *Sin);
int
stream_stream_@1(Stream *Sout, Stream *Sin)
{
	*(stream**)Sout = stream_@1(*(stream**)Sin);
	return GDK_SUCCEED;
}
@c
@:stream_stream(rstream)@
@:stream_stream(wstream)@

@= open_socket
streams_export int stream_socket_@1(Stream *S, int *socket, str name);
int
stream_socket_@1(Stream *S, int *socket, str name)
{
	stream *s;

	if ((s = socket_@1( *socket, name )) == NULL || stream_errnr(s)) {
		if (s)
			stream_destroy(s);
		GDKerror("Could not open socket %s\n", name );
		return GDK_FAIL;
	} else {
		*(stream**)S = s;
		return GDK_SUCCEED;
	}
}
@c
@:open_socket(rstream)@
@:open_socket(wstream)@
@:open_socket(rastream)@
@:open_socket(wastream)@

@c
int
stream_write_string(Stream *S, str data)
{
	stream *s = *(stream **) S;

	return stream_write(s, data, 1, strlen(data)) < 0 ? GDK_FAIL : GDK_SUCCEED;
}

int
stream_writeInt_wrap(Stream *S, int *data)
{
	stream *s = *(stream **) S;

	return stream_writeInt(s, *data) ? GDK_SUCCEED : GDK_FAIL;
}

int
stream_readInt_wrap(int *data, Stream *S)
{
	stream *s = *(stream **) S;

	return stream_readInt(s, data) ? GDK_SUCCEED : GDK_FAIL;
}

#define CHUNK (64*1024)
int
stream_read_string(str *res, Stream *S)
{
	stream *s = *(stream **) S;
	ssize_t len = 0;
	size_t size = CHUNK +1;
	char *buf = GDKmalloc(size), *start = buf;

	while ((len = stream_read(s, start, 1, CHUNK)) > 0) {
		size += len;
		buf = GDKrealloc(buf, size);
		start = buf + size - CHUNK -1;

		*start = '\0';
	}
	if (len < 0)
		return GDK_FAIL;
	start += len;
	*start = '\0';
	*res = buf;
	return GDK_SUCCEED;
}

int
stream_flush_stream(Stream *S)
{
	stream *s = *(stream **) S;

	if (stream_flush(s))
		return GDK_FAIL;
	return GDK_SUCCEED;
}

int
stream_close_stream(Stream *S)
{
	close_stream(*(stream **) S);
	return GDK_SUCCEED;
}

int
open_block_stream(Stream *S, Stream *is)
{
	if ((*(stream **) S = block_stream(*(stream **) is)) == NULL) {
		return GDK_FAIL;
	} else {
		return GDK_SUCCEED;
	}
}

int
bstream_create_wrap(Bstream *BS, Stream *S, int *bufsize)
{
	if ((*(bstream **) BS = bstream_create(*(stream **) S, (size_t) * bufsize)) == NULL) {
		return GDK_FAIL;
	} else {
		return GDK_SUCCEED;
	}
}

int
bstream_destroy_wrap(Bstream *BS)
{
	bstream_destroy(*(bstream **) BS);
	return GDK_SUCCEED;
}

int
bstream_read_wrap(int *res, Bstream *BS, int *size)
{
	*res = (int) bstream_read(*(bstream **) BS, (size_t) * size);
	return GDK_SUCCEED;
}

@- Wrapper
The remainder is the version 5 wrapper
@= open_streamwrap
streams_export str stream_open_@1wrap( Stream *S, str *filename );
str stream_open_@1wrap( Stream *S, str *filename ){
	if( stream_open_@1(S, *filename)== GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.open", "Could not open file");
}
@}
@c
#include "mal_exception.h"

@:open_streamwrap(rstream)@
@:open_streamwrap(wstream)@
@:open_streamwrap(rastream)@
@:open_streamwrap(wastream)@

@-
@= stream_streamwrap
streams_export str stream_stream_@1wrap(Stream *sout, Stream *sin);
str stream_stream_@1wrap(Stream *sout, Stream *sin){
	*(stream**)sout = stream_@1(*(stream**)sin);
	return MAL_SUCCEED;
}
@c
@:stream_streamwrap(rstream)@
@:stream_streamwrap(wstream)@
@-
@= open_socketwrap
streams_export str stream_socket_@1wrap( Stream *S, int *socket, str *name );
str stream_socket_@1wrap( Stream *S, int *socket, str *name ){
	if( stream_socket_@1(S,socket,*name) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.socket","Could not open socket");
}
@c
@:open_socketwrap(rstream)@
@:open_socketwrap(wstream)@
@:open_socketwrap(rastream)@
@:open_socketwrap(wastream)@

str
stream_write_stringwrap(int *ret, Stream *S, str *data)
{
	(void) ret;
	if (stream_write_string(S, *data) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.writeStr", "Failed to write string");
}

str
stream_writeIntwrap(int *ret, Stream *S, int *data)
{
	(void) ret;
	if (stream_writeInt_wrap(S, data) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.writeInt", "Failed to write int");
}

str
stream_readIntwrap(int *ret, Stream *S)
{
	if (stream_readInt_wrap(ret, S) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.readInt", "Failed to read int");
}

str
stream_read_stringwrap(str *res, Stream *s)
{
	if (stream_read_string(res, s) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.readStr", "Failed to read string");
}

str
stream_flush_streamwrap(int *ret, Stream *s)
{
	(void) ret;
	if (stream_flush_stream(s) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.flush", "Failed to flush a stream");
}

str
stream_close_streamwrap(int *ret, Stream *s)
{
	(void) ret;
	if (stream_close_stream(s) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "streams.close", "Failed to close a stream");
}

str
open_block_streamwrap(Stream *S, Stream *is)
{
	if (open_block_stream(S, is) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "bstreams.open", "Failed to open a blocked stream");
}

str
bstream_create_wrapwrap(Bstream *Bs, Stream *S, int *bufsize)
{
	if (bstream_create_wrap(Bs, S, bufsize) == GDK_SUCCEED)
		return MAL_SUCCEED;
	throw(IO, "bstreams.create", "Failed to create a blocked stream");
}

str
bstream_destroy_wrapwrap(int *ret, Bstream *BS)
{
	(void) ret;
	bstream_destroy_wrap(BS);
	return MAL_SUCCEED;
}

str
bstream_read_wrapwrap(int *res, Bstream *BS, int *size)
{
	bstream_read_wrap(res, BS, size);
	return MAL_SUCCEED;
}