various input cleanup, removed line info from input_impl_seq

3 files changed, 121 insertions, 179 deletions

diff --git a/src/input.c b/src/input.c
index 80a6e028..a404b788 100644
--- a/src/input.c
+++ b/src/input.c
@@ -37,8 +37,8 @@
 #include <colm/pool.h>
 #include <colm/struct.h>
 
-struct stream_impl *colm_impl_consumed( char *name, int len );
-struct stream_impl *colm_impl_new_text( char *name, const char *data, int len );
+DEF_INPUT_FUNCS( input_funcs_seq, input_impl_seq );
+extern struct input_funcs_seq input_funcs;
 
 static bool is_tree( struct seq_buf *b )
 {
@@ -77,86 +77,22 @@ static struct seq_buf *new_seq_buf()
 	return rb;
 }
 
-DEF_INPUT_FUNCS( input_funcs_seq, input_impl_seq );
-
-extern struct input_funcs_seq input_funcs;
-
-static bool loc_set( location_t *loc )
-{
-	return loc->line != 0;
-}
-
-static void default_loc( location_t *loc )
+static void input_transfer_loc( struct colm_program *prg, location_t *loc, struct input_impl_seq *ss )
 {
-	loc->name = "--";
-	loc->line = 1;
-	loc->column = 1;
-	loc->byte = 1;
 }
 
-void input_transfer_loc( struct colm_program *prg, location_t *loc, struct input_impl_seq *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 )
 {
 	return is_stream( buf ) && buf->own_si;
 }
 
-void colm_close_stream_file( FILE *file )
-{
-	if ( file != stdin && file != stdout && file != stderr &&
-			fileno(file) != 0 && fileno( file) != 1 && fileno(file) != 2 )
-	{
-		fclose( file );
-	}
-}
-
-void colm_input_destroy( program_t *prg, tree_t **sp, struct_t *s )
+static void colm_input_destroy( program_t *prg, tree_t **sp, struct_t *s )
 {
 	input_t *input = (input_t*) s;
 	struct input_impl *si = input->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 )
-{
-	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_seq( struct input_impl_seq *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 void input_stream_stash_head( struct colm_program *prg, struct input_impl_seq *si, struct seq_buf *seq_buf )
 {
 	debug( prg, REALM_INPUT, "stash_head: stream %p buf %p\n", si, seq_buf );
@@ -177,10 +113,10 @@ static struct seq_buf *input_stream_pop_stash( struct colm_program *prg, struct
 
 static void maybe_split( struct colm_program *prg, struct input_impl_seq *si )
 {
-	if ( si->queue != 0 && is_stream( si->queue ) ) {
+	if ( si->queue.head != 0 && is_stream( si->queue.head ) ) {
 
 		/* NOT A GOOD IDEA. Use func instead */
-		struct stream_impl_data *sid = (struct stream_impl_data*)si->queue->si;
+		struct stream_impl_data *sid = (struct stream_impl_data*)si->queue.head->si;
 		if ( sid->consumed > 0 ) {
 			debug( prg, REALM_INPUT, "maybe split: consumed is > 0, splitting\n" );
 			struct stream_impl *sub_si = colm_impl_consumed( "<text>", sid->consumed );
@@ -206,59 +142,59 @@ void init_input_impl_seq( struct input_impl_seq *is, char *name )
 	memset( is, 0, sizeof(struct input_impl_seq) );
 
 	is->type = 'S';
-	is->name = name;
-	is->line = 1;
-	is->column = 1;
-	is->byte = 0;
+	//is->name = name;
+	//is->line = 1;
+	//is->column = 1;
+	//is->byte = 0;
 }
 
 static struct seq_buf *input_stream_seq_pop_head( struct input_impl_seq *is )
 {
-	struct seq_buf *ret = is->queue;
-	is->queue = is->queue->next;
-	if ( is->queue == 0 )
-		is->queue_tail = 0;
+	struct seq_buf *ret = is->queue.head;
+	is->queue.head = is->queue.head->next;
+	if ( is->queue.head == 0 )
+		is->queue.tail = 0;
 	else
-		is->queue->prev = 0;
+		is->queue.head->prev = 0;
 	return ret;
 }
 
 static void input_stream_seq_append( struct input_impl_seq *is, struct seq_buf *seq_buf )
 {
-	if ( is->queue == 0 ) {
+	if ( is->queue.head == 0 ) {
 		seq_buf->prev = seq_buf->next = 0;
-		is->queue = is->queue_tail = seq_buf;
+		is->queue.head = is->queue.tail = seq_buf;
 	}
 	else {
-		is->queue_tail->next = seq_buf;
-		seq_buf->prev = is->queue_tail;
+		is->queue.tail->next = seq_buf;
+		seq_buf->prev = is->queue.tail;
 		seq_buf->next = 0;
-		is->queue_tail = seq_buf;
+		is->queue.tail = seq_buf;
 	}
 }
 
 static struct seq_buf *input_stream_seq_pop_tail( struct input_impl_seq *is )
 {
-	struct seq_buf *ret = is->queue_tail;
-	is->queue_tail = is->queue_tail->prev;
-	if ( is->queue_tail == 0 )
-		is->queue = 0;
+	struct seq_buf *ret = is->queue.tail;
+	is->queue.tail = is->queue.tail->prev;
+	if ( is->queue.tail == 0 )
+		is->queue.head = 0;
 	else
-		is->queue_tail->next = 0;
+		is->queue.tail->next = 0;
 	return ret;
 }
 
 static void input_stream_seq_prepend( struct input_impl_seq *is, struct seq_buf *seq_buf )
 {
-	if ( is->queue == 0 ) {
+	if ( is->queue.head == 0 ) {
 		seq_buf->prev = seq_buf->next = 0;
-		is->queue = is->queue_tail = seq_buf;
+		is->queue.head = is->queue.tail = seq_buf;
 	}
 	else {
-		is->queue->prev = seq_buf;
+		is->queue.head->prev = seq_buf;
 		seq_buf->prev = 0;
-		seq_buf->next = is->queue;
-		is->queue = seq_buf;
+		seq_buf->next = is->queue.head;
+		is->queue.head = seq_buf;
 	}
 }
 
@@ -269,7 +205,7 @@ void input_set_eof_mark( struct colm_program *prg, struct input_impl_seq *si, ch
 
 static void input_destructor( program_t *prg, tree_t **sp, struct input_impl_seq *si )
 {
-	struct seq_buf *buf = si->queue;
+	struct seq_buf *buf = si->queue.head;
 	while ( buf != 0 ) {
 		if ( is_tree( buf ) )
 			colm_tree_downref( prg, sp, buf->tree );
@@ -292,7 +228,7 @@ static void input_destructor( program_t *prg, tree_t **sp, struct input_impl_seq
 		buf = next;
 	}
 
-	si->queue = 0;
+	si->queue.head = 0;
 
 	/* 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
@@ -310,7 +246,7 @@ static int input_get_parse_block( struct colm_program *prg, struct input_impl_se
 	*copied = 0;
 
 	/* Move over skip bytes. */
-	struct seq_buf *buf = is->queue;
+	struct seq_buf *buf = is->queue.head;
 	while ( true ) {
 		if ( buf == 0 ) {
 			/* Got through the in-mem buffers without copying anything. */
@@ -380,7 +316,7 @@ static int input_get_data( struct colm_program *prg, struct input_impl_seq *is,
 	int copied = 0;
 
 	/* Move over skip bytes. */
-	struct seq_buf *buf = is->queue;
+	struct seq_buf *buf = is->queue.head;
 	while ( true ) {
 		if ( buf == 0 ) {
 			/* Got through the in-mem buffers without copying anything. */
@@ -428,7 +364,7 @@ static int input_consume_data( struct colm_program *prg, struct input_impl_seq *
 
 	/* Move over skip bytes. */
 	while ( true ) {
-		struct seq_buf *buf = si->queue;
+		struct seq_buf *buf = si->queue.head;
 
 		if ( buf == 0 )
 			break;
@@ -446,12 +382,7 @@ static int input_consume_data( struct colm_program *prg, struct input_impl_seq *
 		else if ( buf->type == SB_IGNORE )
 			break;
 		else {
-			if ( !loc_set( loc ) ) {
-				if ( si->line > 0 )
-					input_transfer_loc( prg, loc, si );
-				else
-					default_loc( loc );
-			}
+			assert(false);
 		}
 
 		if ( length == 0 ) {
@@ -478,8 +409,8 @@ static int input_undo_consume_data( struct colm_program *prg, struct input_impl_
 	int remaining = length;
 
 	while ( true ) {
-		if ( is_stream( si->queue ) ) {
-			struct stream_impl *sub = si->queue->si;
+		if ( is_stream( si->queue.head ) ) {
+			struct stream_impl *sub = si->queue.head->si;
 			int pushed_back = sub->funcs->undo_consume_data( prg, sub, data, remaining );
 			remaining -= pushed_back;
 			offset += pushed_back;
@@ -499,14 +430,14 @@ static tree_t *input_consume_tree( struct colm_program *prg, struct input_impl_s
 {
 	debug( prg, REALM_INPUT, "input_consume_tree: stream %p\n", si );
 
-	while ( si->queue != 0 && is_stream( si->queue ) )
+	while ( si->queue.head != 0 && is_stream( si->queue.head ) )
 	{
 		debug( prg, REALM_INPUT, "  stream %p consume: clearing source type\n", si );
 		struct seq_buf *seq_buf = input_stream_seq_pop_head( si );
 		input_stream_stash_head( prg, si, seq_buf );
 	}
 
-	assert( si->queue != 0 && ( si->queue->type == SB_TOKEN || si->queue->type == SB_IGNORE ) );
+	assert( si->queue.head != 0 && ( si->queue.head->type == SB_TOKEN || si->queue.head->type == SB_IGNORE ) );
 
 	{
 		struct seq_buf *seq_buf = input_stream_seq_pop_head( si );
@@ -585,8 +516,8 @@ static tree_t *input_undo_prepend_tree( struct colm_program *prg, struct input_i
 {
 	debug( prg, REALM_INPUT, "input_undo_prepend_tree: stream %p undo prepend tree\n", si );
 
-	assert( si->queue != 0 && ( si->queue->type == SB_TOKEN ||
-			si->queue->type == SB_IGNORE ) );
+	assert( si->queue.head != 0 && ( si->queue.head->type == SB_TOKEN ||
+			si->queue.head->type == SB_IGNORE ) );
 
 	struct seq_buf *seq_buf = input_stream_seq_pop_head( si );
 
@@ -626,7 +557,7 @@ static void input_append_data( struct colm_program *prg, struct input_impl_seq *
 	debug( prg, REALM_INPUT, "input_append_data: stream %p append data length %d\n", si, length );
 
 #ifdef OPTIM_APPEND
-	if ( si->queue_tail == 0 || si->queue_tail->type != SB_ACCUM ) { 
+	if ( si->queue.tail == 0 || si->queue.tail->type != SB_ACCUM ) { 
 		debug( prg, REALM_INPUT, "input_append_data: creating accum\n" );
 
 		struct stream_impl *sub_si = colm_impl_new_accum( "<text>" );
@@ -639,9 +570,9 @@ static void input_append_data( struct colm_program *prg, struct input_impl_seq *
 		input_stream_seq_append( si, new_buf );
 	}
 
-	struct stream_impl_data *sub_si = (struct stream_impl_data*)si->queue_tail->si;
+	struct stream_impl_data *sub_si = (struct stream_impl_data*)si->queue.tail->si;
 
-	struct run_buf *tail = sub_si->queue_tail;
+	struct run_buf *tail = sub_si->queue.tail;
 	if ( tail == 0 || length > (FSM_BUFSIZE - tail->length) ) {
 		debug( prg, REALM_INPUT, "input_append_data: allocating run buf\n" );
 		tail = new_run_buf( length );
@@ -672,13 +603,13 @@ static tree_t *input_undo_append_data( struct colm_program *prg, struct input_im
 	debug( prg, REALM_INPUT, "input_undo_append_data: stream %p undo append data length %d\n", si, length );
 
 #ifdef OPTIM_APPEND
-	struct stream_impl_data *sub_si = (struct stream_impl_data*)si->queue_tail->si;
+	struct stream_impl_data *sub_si = (struct stream_impl_data*)si->queue.tail->si;
 
-	while ( sub_si->queue_tail->length == 0 )
+	while ( sub_si->queue.tail->length == 0 )
 		source_stream_data_pop_tail( sub_si );
 		
 	while ( length > 0 ) {
-		struct run_buf *tail = sub_si->queue_tail;
+		struct run_buf *tail = sub_si->queue.tail;
 
 		debug( prg, REALM_INPUT, "input_undo_append_data: removing from "
 				"accum tail, offset: %d, length: %d, dlen: %d\n", tail->offset, tail->length, length );
@@ -694,12 +625,12 @@ static tree_t *input_undo_append_data( struct colm_program *prg, struct input_im
 			debug( prg, REALM_INPUT, "input_undo_append_data: removing tail\n" );
 		}
 
-		while ( sub_si->queue_tail->length == 0 )
+		while ( sub_si->queue.tail->length == 0 )
 			source_stream_data_pop_tail( sub_si );
 
 		if ( sub_si->queue == 0 ) {
 			input_stream_seq_pop_tail( si );
-			sub_si = (struct stream_impl_data*)si->queue_tail->si;
+			sub_si = (struct stream_impl_data*)si->queue.tail->si;
 			debug( prg, REALM_INPUT, "input_undo_append_data: removing sub_si\n" );
 		}
 
diff --git a/src/input.h b/src/input.h
index 43962553..e4cfbde4 100644
--- a/src/input.h
+++ b/src/input.h
@@ -151,17 +151,13 @@ struct input_impl_seq
 	char eof_mark;
 	char eof_sent;
 
-	struct seq_buf *queue;
-	struct seq_buf *queue_tail;
+	struct {
+		struct seq_buf *head;
+		struct seq_buf *tail;
+	} queue;
 
 	struct seq_buf *stash;
 
-	long line;
-	long column;
-	long byte;
-
-	char *name;
-
 	int consumed;
 };
 
@@ -171,8 +167,10 @@ struct stream_impl_data
 	struct stream_funcs *funcs;
 	char type;
 
-	struct run_buf *queue;
-	struct run_buf *queue_tail;
+	struct {
+		struct run_buf *head;
+		struct run_buf *tail;
+	} queue;
 
 	const char *data;
 	long dlen;
@@ -204,9 +202,10 @@ struct stream_impl *colm_stream_impl( struct colm_struct *s );
 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 );
+struct stream_impl *colm_impl_consumed( char *name, int len );
+struct stream_impl *colm_impl_new_text( char *name, const char *data, int len );
 
 #ifdef __cplusplus
 }
diff --git a/src/stream.c b/src/stream.c
index ae85fbef..8dc68c32 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -47,7 +47,16 @@ static bool loc_set( location_t *loc )
 	return loc->line != 0;
 }
 
-void init_stream_impl_data( struct stream_impl_data *is, char *name )
+static void close_stream_file( FILE *file )
+{
+	if ( file != stdin && file != stdout && file != stderr &&
+			fileno(file) != 0 && fileno( file) != 1 && fileno(file) != 2 )
+	{
+		fclose( file );
+	}
+}
+
+static void si_data_init( struct stream_impl_data *is, char *name )
 {
 	memset( is, 0, sizeof(struct stream_impl_data) );
 
@@ -60,56 +69,59 @@ void init_stream_impl_data( struct stream_impl_data *is, char *name )
 	/* Indentation turned off. */
 	is->level = COLM_INDENT_OFF;
 }
-static struct run_buf *source_stream_data_pop_head( struct stream_impl_data *ss )
+
+static void si_data_push_tail( struct stream_impl_data *ss, struct run_buf *run_buf )
 {
-	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;
+	if ( ss->queue.head == 0 ) {
+		run_buf->prev = run_buf->next = 0;
+		ss->queue.head = 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 struct run_buf *source_stream_data_pop_tail( struct stream_impl_data *ss )
+static struct run_buf *si_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;
+	struct run_buf *ret = ss->queue.tail;
+	ss->queue.tail = ss->queue.tail->prev;
+	if ( ss->queue.tail == 0 )
+		ss->queue.head = 0;
 	else
-		ss->queue_tail->next = 0;
+		ss->queue.tail->next = 0;
 	return ret;
 }
 
-static void source_stream_data_append( struct stream_impl_data *ss, struct run_buf *run_buf )
+
+static void si_data_push_head( struct stream_impl_data *ss, struct run_buf *run_buf )
 {
-	if ( ss->queue == 0 ) {
+	if ( ss->queue.head == 0 ) {
 		run_buf->prev = run_buf->next = 0;
-		ss->queue = ss->queue_tail = run_buf;
+		ss->queue.head = 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;
+		ss->queue.head->prev = run_buf;
+		run_buf->prev = 0;
+		run_buf->next = ss->queue.head;
+		ss->queue.head = run_buf;
 	}
 }
 
-static void source_stream_data_prepend( struct stream_impl_data *ss, struct run_buf *run_buf )
+static struct run_buf *source_stream_data_pop_head( struct stream_impl_data *ss )
 {
-	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 *ret = ss->queue.head;
+	ss->queue.head = ss->queue.head->next;
+	if ( ss->queue.head == 0 )
+		ss->queue.tail = 0;
+	else
+		ss->queue.head->prev = 0;
+	return ret;
 }
 
+
 struct run_buf *new_run_buf( int sz )
 {
 	struct run_buf *rb;
@@ -177,12 +189,12 @@ static int data_get_data( struct colm_program *prg, struct stream_impl_data *ss,
 	int copied = 0;
 
 	/* Move over skip bytes. */
-	struct run_buf *buf = ss->queue;
+	struct run_buf *buf = ss->queue.head;
 	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 );
+			si_data_push_tail( 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 )
@@ -219,21 +231,21 @@ static int data_get_data( struct colm_program *prg, struct stream_impl_data *ss,
 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 );
+		close_stream_file( si->file );
 	
 	if ( si->collect != 0 ) {
 		str_collect_destroy( si->collect );
 		free( si->collect );
 	}
 
-	struct run_buf *buf = si->queue;
+	struct run_buf *buf = si->queue.head;
 	while ( buf != 0 ) {
 		struct run_buf *next = buf->next;
 		free( buf );
 		buf = next;
 	}
 
-	si->queue = 0;
+	si->queue.head = 0;
 
 	if ( si->data != 0 )
 		free( (char*)si->data );
@@ -261,7 +273,7 @@ static void data_flush_stream( struct colm_program *prg, struct stream_impl_data
 static void data_close_stream( struct colm_program *prg, struct stream_impl_data *si )
 {
 	if ( si->file != 0 ) {
-		colm_close_stream_file( si->file );
+		close_stream_file( si->file );
 		si->file = 0;
 	}
 }
@@ -281,12 +293,12 @@ static int data_get_parse_block( struct colm_program *prg, struct stream_impl_da
 	*copied = 0;
 
 	/* Move over skip bytes. */
-	struct run_buf *buf = ss->queue;
+	struct run_buf *buf = ss->queue.head;
 	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 );
+			si_data_push_tail( 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;
@@ -341,7 +353,7 @@ static int data_consume_data( struct colm_program *prg, struct stream_impl_data
 
 	/* Move over skip bytes. */
 	while ( true ) {
-		struct run_buf *buf = si->queue;
+		struct run_buf *buf = si->queue.head;
 
 		if ( buf == 0 )
 			break;
@@ -384,7 +396,7 @@ static int data_undo_consume_data( struct colm_program *prg, struct stream_impl_
 		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 );
+		si_data_push_head( si, new_buf );
 		undo_position_data( si, data, amount );
 		si->consumed -= amount;
 	}
@@ -466,7 +478,7 @@ struct stream_funcs_data accum_funcs =
 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_data_init( si, name );
 	si->funcs = (struct stream_funcs*)&accum_funcs;
 
 	return (struct stream_impl*)si;
@@ -476,7 +488,7 @@ struct stream_impl *colm_impl_new_accum( char *name )
 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 );
+	si_data_init( ss, name );
 	ss->funcs = (struct stream_funcs*)&file_funcs;
 	ss->file = file;
 	return (struct stream_impl*)ss;
@@ -485,7 +497,7 @@ static struct stream_impl *colm_impl_new_file( char *name, FILE *file )
 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_data_init( si, name );
 	si->funcs = (struct stream_funcs*)&file_funcs;
 	si->file = fdopen( fd, ( fd == 0 ) ? "r" : "w" );
 	return (struct stream_impl*)si;
@@ -494,7 +506,7 @@ static struct stream_impl *colm_impl_new_fd( char *name, long fd )
 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_data_init( si, name );
 	si->funcs = (struct stream_funcs*)&accum_funcs;
 
 	si->data = 0;
@@ -509,7 +521,7 @@ struct stream_impl *colm_impl_consumed( char *name, int len )
 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_data_init( si, name );
 	si->funcs = (struct stream_funcs*)&accum_funcs;
 
 	char *buf = (char*)malloc( len );
@@ -524,7 +536,7 @@ struct stream_impl *colm_impl_new_text( char *name, const char *data, int len )
 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 );
+	si_data_init( 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 );