split stream funcs out of input.c

4 files changed, 621 insertions, 574 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 4f2eb073..12497176 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -28,7 +28,7 @@ bin_PROGRAMS = colm
 noinst_PROGRAMS = bootstrap0 bootstrap1
 
 RUNTIME_SRC = \
-	map.c pdarun.c list.c input.c debug.c \
+	map.c pdarun.c list.c input.c stream.c debug.c \
 	codevect.c pool.c string.c tree.c iter.c \
 	bytecode.c program.c struct.c commit.c \
 	print.c
diff --git a/src/input.c b/src/input.c
index 6b9de337..80a6e028 100644
--- a/src/input.c
+++ b/src/input.c
@@ -37,8 +37,8 @@
 #include <colm/pool.h>
 #include <colm/struct.h>
 
-static struct stream_impl *colm_impl_consumed( char *name, int len );
-static struct stream_impl *colm_impl_new_text( char *name, const char *data, int len );
+struct stream_impl *colm_impl_consumed( char *name, int len );
+struct stream_impl *colm_impl_new_text( char *name, const char *data, int len );
 
 static bool is_tree( struct seq_buf *b )
 {
@@ -77,27 +77,9 @@ static struct seq_buf *new_seq_buf()
 	return rb;
 }
 
-struct run_buf *new_run_buf( int sz )
-{
-	struct run_buf *rb;
-	if ( sz > FSM_BUFSIZE ) {
-		int ssz = sizeof(struct run_buf) + sz - FSM_BUFSIZE;
-		rb = (struct run_buf*) malloc( ssz );
-		memset( rb, 0, ssz );
-	}
-	else {
-		rb = (struct run_buf*) malloc( sizeof(struct run_buf) );
-		memset( rb, 0, sizeof(struct run_buf) );
-	}
-	return rb;
-}
-
 DEF_INPUT_FUNCS( input_funcs_seq, input_impl_seq );
-DEF_STREAM_FUNCS( stream_funcs_data, stream_impl_data );
 
 extern struct input_funcs_seq input_funcs;
-extern struct stream_funcs_data file_funcs;
-extern struct stream_funcs_data accum_funcs;
 
 static bool loc_set( location_t *loc )
 {
@@ -120,13 +102,6 @@ void input_transfer_loc( struct colm_program *prg, location_t *loc, struct input
 	loc->byte = ss->byte;
 }
 
-static void data_transfer_loc( struct colm_program *prg, location_t *loc, struct stream_impl_data *ss )
-{
-	loc->name = ss->name;
-	loc->line = ss->line;
-	loc->column = ss->column;
-	loc->byte = ss->byte;
-}
 
 static bool call_destructor( struct seq_buf *buf )
 {
@@ -149,13 +124,6 @@ void colm_input_destroy( program_t *prg, tree_t **sp, struct_t *s )
 	si->funcs->destructor( prg, sp, si );
 }
 
-void colm_stream_destroy( program_t *prg, tree_t **sp, struct_t *s )
-{
-	stream_t *stream = (stream_t*) s;
-	struct stream_impl *si = stream->impl;
-	si->funcs->destructor( prg, sp, si );
-}
-
 /* Keep the position up to date after consuming text. */
 void update_position_seq( struct input_impl_seq *is, const char *data, long length )
 {
@@ -188,250 +156,6 @@ void undo_position_seq( struct input_impl_seq *is, const char *data, long length
 
 
 
-/* Keep the position up to date after consuming text. */
-void update_position_data( struct stream_impl_data *is, const char *data, long length )
-{
-	int i;
-	for ( i = 0; i < length; i++ ) {
-		if ( data[i] != '\n' )
-			is->column += 1;
-		else {
-			is->line += 1;
-			is->column = 1;
-		}
-	}
-
-	is->byte += length;
-}
-
-/* Keep the position up to date after sending back text. */
-void undo_position_data( struct stream_impl_data *is, const char *data, long length )
-{
-	/* FIXME: this needs to fetch the position information from the parsed
-	 * token and restore based on that.. */
-	int i;
-	for ( i = 0; i < length; i++ ) {
-		if ( data[i] == '\n' )
-			is->line -= 1;
-	}
-
-	is->byte -= length;
-}
-
-static struct run_buf *source_stream_data_pop_head( struct stream_impl_data *ss )
-{
-	struct run_buf *ret = ss->queue;
-	ss->queue = ss->queue->next;
-	if ( ss->queue == 0 )
-		ss->queue_tail = 0;
-	else
-		ss->queue->prev = 0;
-	return ret;
-}
-
-static struct run_buf *source_stream_data_pop_tail( struct stream_impl_data *ss )
-{
-	struct run_buf *ret = ss->queue_tail;
-	ss->queue_tail = ss->queue_tail->prev;
-	if ( ss->queue_tail == 0 )
-		ss->queue = 0;
-	else
-		ss->queue_tail->next = 0;
-	return ret;
-}
-
-static void source_stream_data_append( struct stream_impl_data *ss, struct run_buf *run_buf )
-{
-	if ( ss->queue == 0 ) {
-		run_buf->prev = run_buf->next = 0;
-		ss->queue = ss->queue_tail = run_buf;
-	}
-	else {
-		ss->queue_tail->next = run_buf;
-		run_buf->prev = ss->queue_tail;
-		run_buf->next = 0;
-		ss->queue_tail = run_buf;
-	}
-}
-
-static void source_stream_data_prepend( struct stream_impl_data *ss, struct run_buf *run_buf )
-{
-	if ( ss->queue == 0 ) {
-		run_buf->prev = run_buf->next = 0;
-		ss->queue = ss->queue_tail = run_buf;
-	}
-	else {
-		ss->queue->prev = run_buf;
-		run_buf->prev = 0;
-		run_buf->next = ss->queue;
-		ss->queue = run_buf;
-	}
-}
-
-/*
- * Data inputs: files, strings, etc.
- */
-
-static int data_get_data( struct colm_program *prg, struct stream_impl_data *ss, char *dest, int length )
-{
-	int copied = 0;
-
-	/* Move over skip bytes. */
-	struct run_buf *buf = ss->queue;
-	while ( true ) {
-		if ( buf == 0 ) {
-			/* Got through the in-mem buffers without copying anything. */
-			struct run_buf *run_buf = new_run_buf( 0 );
-			source_stream_data_append( ss, run_buf );
-			int received = ss->funcs->get_data_source( prg, (struct stream_impl*)ss, run_buf->data, FSM_BUFSIZE );
-			run_buf->length = received;
-			if ( received == 0 )
-				break;
-
-			buf = run_buf;
-		}
-
-		int avail = buf->length - buf->offset;
-
-		/* Anything available in the current buffer. */
-		if ( avail > 0 ) {
-			/* The source data from the current buffer. */
-			char *src = &buf->data[buf->offset];
-
-			int slen = avail < length ? avail : length;
-			memcpy( dest+copied, src, slen ) ;
-			copied += slen;
-			length -= slen;
-		}
-
-		if ( length == 0 ) {
-			//debug( REALM_INPUT, "exiting get data\n", length );
-			break;
-		}
-
-		buf = buf->next;
-	}
-
-	return copied;
-}
-
-static void data_destructor( program_t *prg, tree_t **sp, struct stream_impl_data *si )
-{
-	if ( si->file != 0 )
-		colm_close_stream_file( si->file );
-	
-	if ( si->collect != 0 ) {
-		str_collect_destroy( si->collect );
-		free( si->collect );
-	}
-
-	struct run_buf *buf = si->queue;
-	while ( buf != 0 ) {
-		struct run_buf *next = buf->next;
-		free( buf );
-		buf = next;
-	}
-
-	si->queue = 0;
-
-	if ( si->data != 0 )
-		free( (char*)si->data );
-
-	/* FIXME: Need to leak this for now. Until we can return strings to a
-	 * program loader and free them at a later date (after the colm program is
-	 * deleted). */
-	// if ( si->name != 0 )
-	//	free( si->name );
-
-	free( si );
-}
-
-static str_collect_t *data_get_collect( struct colm_program *prg, struct stream_impl_data *si )
-{
-	return si->collect;
-}
-
-static void data_flush_stream( struct colm_program *prg, struct stream_impl_data *si )
-{
-	if ( si->file != 0 )
-		fflush( si->file );
-}
-
-static void data_close_stream( struct colm_program *prg, struct stream_impl_data *si )
-{
-	if ( si->file != 0 ) {
-		colm_close_stream_file( si->file );
-		si->file = 0;
-	}
-}
-
-static void data_print_tree( struct colm_program *prg, tree_t **sp,
-		struct stream_impl_data *si, tree_t *tree, int trim )
-{
-	if ( si->file != 0 )
-		colm_print_tree_file( prg, sp, (struct stream_impl*)si, tree, false );
-	else if ( si->collect != 0 )
-		colm_print_tree_collect( prg, sp, si->collect, tree, false );
-}
-
-static int data_get_parse_block( struct colm_program *prg, struct stream_impl_data *ss, int *pskip, char **pdp, int *copied )
-{
-	int ret = 0;
-	*copied = 0;
-
-	/* Move over skip bytes. */
-	struct run_buf *buf = ss->queue;
-	while ( true ) {
-		if ( buf == 0 ) {
-			/* Got through the in-mem buffers without copying anything. */
-			struct run_buf *run_buf = new_run_buf( 0 );
-			source_stream_data_append( ss, run_buf );
-			int received = ss->funcs->get_data_source( prg, (struct stream_impl*)ss, run_buf->data, FSM_BUFSIZE );
-			if ( received == 0 ) {
-				ret = INPUT_EOD;
-				break;
-			}
-			run_buf->length = received;
-
-			int slen = received;
-			*pdp = run_buf->data;
-			*copied = slen;
-			ret = INPUT_DATA;
-			break;
-		}
-
-		int avail = buf->length - buf->offset;
-
-		/* Anything available in the current buffer. */
-		if ( avail > 0 ) {
-			/* The source data from the current buffer. */
-			char *src = &buf->data[buf->offset];
-
-			/* Need to skip? */
-			if ( *pskip > 0 && *pskip >= avail ) {
-				/* Skipping the the whole source. */
-				*pskip -= avail;
-			}
-			else {
-				/* Either skip is zero, or less than slen. Skip goes to zero.
-				 * Some data left over, copy it. */
-				src += *pskip;
-				avail -= *pskip;
-				*pskip = 0;
-
-				int slen = avail;
-				*pdp = src;
-				*copied += slen;
-				ret = INPUT_DATA;
-				break;
-			}
-		}
-
-		buf = buf->next;
-	}
-
-	return ret;
-}
 
 static void input_stream_stash_head( struct colm_program *prg, struct input_impl_seq *si, struct seq_buf *seq_buf )
 {
@@ -472,94 +196,6 @@ static void maybe_split( struct colm_program *prg, struct input_impl_seq *si )
 	}
 }
 
-static int data_consume_data( struct colm_program *prg, struct stream_impl_data *si, int length, location_t *loc )
-{
-	int consumed = 0;
-	int remaining = length;
-
-	/* Move over skip bytes. */
-	while ( true ) {
-		struct run_buf *buf = si->queue;
-
-		if ( buf == 0 )
-			break;
-
-		if ( !loc_set( loc ) )
-			data_transfer_loc( prg, loc, si );
-
-		/* Anything available in the current buffer. */
-		int avail = buf->length - buf->offset;
-		if ( avail > 0 ) {
-			/* The source data from the current buffer. */
-			int slen = avail <= remaining ? avail : remaining;
-			consumed += slen;
-			remaining -= slen;
-			update_position_data( si, buf->data + buf->offset, slen );
-			buf->offset += slen;
-			si->consumed += slen;
-		}
-
-		if ( remaining == 0 )
-			break;
-
-		struct run_buf *run_buf = source_stream_data_pop_head( si );
-		free( run_buf );
-	}
-
-	debug( prg, REALM_INPUT, "data_consume_data: stream %p "
-			"ask: %d, consumed: %d, now: %d\n", si, length, consumed, si->consumed );
-
-	return consumed;
-}
-
-static int data_undo_consume_data( struct colm_program *prg, struct stream_impl_data *si, const char *data, int length )
-{
-	int amount = length;
-	if ( amount > si->consumed )
-		amount = si->consumed;
-
-	if ( amount > 0 ) {
-		struct run_buf *new_buf = new_run_buf( 0 );
-		new_buf->length = amount;
-		memcpy( new_buf->data, data + ( length - amount ), amount );
-		source_stream_data_prepend( si, new_buf );
-		undo_position_data( si, data, amount );
-		si->consumed -= amount;
-	}
-
-	debug( prg, REALM_INPUT, "data_undo_consume_data: stream %p "
-			"undid consume %d of %d bytes, consumed now %d, \n", si, amount, length, si->consumed );
-
-	return amount;
-}
-
-/*
- * File Inputs
- */
-
-static int file_get_data_source( struct colm_program *prg, struct stream_impl_data *si, char *dest, int length )
-{
-	return fread( dest, 1, length, si->file );
-}
-
-void init_file_funcs()
-{
-	memset( &file_funcs, 0, sizeof(struct stream_funcs) );
-}
-
-/*
- * Text inputs
- */
-
-static int accum_get_data_source( struct colm_program *prg, struct stream_impl_data *si, char *dest, int want )
-{
-	long avail = si->dlen - si->offset;
-	long take = avail < want ? avail : want;
-	if ( take > 0 ) 
-		memcpy( dest, si->data + si->offset, take );
-	si->offset += take;
-	return take;
-}
 
 /*
  * StreamImpl struct, this wraps the list of input streams.
@@ -576,20 +212,6 @@ void init_input_impl_seq( struct input_impl_seq *is, char *name )
 	is->byte = 0;
 }
 
-void init_stream_impl_data( struct stream_impl_data *is, char *name )
-{
-	memset( is, 0, sizeof(struct stream_impl_data) );
-
-	is->type = 'D';
-	is->name = name;
-	is->line = 1;
-	is->column = 1;
-	is->byte = 0;
-
-	/* Indentation turned off. */
-	is->level = COLM_INDENT_OFF;
-}
-
 static struct seq_buf *input_stream_seq_pop_head( struct input_impl_seq *is )
 {
 	struct seq_buf *ret = is->queue;
@@ -681,16 +303,6 @@ static void input_destructor( program_t *prg, tree_t **sp, struct input_impl_seq
 	free( si );
 }
 
-char stream_get_eof_sent( struct colm_program *prg, struct input_impl_seq *si )
-{
-	return si->eof_sent;
-}
-
-void stream_set_eof_sent( struct colm_program *prg, struct input_impl_seq *si, char eof_sent )
-{
-	si->eof_sent = eof_sent;
-}
-
 
 static int input_get_parse_block( struct colm_program *prg, struct input_impl_seq *is, int *pskip, char **pdp, int *copied )
 {
@@ -1009,15 +621,6 @@ static tree_t *input_undo_prepend_stream( struct colm_program *prg, struct input
 	return 0;
 }
 
-static struct stream_impl *colm_impl_new_accum( char *name )
-{
-	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
-	init_stream_impl_data( si, name );
-	si->funcs = (struct stream_funcs*)&accum_funcs;
-
-	return (struct stream_impl*)si;
-}
-
 static void input_append_data( struct colm_program *prg, struct input_impl_seq *si, const char *data, long length )
 {
 	debug( prg, REALM_INPUT, "input_append_data: stream %p append data length %d\n", si, length );
@@ -1195,99 +798,6 @@ struct input_funcs_seq input_funcs =
 	&input_destructor,
 };
 
-struct stream_funcs_data file_funcs = 
-{
-	&data_get_parse_block,
-	&data_get_data,
-	&file_get_data_source,
-
-	&data_consume_data,
-	&data_undo_consume_data,
-
-	&data_transfer_loc,
-	&data_get_collect,
-	&data_flush_stream,
-	&data_close_stream,
-	&data_print_tree,
-	&data_destructor,
-};
-
-struct stream_funcs_data accum_funcs = 
-{
-	&data_get_parse_block,
-	&data_get_data,
-	&accum_get_data_source,
-
-	&data_consume_data,
-	&data_undo_consume_data,
-
-	&data_transfer_loc,
-	&data_get_collect,
-	&data_flush_stream,
-	&data_close_stream,
-	&data_print_tree,
-
-	&data_destructor,
-};
-
-static struct stream_impl *colm_impl_new_file( char *name, FILE *file )
-{
-	struct stream_impl_data *ss = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
-	init_stream_impl_data( ss, name );
-	ss->funcs = (struct stream_funcs*)&file_funcs;
-	ss->file = file;
-	return (struct stream_impl*)ss;
-}
-
-static struct stream_impl *colm_impl_new_fd( char *name, long fd )
-{
-	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
-	init_stream_impl_data( si, name );
-	si->funcs = (struct stream_funcs*)&file_funcs;
-	si->file = fdopen( fd, ( fd == 0 ) ? "r" : "w" );
-	return (struct stream_impl*)si;
-}
-
-static struct stream_impl *colm_impl_consumed( char *name, int len )
-{
-	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
-	init_stream_impl_data( si, name );
-	si->funcs = (struct stream_funcs*)&accum_funcs;
-
-	si->data = 0;
-	si->consumed = len;
-	si->offset = len;
-
-	si->dlen = len;
-
-	return (struct stream_impl*)si;
-}
-
-static struct stream_impl *colm_impl_new_text( char *name, const char *data, int len )
-{
-	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
-	init_stream_impl_data( si, name );
-	si->funcs = (struct stream_funcs*)&accum_funcs;
-
-	char *buf = (char*)malloc( len );
-	memcpy( buf, data, len );
-
-	si->data = buf;
-	si->dlen = len;
-
-	return (struct stream_impl*)si;
-}
-
-struct stream_impl *colm_impl_new_collect( char *name )
-{
-	struct stream_impl_data *ss = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
-	init_stream_impl_data( ss, name );
-	ss->funcs = (struct stream_funcs*)&accum_funcs;
-	ss->collect = (struct colm_str_collect*) malloc( sizeof( struct colm_str_collect ) );
-	init_str_collect( ss->collect );
-	return (struct stream_impl*)ss;
-}
-
 struct input_impl *colm_impl_new_generic( char *name )
 {
 	struct input_impl_seq *ss = (struct input_impl_seq*)malloc(sizeof(struct input_impl_seq));
@@ -1307,49 +817,6 @@ input_t *colm_input_new_struct( program_t *prg )
 	return input;
 }
 
-stream_t *colm_stream_open_fd( program_t *prg, char *name, long fd )
-{
-	struct stream_impl *impl = colm_impl_new_fd( colm_filename_add( prg, name ), fd );
-
-	struct colm_stream *s = colm_stream_new_struct( prg );
-	s->impl = impl;
-	return s;
-}
-
-stream_t *colm_stream_open_file( program_t *prg, tree_t *name, tree_t *mode )
-{
-	head_t *head_name = ((str_t*)name)->value;
-	head_t *head_mode = ((str_t*)mode)->value;
-	stream_t *stream = 0;
-
-	const char *given_mode = string_data(head_mode);
-	const char *fopen_mode = 0;
-	if ( memcmp( given_mode, "r", string_length(head_mode) ) == 0 )
-		fopen_mode = "rb";
-	else if ( memcmp( given_mode, "w", string_length(head_mode) ) == 0 )
-		fopen_mode = "wb";
-	else if ( memcmp( given_mode, "a", string_length(head_mode) ) == 0 )
-		fopen_mode = "ab";
-	else {
-		fatal( "unknown file open mode: %s\n", given_mode );
-	}
-	
-	/* Need to make a C-string (null terminated). */
-	char *file_name = (char*)malloc(string_length(head_name)+1);
-	memcpy( file_name, string_data(head_name), string_length(head_name) );
-	file_name[string_length(head_name)] = 0;
-
-	FILE *file = fopen( file_name, fopen_mode );
-	if ( file != 0 ) {
-		stream = colm_stream_new_struct( prg );
-		stream->impl = colm_impl_new_file( colm_filename_add( prg, file_name ), file );
-	}
-
-	free( file_name );
-
-	return stream;
-}
-
 input_t *colm_input_new( program_t *prg )
 {
 	struct input_impl *impl = colm_impl_new_generic( colm_filename_add( prg, "<internal>" ) );
@@ -1358,44 +825,6 @@ input_t *colm_input_new( program_t *prg )
 	return input;
 }
 
-stream_t *colm_stream_new_struct( program_t *prg )
-{
-	size_t memsize = sizeof(struct colm_stream);
-	struct colm_stream *stream = (struct colm_stream*) malloc( memsize );
-	memset( stream, 0, memsize );
-	colm_struct_add( prg, (struct colm_struct *)stream );
-	stream->id = prg->rtd->struct_stream_id;
-	stream->destructor = &colm_stream_destroy;
-	return stream;
-}
-
-
-str_t *collect_string( program_t *prg, stream_t *s )
-{
-	str_collect_t *collect = s->impl->funcs->get_collect( prg, s->impl );
-	head_t *head = string_alloc_full( prg, collect->data, collect->length );
-	str_t *str = (str_t*)construct_string( prg, head );
-	return str;
-}
-
-stream_t *colm_stream_open_collect( program_t *prg )
-{
-	struct stream_impl *impl = colm_impl_new_collect( colm_filename_add( prg, "<internal>" ) );
-	struct colm_stream *stream = colm_stream_new_struct( prg );
-	stream->impl = impl;
-	return stream;
-}
-
-struct stream_impl *colm_stream_impl( struct colm_struct *s )
-{
-	return ((stream_t*)s)->impl;
-}
-
-struct stream_impl *stream_to_impl( stream_t *ptr )
-{
-	return ptr->impl;
-}
-
 struct input_impl *input_to_impl( input_t *ptr )
 {
 	return ptr->impl;
diff --git a/src/input.h b/src/input.h
index 535f2603..43962553 100644
--- a/src/input.h
+++ b/src/input.h
@@ -205,6 +205,8 @@ struct colm_str *collect_string( struct colm_program *prg, struct colm_stream *s
 struct colm_stream *colm_stream_open_collect( struct colm_program *prg );
 
 void colm_close_stream_file( FILE *file );
+char *colm_filename_add( struct colm_program *prg, const char *fn );
+struct stream_impl *colm_impl_new_accum( char *name );
 
 #ifdef __cplusplus
 }
diff --git a/src/stream.c b/src/stream.c
new file mode 100644
index 00000000..ae85fbef
--- /dev/null
+++ b/src/stream.c
@@ -0,0 +1,616 @@
+/*
+ * Copyright 2007-2018 Adrian Thurston <thurston@colm.net>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+#include <colm/input.h>
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <unistd.h>
+#include <stdbool.h>
+
+#include <colm/pdarun.h>
+#include <colm/debug.h>
+#include <colm/program.h>
+#include <colm/tree.h>
+#include <colm/bytecode.h>
+#include <colm/pool.h>
+#include <colm/struct.h>
+
+DEF_STREAM_FUNCS( stream_funcs_data, stream_impl_data );
+
+extern struct stream_funcs_data file_funcs;
+extern struct stream_funcs_data accum_funcs;
+
+static bool loc_set( location_t *loc )
+{
+	return loc->line != 0;
+}
+
+void init_stream_impl_data( struct stream_impl_data *is, char *name )
+{
+	memset( is, 0, sizeof(struct stream_impl_data) );
+
+	is->type = 'D';
+	is->name = name;
+	is->line = 1;
+	is->column = 1;
+	is->byte = 0;
+
+	/* Indentation turned off. */
+	is->level = COLM_INDENT_OFF;
+}
+static struct run_buf *source_stream_data_pop_head( struct stream_impl_data *ss )
+{
+	struct run_buf *ret = ss->queue;
+	ss->queue = ss->queue->next;
+	if ( ss->queue == 0 )
+		ss->queue_tail = 0;
+	else
+		ss->queue->prev = 0;
+	return ret;
+}
+
+static struct run_buf *source_stream_data_pop_tail( struct stream_impl_data *ss )
+{
+	struct run_buf *ret = ss->queue_tail;
+	ss->queue_tail = ss->queue_tail->prev;
+	if ( ss->queue_tail == 0 )
+		ss->queue = 0;
+	else
+		ss->queue_tail->next = 0;
+	return ret;
+}
+
+static void source_stream_data_append( struct stream_impl_data *ss, struct run_buf *run_buf )
+{
+	if ( ss->queue == 0 ) {
+		run_buf->prev = run_buf->next = 0;
+		ss->queue = ss->queue_tail = run_buf;
+	}
+	else {
+		ss->queue_tail->next = run_buf;
+		run_buf->prev = ss->queue_tail;
+		run_buf->next = 0;
+		ss->queue_tail = run_buf;
+	}
+}
+
+static void source_stream_data_prepend( struct stream_impl_data *ss, struct run_buf *run_buf )
+{
+	if ( ss->queue == 0 ) {
+		run_buf->prev = run_buf->next = 0;
+		ss->queue = ss->queue_tail = run_buf;
+	}
+	else {
+		ss->queue->prev = run_buf;
+		run_buf->prev = 0;
+		run_buf->next = ss->queue;
+		ss->queue = run_buf;
+	}
+}
+
+struct run_buf *new_run_buf( int sz )
+{
+	struct run_buf *rb;
+	if ( sz > FSM_BUFSIZE ) {
+		int ssz = sizeof(struct run_buf) + sz - FSM_BUFSIZE;
+		rb = (struct run_buf*) malloc( ssz );
+		memset( rb, 0, ssz );
+	}
+	else {
+		rb = (struct run_buf*) malloc( sizeof(struct run_buf) );
+		memset( rb, 0, sizeof(struct run_buf) );
+	}
+	return rb;
+}
+
+/* Keep the position up to date after consuming text. */
+void update_position_data( struct stream_impl_data *is, const char *data, long length )
+{
+	int i;
+	for ( i = 0; i < length; i++ ) {
+		if ( data[i] != '\n' )
+			is->column += 1;
+		else {
+			is->line += 1;
+			is->column = 1;
+		}
+	}
+
+	is->byte += length;
+}
+
+/* Keep the position up to date after sending back text. */
+void undo_position_data( struct stream_impl_data *is, const char *data, long length )
+{
+	/* FIXME: this needs to fetch the position information from the parsed
+	 * token and restore based on that.. */
+	int i;
+	for ( i = 0; i < length; i++ ) {
+		if ( data[i] == '\n' )
+			is->line -= 1;
+	}
+
+	is->byte -= length;
+}
+
+
+/*
+ * Interface
+ */
+
+static void data_transfer_loc( struct colm_program *prg, location_t *loc, struct stream_impl_data *ss )
+{
+	loc->name = ss->name;
+	loc->line = ss->line;
+	loc->column = ss->column;
+	loc->byte = ss->byte;
+}
+
+/*
+ * Data inputs: files, strings, etc.
+ */
+
+static int data_get_data( struct colm_program *prg, struct stream_impl_data *ss, char *dest, int length )
+{
+	int copied = 0;
+
+	/* Move over skip bytes. */
+	struct run_buf *buf = ss->queue;
+	while ( true ) {
+		if ( buf == 0 ) {
+			/* Got through the in-mem buffers without copying anything. */
+			struct run_buf *run_buf = new_run_buf( 0 );
+			source_stream_data_append( ss, run_buf );
+			int received = ss->funcs->get_data_source( prg, (struct stream_impl*)ss, run_buf->data, FSM_BUFSIZE );
+			run_buf->length = received;
+			if ( received == 0 )
+				break;
+
+			buf = run_buf;
+		}
+
+		int avail = buf->length - buf->offset;
+
+		/* Anything available in the current buffer. */
+		if ( avail > 0 ) {
+			/* The source data from the current buffer. */
+			char *src = &buf->data[buf->offset];
+
+			int slen = avail < length ? avail : length;
+			memcpy( dest+copied, src, slen ) ;
+			copied += slen;
+			length -= slen;
+		}
+
+		if ( length == 0 ) {
+			//debug( REALM_INPUT, "exiting get data\n", length );
+			break;
+		}
+
+		buf = buf->next;
+	}
+
+	return copied;
+}
+
+
+static void data_destructor( program_t *prg, tree_t **sp, struct stream_impl_data *si )
+{
+	if ( si->file != 0 )
+		colm_close_stream_file( si->file );
+	
+	if ( si->collect != 0 ) {
+		str_collect_destroy( si->collect );
+		free( si->collect );
+	}
+
+	struct run_buf *buf = si->queue;
+	while ( buf != 0 ) {
+		struct run_buf *next = buf->next;
+		free( buf );
+		buf = next;
+	}
+
+	si->queue = 0;
+
+	if ( si->data != 0 )
+		free( (char*)si->data );
+
+	/* FIXME: Need to leak this for now. Until we can return strings to a
+	 * program loader and free them at a later date (after the colm program is
+	 * deleted). */
+	// if ( si->name != 0 )
+	//	free( si->name );
+
+	free( si );
+}
+
+static str_collect_t *data_get_collect( struct colm_program *prg, struct stream_impl_data *si )
+{
+	return si->collect;
+}
+
+static void data_flush_stream( struct colm_program *prg, struct stream_impl_data *si )
+{
+	if ( si->file != 0 )
+		fflush( si->file );
+}
+
+static void data_close_stream( struct colm_program *prg, struct stream_impl_data *si )
+{
+	if ( si->file != 0 ) {
+		colm_close_stream_file( si->file );
+		si->file = 0;
+	}
+}
+
+static void data_print_tree( struct colm_program *prg, tree_t **sp,
+		struct stream_impl_data *si, tree_t *tree, int trim )
+{
+	if ( si->file != 0 )
+		colm_print_tree_file( prg, sp, (struct stream_impl*)si, tree, false );
+	else if ( si->collect != 0 )
+		colm_print_tree_collect( prg, sp, si->collect, tree, false );
+}
+
+static int data_get_parse_block( struct colm_program *prg, struct stream_impl_data *ss, int *pskip, char **pdp, int *copied )
+{
+	int ret = 0;
+	*copied = 0;
+
+	/* Move over skip bytes. */
+	struct run_buf *buf = ss->queue;
+	while ( true ) {
+		if ( buf == 0 ) {
+			/* Got through the in-mem buffers without copying anything. */
+			struct run_buf *run_buf = new_run_buf( 0 );
+			source_stream_data_append( ss, run_buf );
+			int received = ss->funcs->get_data_source( prg, (struct stream_impl*)ss, run_buf->data, FSM_BUFSIZE );
+			if ( received == 0 ) {
+				ret = INPUT_EOD;
+				break;
+			}
+			run_buf->length = received;
+
+			int slen = received;
+			*pdp = run_buf->data;
+			*copied = slen;
+			ret = INPUT_DATA;
+			break;
+		}
+
+		int avail = buf->length - buf->offset;
+
+		/* Anything available in the current buffer. */
+		if ( avail > 0 ) {
+			/* The source data from the current buffer. */
+			char *src = &buf->data[buf->offset];
+
+			/* Need to skip? */
+			if ( *pskip > 0 && *pskip >= avail ) {
+				/* Skipping the the whole source. */
+				*pskip -= avail;
+			}
+			else {
+				/* Either skip is zero, or less than slen. Skip goes to zero.
+				 * Some data left over, copy it. */
+				src += *pskip;
+				avail -= *pskip;
+				*pskip = 0;
+
+				int slen = avail;
+				*pdp = src;
+				*copied += slen;
+				ret = INPUT_DATA;
+				break;
+			}
+		}
+
+		buf = buf->next;
+	}
+
+	return ret;
+}
+
+static int data_consume_data( struct colm_program *prg, struct stream_impl_data *si, int length, location_t *loc )
+{
+	int consumed = 0;
+	int remaining = length;
+
+	/* Move over skip bytes. */
+	while ( true ) {
+		struct run_buf *buf = si->queue;
+
+		if ( buf == 0 )
+			break;
+
+		if ( !loc_set( loc ) )
+			data_transfer_loc( prg, loc, si );
+
+		/* Anything available in the current buffer. */
+		int avail = buf->length - buf->offset;
+		if ( avail > 0 ) {
+			/* The source data from the current buffer. */
+			int slen = avail <= remaining ? avail : remaining;
+			consumed += slen;
+			remaining -= slen;
+			update_position_data( si, buf->data + buf->offset, slen );
+			buf->offset += slen;
+			si->consumed += slen;
+		}
+
+		if ( remaining == 0 )
+			break;
+
+		struct run_buf *run_buf = source_stream_data_pop_head( si );
+		free( run_buf );
+	}
+
+	debug( prg, REALM_INPUT, "data_consume_data: stream %p "
+			"ask: %d, consumed: %d, now: %d\n", si, length, consumed, si->consumed );
+
+	return consumed;
+}
+
+static int data_undo_consume_data( struct colm_program *prg, struct stream_impl_data *si, const char *data, int length )
+{
+	int amount = length;
+	if ( amount > si->consumed )
+		amount = si->consumed;
+
+	if ( amount > 0 ) {
+		struct run_buf *new_buf = new_run_buf( 0 );
+		new_buf->length = amount;
+		memcpy( new_buf->data, data + ( length - amount ), amount );
+		source_stream_data_prepend( si, new_buf );
+		undo_position_data( si, data, amount );
+		si->consumed -= amount;
+	}
+
+	debug( prg, REALM_INPUT, "data_undo_consume_data: stream %p "
+			"undid consume %d of %d bytes, consumed now %d, \n", si, amount, length, si->consumed );
+
+	return amount;
+}
+
+/*
+ * File Inputs
+ */
+
+static int file_get_data_source( struct colm_program *prg, struct stream_impl_data *si, char *dest, int length )
+{
+	return fread( dest, 1, length, si->file );
+}
+
+/*
+ * Text inputs
+ */
+
+static int accum_get_data_source( struct colm_program *prg, struct stream_impl_data *si, char *dest, int want )
+{
+	long avail = si->dlen - si->offset;
+	long take = avail < want ? avail : want;
+	if ( take > 0 ) 
+		memcpy( dest, si->data + si->offset, take );
+	si->offset += take;
+	return take;
+}
+
+char stream_get_eof_sent( struct colm_program *prg, struct input_impl_seq *si )
+{
+	return si->eof_sent;
+}
+
+void stream_set_eof_sent( struct colm_program *prg, struct input_impl_seq *si, char eof_sent )
+{
+	si->eof_sent = eof_sent;
+}
+
+struct stream_funcs_data file_funcs = 
+{
+	&data_get_parse_block,
+	&data_get_data,
+	&file_get_data_source,
+
+	&data_consume_data,
+	&data_undo_consume_data,
+
+	&data_transfer_loc,
+	&data_get_collect,
+	&data_flush_stream,
+	&data_close_stream,
+	&data_print_tree,
+	&data_destructor,
+};
+
+struct stream_funcs_data accum_funcs = 
+{
+	&data_get_parse_block,
+	&data_get_data,
+	&accum_get_data_source,
+
+	&data_consume_data,
+	&data_undo_consume_data,
+
+	&data_transfer_loc,
+	&data_get_collect,
+	&data_flush_stream,
+	&data_close_stream,
+	&data_print_tree,
+
+	&data_destructor,
+};
+
+struct stream_impl *colm_impl_new_accum( char *name )
+{
+	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
+	init_stream_impl_data( si, name );
+	si->funcs = (struct stream_funcs*)&accum_funcs;
+
+	return (struct stream_impl*)si;
+}
+
+
+static struct stream_impl *colm_impl_new_file( char *name, FILE *file )
+{
+	struct stream_impl_data *ss = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
+	init_stream_impl_data( ss, name );
+	ss->funcs = (struct stream_funcs*)&file_funcs;
+	ss->file = file;
+	return (struct stream_impl*)ss;
+}
+
+static struct stream_impl *colm_impl_new_fd( char *name, long fd )
+{
+	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
+	init_stream_impl_data( si, name );
+	si->funcs = (struct stream_funcs*)&file_funcs;
+	si->file = fdopen( fd, ( fd == 0 ) ? "r" : "w" );
+	return (struct stream_impl*)si;
+}
+
+struct stream_impl *colm_impl_consumed( char *name, int len )
+{
+	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
+	init_stream_impl_data( si, name );
+	si->funcs = (struct stream_funcs*)&accum_funcs;
+
+	si->data = 0;
+	si->consumed = len;
+	si->offset = len;
+
+	si->dlen = len;
+
+	return (struct stream_impl*)si;
+}
+
+struct stream_impl *colm_impl_new_text( char *name, const char *data, int len )
+{
+	struct stream_impl_data *si = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
+	init_stream_impl_data( si, name );
+	si->funcs = (struct stream_funcs*)&accum_funcs;
+
+	char *buf = (char*)malloc( len );
+	memcpy( buf, data, len );
+
+	si->data = buf;
+	si->dlen = len;
+
+	return (struct stream_impl*)si;
+}
+
+struct stream_impl *colm_impl_new_collect( char *name )
+{
+	struct stream_impl_data *ss = (struct stream_impl_data*)malloc(sizeof(struct stream_impl_data));
+	init_stream_impl_data( ss, name );
+	ss->funcs = (struct stream_funcs*)&accum_funcs;
+	ss->collect = (struct colm_str_collect*) malloc( sizeof( struct colm_str_collect ) );
+	init_str_collect( ss->collect );
+	return (struct stream_impl*)ss;
+}
+
+struct stream_impl *stream_to_impl( stream_t *ptr )
+{
+	return ptr->impl;
+}
+
+str_t *collect_string( program_t *prg, stream_t *s )
+{
+	str_collect_t *collect = s->impl->funcs->get_collect( prg, s->impl );
+	head_t *head = string_alloc_full( prg, collect->data, collect->length );
+	str_t *str = (str_t*)construct_string( prg, head );
+	return str;
+}
+
+stream_t *colm_stream_open_fd( program_t *prg, char *name, long fd )
+{
+	struct stream_impl *impl = colm_impl_new_fd( colm_filename_add( prg, name ), fd );
+
+	struct colm_stream *s = colm_stream_new_struct( prg );
+	s->impl = impl;
+	return s;
+}
+
+stream_t *colm_stream_open_file( program_t *prg, tree_t *name, tree_t *mode )
+{
+	head_t *head_name = ((str_t*)name)->value;
+	head_t *head_mode = ((str_t*)mode)->value;
+	stream_t *stream = 0;
+
+	const char *given_mode = string_data(head_mode);
+	const char *fopen_mode = 0;
+	if ( memcmp( given_mode, "r", string_length(head_mode) ) == 0 )
+		fopen_mode = "rb";
+	else if ( memcmp( given_mode, "w", string_length(head_mode) ) == 0 )
+		fopen_mode = "wb";
+	else if ( memcmp( given_mode, "a", string_length(head_mode) ) == 0 )
+		fopen_mode = "ab";
+	else {
+		fatal( "unknown file open mode: %s\n", given_mode );
+	}
+	
+	/* Need to make a C-string (null terminated). */
+	char *file_name = (char*)malloc(string_length(head_name)+1);
+	memcpy( file_name, string_data(head_name), string_length(head_name) );
+	file_name[string_length(head_name)] = 0;
+
+	FILE *file = fopen( file_name, fopen_mode );
+	if ( file != 0 ) {
+		stream = colm_stream_new_struct( prg );
+		stream->impl = colm_impl_new_file( colm_filename_add( prg, file_name ), file );
+	}
+
+	free( file_name );
+
+	return stream;
+}
+
+
+void colm_stream_destroy( program_t *prg, tree_t **sp, struct_t *s )
+{
+	stream_t *stream = (stream_t*) s;
+	struct stream_impl *si = stream->impl;
+	si->funcs->destructor( prg, sp, si );
+}
+
+stream_t *colm_stream_new_struct( program_t *prg )
+{
+	size_t memsize = sizeof(struct colm_stream);
+	struct colm_stream *stream = (struct colm_stream*) malloc( memsize );
+	memset( stream, 0, memsize );
+	colm_struct_add( prg, (struct colm_struct *)stream );
+	stream->id = prg->rtd->struct_stream_id;
+	stream->destructor = &colm_stream_destroy;
+	return stream;
+}
+
+stream_t *colm_stream_open_collect( program_t *prg )
+{
+	struct stream_impl *impl = colm_impl_new_collect( colm_filename_add( prg, "<internal>" ) );
+	struct colm_stream *stream = colm_stream_new_struct( prg );
+	stream->impl = impl;
+	return stream;
+}
+