ssize_t i_stream_read_copy_from_parent(struct istream *istream)
{
	struct istream_private *stream = istream->real_stream;
	size_t pos;
	ssize_t ret;

	stream->pos -= stream->skip;
	stream->skip = 0;

	stream->buffer = i_stream_get_data(stream->parent, &pos);
	if (pos > stream->pos)
		ret = 0;
	else do {
		if ((ret = i_stream_read(stream->parent)) == -2)
			return -2;

		stream->istream.stream_errno = stream->parent->stream_errno;
		stream->istream.eof = stream->parent->eof;
		stream->buffer = i_stream_get_data(stream->parent, &pos);
		/* check again, in case the parent stream had been seeked
		   backwards and the previous read() didn't get us far
		   enough. */
	} while (pos <= stream->pos && ret > 0);

	ret = pos > stream->pos ? (ssize_t)(pos - stream->pos) :
		(ret == 0 ? 0 : -1);
	stream->pos = pos;
	i_assert(ret != -1 || stream->istream.eof ||
		 stream->istream.stream_errno != 0);
	return ret;
}

static ssize_t i_stream_tee_read(struct istream_private *stream)
{
	struct tee_child_istream *tstream = (struct tee_child_istream *)stream;
	struct istream *input = tstream->tee->input;
	const unsigned char *data;
	size_t size;
	uoff_t last_high_offset;
	ssize_t ret;

	tstream->last_read_waiting = FALSE;
	if (stream->buffer == NULL) {
		/* initial read */
		tee_streams_update_buffer(tstream->tee);
	}
	data = i_stream_get_data(input, &size);

	/* last_high_offset contains how far we have read this child tee stream
	   so far. input->v_offset + size contains how much is available in
	   the parent stream without having to read more. */
	last_high_offset = stream->istream.v_offset +
		(stream->pos - stream->skip);
	if (stream->pos == size) {
		/* we've read everything, need to read more */
		i_assert(last_high_offset == input->v_offset + size);
		tee_streams_skip(tstream->tee);
		ret = i_stream_read(input);
		if (ret <= 0) {
			size = i_stream_get_data_size(input);
			if (ret == -2 && stream->skip != 0) {
				/* someone else is holding the data,
				   wait for it */
				tstream->last_read_waiting = TRUE;
				return 0;
			}
			stream->istream.stream_errno = input->stream_errno;
			stream->istream.eof = input->eof;
			return ret;
		}
		tee_streams_update_buffer(tstream->tee);
		data = i_stream_get_data(input, &size);
	} else {
		/* there's still some data available from parent */
		i_assert(last_high_offset < input->v_offset + size);
		tee_streams_update_buffer(tstream->tee);
		i_assert(stream->pos < size);
	}

	i_assert(stream->buffer == data);
	ret = size - stream->pos;
	i_assert(ret > 0);
	stream->pos = size;

	i_assert(stream->istream.v_offset + (stream->pos - stream->skip) ==
		 input->v_offset + size);
	return ret;
}

static void test_istream_children(void)
{
	struct istream *parent, *child1, *child2;
	const unsigned char *data;
	size_t size;

	test_begin("istream children");

	parent = test_istream_create_data("123456789", 9);
	test_istream_set_max_buffer_size(parent, 3);

	child1 = i_stream_create_limit(parent, (uoff_t)-1);
	child2 = i_stream_create_limit(parent, (uoff_t)-1);

	/* child1 read beginning */
	test_assert(i_stream_read(child1) == 3);
	data = i_stream_get_data(child1, &size);
	test_assert(size == 3 && memcmp(data, "123", 3) == 0);
	i_stream_skip(child1, 3);
	/* child1 read middle.. */
	test_assert(i_stream_read(child1) == 3);
	data = i_stream_get_data(child1, &size);
	test_assert(size == 3 && memcmp(data, "456", 3) == 0);
	/* child2 read beginning.. */
	test_assert(i_stream_read(child2) == 3);
	data = i_stream_get_data(child2, &size);
	test_assert(size == 3 && memcmp(data, "123", 3) == 0);
	/* child1 check middle again.. the parent has been modified,
	   so it can't return the original data (without some code changes). */
	test_assert(i_stream_get_data_size(child1) == 0);
	i_stream_skip(child1, 3);
	/* child1 read end */
	test_assert(i_stream_read(child1) == 3);
	data = i_stream_get_data(child1, &size);
	test_assert(size == 3 && memcmp(data, "789", 3) == 0);
	i_stream_skip(child1, 3);
	test_assert(i_stream_read(child1) == -1);
	/* child2 check beginning again.. */
	test_assert(i_stream_get_data_size(child1) == 0);
	i_stream_skip(child2, 3);
	/* child2 read middle */
	test_assert(i_stream_read(child2) == 3);
	data = i_stream_get_data(child2, &size);
	test_assert(size == 3 && memcmp(data, "456", 3) == 0);
	i_stream_skip(child2, 3);

	i_stream_destroy(&child1);
	i_stream_destroy(&child2);
	i_stream_destroy(&parent);

	test_end();
}

static void test_istream_binary_converter(void)
{
	struct istream *datainput, *input;
	const unsigned char *data;
	size_t i, size;
	int ret;

	test_begin("istream binary converter");
	datainput = test_istream_create_data(mail_input, sizeof(mail_input));
	test_istream_set_allow_eof(datainput, FALSE);
	input = i_stream_create_binary_converter(datainput);

	for (i = 1; i <= sizeof(mail_input); i++) {
		test_istream_set_size(datainput, i);
		while ((ret = i_stream_read(input)) > 0) ;
		test_assert(ret == 0);
	}
	test_istream_set_allow_eof(datainput, TRUE);
	while ((ret = i_stream_read(input)) > 0) ;
	test_assert(ret == -1);

	data = i_stream_get_data(input, &size);
	test_assert(size == sizeof(mail_output) &&
		    memcmp(data, mail_output, size) == 0);
	i_stream_unref(&input);
	i_stream_unref(&datainput);
	test_end();
}

static void test_dsync_proxy_msg_save(void)
{
	static const char *input = "..dotty\n..behavior\nfrom you\n.\nstop";
	struct test_dsync_msg_event event;
	const unsigned char *data;
	size_t size;

	test_begin("proxy server msg save");

	server->input = i_stream_create_from_data(input, strlen(input));

	test_assert(run_cmd("MSG-SAVE", "28492428", "pop3uidl",
			    "saveguid", "874", "33982482882924", "\\Flagged bar \\Answered",
			    "8294284", NULL) == 1);
	test_assert(test_dsync_worker_next_msg_event(test_worker, &event));
	test_assert(event.type == LAST_MSG_TYPE_SAVE);
	test_assert(event.save_data.received_date == 28492428);
	test_assert(strcmp(event.save_data.pop3_uidl, "pop3uidl") == 0);
	test_assert(strcmp(event.save_body, ".dotty\n.behavior\nfrom you") == 0);

	test_assert(strcmp(event.msg.guid, "saveguid") == 0);
	test_assert(event.msg.uid == 874);
	test_assert(event.msg.modseq == 33982482882924);
	test_assert(event.msg.flags == (MAIL_FLAGGED | MAIL_ANSWERED));
	test_assert(strcmp(event.msg.keywords[0], "bar") == 0);
	test_assert(event.msg.keywords[1] == NULL);
	test_assert(event.msg.save_date == 8294284);

	data = i_stream_get_data(server->input, &size);
	test_assert(size == 4 && memcmp(data, "stop", 4) == 0);
	i_stream_destroy(&server->input);

	test_end();
}

size_t i_stream_get_data_size(struct istream *stream)
{
	size_t size;

	(void)i_stream_get_data(stream, &size);
	return size;
}

static ssize_t
i_stream_hash_read(struct istream_private *stream)
{
	struct hash_istream *hstream = (struct hash_istream *)stream;
	const unsigned char *data;
	size_t size;
	uoff_t skip;
	ssize_t ret;

	i_stream_seek(stream->parent, stream->parent_start_offset +
		      stream->istream.v_offset);

	ret = i_stream_read_copy_from_parent(&stream->istream);
	if (ret > 0 && hstream->hash_context != NULL) {
		data = i_stream_get_data(&stream->istream, &size);
		i_assert((size_t)ret <= size);

		i_assert(stream->istream.v_offset <= hstream->high_offset);
		skip = hstream->high_offset - stream->istream.v_offset;
		if (skip < (size_t)size) {
			hstream->high_offset += (size-skip);
			hstream->method->loop(hstream->hash_context,
					      data+skip, size-skip);
		}
	} else if (ret < 0) {
		/* we finished hashing it. don't access it anymore, because
		   the memory pointed by the hash may be freed before the
		   istream itself */
		hstream->hash_context = NULL;
	}
	return ret;
}

static void
decode_test(const char *base64_input, const char *output, bool broken_input)
{
	unsigned int base64_input_len = strlen(base64_input);
	struct istream *input_data, *input;
	const unsigned char *data;
	size_t i, size;
	int ret = 0;

	input_data = test_istream_create_data(base64_input, base64_input_len);
	test_istream_set_allow_eof(input_data, FALSE);
	input = i_stream_create_base64_decoder(input_data);

	for (i = 1; i <= base64_input_len; i++) {
		test_istream_set_size(input_data, i);
		while ((ret = i_stream_read(input)) > 0) ;
		if (ret == -1 && broken_input)
			break;
		test_assert(ret == 0);
	}
	if (ret == 0) {
		test_istream_set_allow_eof(input_data, TRUE);
		while ((ret = i_stream_read(input)) > 0) ;
	}
	test_assert(ret == -1);
	test_assert((input->stream_errno == 0 && !broken_input) ||
		    (input->stream_errno == EINVAL && broken_input));

	data = i_stream_get_data(input, &size);
	test_assert(size == strlen(output));
	if (size > 0)
		test_assert(memcmp(data, output, size) == 0);
	i_stream_unref(&input);
	i_stream_unref(&input_data);
}

static ssize_t i_stream_nonuls_read(struct istream_private *stream)
{
	struct nonuls_istream *nstream = (struct nonuls_istream *)stream;
	const unsigned char *data, *p;
	size_t i, size, avail_size;
	int ret;

	if ((ret = i_stream_read_parent(stream)) <= 0)
		return ret;

	data = i_stream_get_data(stream->parent, &size);
	if (!i_stream_try_alloc(stream, size, &avail_size))
		return -2;
	if (size > avail_size)
		size = avail_size;
	i_assert(size > 0);

	p = memchr(data, '\0', size);
	if (p == NULL) {
		/* no NULs in this block */
		memcpy(stream->w_buffer+stream->pos, data, size);
	} else {
		i = p-data;
		memcpy(stream->w_buffer+stream->pos, data, i);
		for (; i < size; i++) {
			stream->w_buffer[stream->pos+i] = data[i] == '\0' ?
				nstream->replace_chr : data[i];
		}
	}
	stream->pos += size;
	i_stream_skip(stream->parent, size);
	return size;
}

static void tee_streams_update_buffer(struct tee_istream *tee)
{
	struct tee_child_istream *tstream = tee->children;
	const unsigned char *data;
	size_t size, old_used;

	data = i_stream_get_data(tee->input, &size);
	for (; tstream != NULL; tstream = tstream->next) {
		if (tstream->istream.istream.closed) {
			tstream->istream.skip = tstream->istream.pos = 0;
			continue;
		}
		old_used = tstream->istream.pos - tstream->istream.skip;

		tstream->istream.buffer = data;
		i_assert(tstream->istream.istream.v_offset >= tee->input->v_offset);
		tstream->istream.skip = tstream->istream.istream.v_offset -
			tee->input->v_offset;
		i_assert(tstream->istream.skip + old_used <= size);
		tstream->istream.pos = tstream->istream.skip + old_used;

		tstream->istream.parent_expected_offset =
			tee->input->v_offset;
		tstream->istream.access_counter =
			tee->input->real_stream->access_counter;
	}
}

static ssize_t i_stream_base64_encoder_read(struct istream_private *stream)
{
	struct base64_encoder_istream *bstream =
		(struct base64_encoder_istream *)stream;
	size_t pre_count, post_count, size;
	int ret;

	do {
		ret = i_stream_read_parent(stream);
		if (ret <= 0)
			return ret;
		(void)i_stream_get_data(stream->parent, &size);
	} while (size < 4 && !stream->parent->eof);

	/* encode as many lines as fits into destination buffer */
	pre_count = stream->pos - stream->skip;
	while (i_stream_base64_try_encode_line(bstream)) ;
	post_count = stream->pos - stream->skip;

	if (pre_count == post_count) {
		i_assert(stream->buffer_size - stream->pos < 4);
		return -2;
	}

	i_assert(post_count > pre_count);
	return post_count - pre_count;
}

static const struct stat *
i_stream_bzlib_stat(struct istream_private *stream, bool exact)
{
	struct bzlib_istream *zstream = (struct bzlib_istream *) stream;
	const struct stat *st;
	size_t size;

	st = i_stream_stat(stream->parent, exact);
	if (st == NULL)
		return NULL;

	if (zstream->eof_offset == (uoff_t)-1 && !exact)
		return st;

	stream->statbuf = *st;
	if (zstream->eof_offset == (uoff_t)-1) {
		uoff_t old_offset = stream->istream.v_offset;

		do {
			(void)i_stream_get_data(&stream->istream, &size);
			i_stream_skip(&stream->istream, size);
		} while (i_stream_read(&stream->istream) > 0);

		i_stream_seek(&stream->istream, old_offset);
		if (zstream->eof_offset == (uoff_t)-1)
			return NULL;
	}
	stream->statbuf.st_size = zstream->eof_offset;
	return &stream->statbuf;
}

static void
test_istream_base64_encoder_seek(const char *textin, const char *textout)
{
	unsigned int offset, len = strlen(textout);
	struct istream *input, *input_data;
	const unsigned char *data;
	size_t size;
	ssize_t ret;

	input_data = i_stream_create_from_data(textin, strlen(textin));
	input = i_stream_create_base64_encoder(input_data, 4, TRUE);

	while (i_stream_read(input) > 0) ;
	i_stream_skip(input, i_stream_get_data_size(input));

	for (offset = 0; offset < len; offset++) {
		i_stream_seek(input, offset);
		while ((ret = i_stream_read(input)) > 0) ;
		test_assert(ret == -1);

		data = i_stream_get_data(input, &size);
		test_assert(size == len-offset);
		test_assert(memcmp(data, textout+offset, size) == 0);
		i_stream_skip(input, size);
	}

	i_stream_unref(&input);
	i_stream_unref(&input_data);
}

static ssize_t i_stream_base64_decoder_read(struct istream_private *stream)
{
	struct base64_decoder_istream *bstream =
		(struct base64_decoder_istream *)stream;
	const unsigned char *data;
	size_t pre_count, post_count, size;
	int ret;

	do {
		ret = i_stream_read_parent(stream);
		if (ret <= 0) {
			if (ret < 0 && stream->istream.stream_errno == 0 &&
			    i_stream_get_data_size(stream->parent) > 0) {
				/* base64 input with a partial block */
				data = i_stream_get_data(stream->parent, &size);
				io_stream_set_error(&stream->iostream,
					"base64 input ends with a partial block: 0x%s",
					binary_to_hex(data, size));
				stream->istream.stream_errno = EINVAL;
			}
			return ret;
		}

		/* encode as many blocks as fits into destination buffer */
		pre_count = stream->pos - stream->skip;
		while ((ret = i_stream_base64_try_decode_block(bstream)) > 0) ;
		post_count = stream->pos - stream->skip;
	} while (ret == 0 && pre_count == post_count);

	if (ret < 0 && pre_count == post_count)
		return ret;

	i_assert(post_count > pre_count);
	return post_count - pre_count;
}

static void encode_test(const char *text, unsigned int chars_per_line,
			bool crlf, const char *output)
{
	unsigned int i, text_len = strlen(text);
	struct istream *input, *input_data;
	const unsigned char *data;
	size_t size;
	ssize_t ret;

	input_data = test_istream_create_data(text, text_len);
	test_istream_set_allow_eof(input_data, FALSE);
	input = i_stream_create_base64_encoder(input_data, chars_per_line, crlf);

	for (i = 1; i <= text_len; i++) {
		test_istream_set_size(input_data, i);
		while ((ret = i_stream_read(input)) > 0) ;
		test_assert(ret == 0);
	}
	test_istream_set_allow_eof(input_data, TRUE);
	while ((ret = i_stream_read(input)) > 0) ;
	test_assert(ret == -1);

	data = i_stream_get_data(input, &size);
	test_assert(size == strlen(output) && memcmp(data, output, size) == 0);

	i_stream_unref(&input);
	i_stream_unref(&input_data);
}

static void test_ostream_dot_one(const struct dot_test *test)
{
	struct istream *test_input;
	struct ostream *output, *test_output;
	buffer_t *output_data;
	const unsigned char *data;
	size_t size;
	ssize_t ret;

	test_input = test_istream_create(test->input);
	output_data = buffer_create_dynamic(pool_datastack_create(), 1024);
	test_output = o_stream_create_buffer(output_data);

	output = o_stream_create_dot(test_output, FALSE);

	while ((ret = i_stream_read(test_input)) > 0 || ret == -2) {
		data = i_stream_get_data(test_input, &size);
		ret = o_stream_send(output, data, size);
		test_assert(ret >= 0);
		if (ret <= 0)
			break;
		i_stream_skip(test_input, ret);
	}

	test_assert(test_input->eof);

	test_assert(o_stream_flush(output) > 0);
	o_stream_unref(&output);
	o_stream_unref(&test_output);

	test_assert(strcmp(str_c(output_data), test->output) == 0);

	i_stream_unref(&test_input);
}

static int json_parser_read_more(struct json_parser *parser)
{
	uoff_t cur_highwater = parser->input->v_offset +
		i_stream_get_data_size(parser->input);
	size_t size;
	ssize_t ret;

	i_assert(parser->highwater_offset <= cur_highwater);

	if (parser->error != NULL)
		return -1;

	if (parser->highwater_offset == cur_highwater) {
		ret = i_stream_read(parser->input);
		if (ret == -2) {
			parser->error = "Token too large";
			return -1;
		}
		if (ret <= 0)
			return ret;

		cur_highwater = parser->input->v_offset +
			i_stream_get_data_size(parser->input);
		i_assert(parser->highwater_offset < cur_highwater);
		parser->highwater_offset = cur_highwater;
	}

	parser->start = parser->data = i_stream_get_data(parser->input, &size);
	parser->end = parser->start + size;
	i_assert(size > 0);
	return 1;
}

static int program_client_program_output(struct program_client *pclient)
{
	struct istream *input = pclient->input;
	struct ostream *output = pclient->program_output;
	const unsigned char *data;
	size_t size;
	int ret = 0;

	if ((ret = o_stream_flush(output)) <= 0) {
		if (ret < 0)
			program_client_fail(pclient, PROGRAM_CLIENT_ERROR_IO);
		return ret;
	}

	if ( input != NULL && output != NULL ) {
		do {
			while ( (data=i_stream_get_data(input, &size)) != NULL ) {
				ssize_t sent;
	
				if ( (sent=o_stream_send(output, data, size)) < 0 ) {
					program_client_fail(pclient, PROGRAM_CLIENT_ERROR_IO);
					return -1;
				}
	
				if ( sent == 0 )
					return 0;
				i_stream_skip(input, sent);
			}
		} while ( (ret=i_stream_read(input)) > 0 );

		if ( ret == 0 )
			return 1;

		if ( ret < 0 ) {
			if ( !input->eof ) {
				program_client_fail(pclient, PROGRAM_CLIENT_ERROR_IO);
				return -1;
			} else if ( !i_stream_have_bytes_left(input) ) {
				i_stream_unref(&pclient->input);
				input = NULL;

				if ( (ret = o_stream_flush(output)) <= 0 ) {
					if ( ret < 0 )
						program_client_fail(pclient, PROGRAM_CLIENT_ERROR_IO);
					return ret;
				}
			} 
		}
	}

	if ( input == NULL ) {
		if ( !program_client_input_pending(pclient) ) {
			program_client_disconnect(pclient, FALSE);
		} else if (program_client_close_output(pclient) < 0) {
			program_client_fail(pclient, PROGRAM_CLIENT_ERROR_IO);
		}
	}
	return 1;
}

static ssize_t i_stream_limit_read(struct istream_private *stream)
{
	struct limit_istream *lstream = (struct limit_istream *) stream;
	uoff_t left;
	ssize_t ret;
	size_t pos;

	i_stream_seek(stream->parent, lstream->istream.parent_start_offset +
		      stream->istream.v_offset);

	if (stream->istream.v_offset +
	    (stream->pos - stream->skip) >= lstream->v_size) {
		stream->istream.eof = TRUE;
		return -1;
	}

	stream->pos -= stream->skip;
	stream->skip = 0;

	stream->buffer = i_stream_get_data(stream->parent, &pos);
	if (pos > stream->pos)
		ret = 0;
	else do {
		if ((ret = i_stream_read(stream->parent)) == -2)
			return -2;

		stream->istream.stream_errno = stream->parent->stream_errno;
		stream->istream.eof = stream->parent->eof;
		stream->buffer = i_stream_get_data(stream->parent, &pos);
	} while (pos <= stream->pos && ret > 0);

	if (lstream->v_size != (uoff_t)-1) {
		left = lstream->v_size - stream->istream.v_offset;
		if (pos >= left) {
			pos = left;
			stream->istream.eof = TRUE;
		}
	}

	ret = pos > stream->pos ? (ssize_t)(pos - stream->pos) :
		(ret == 0 ? 0 : -1);
	stream->pos = pos;
	i_assert(ret != -1 || stream->istream.eof ||
		 stream->istream.stream_errno != 0);
	return ret;
}

void login_proxy_detach(struct login_proxy *proxy)
{
	struct client *client = proxy->client;
	const unsigned char *data;
	size_t size;

	i_assert(proxy->client_fd == -1);
	i_assert(proxy->server_input != NULL);
	i_assert(proxy->server_output != NULL);

	if (proxy->to != NULL)
		timeout_remove(&proxy->to);

	proxy->client_fd = i_stream_get_fd(client->input);
	proxy->client_input = client->input;
	proxy->client_output = client->output;

	i_stream_set_persistent_buffers(client->input, FALSE);
	o_stream_set_max_buffer_size(client->output, (size_t)-1);
	o_stream_set_flush_callback(client->output, proxy_client_output, proxy);
	client->input = NULL;
	client->output = NULL;

	/* send all pending client input to proxy */
	data = i_stream_get_data(proxy->client_input, &size);
	if (size != 0)
		o_stream_nsend(proxy->server_output, data, size);

	/* from now on, just do dummy proxying */
	io_remove(&proxy->server_io);
	proxy->server_io =
		io_add(proxy->server_fd, IO_READ, server_input, proxy);
	proxy->client_io =
		io_add_istream(proxy->client_input, proxy_client_input, proxy);
	o_stream_set_flush_callback(proxy->server_output, server_output, proxy);
	i_stream_destroy(&proxy->server_input);

	if (proxy->notify_refresh_secs != 0) {
		proxy->to_notify =
			timeout_add(proxy->notify_refresh_secs * 1000,
				    login_proxy_notify, proxy);
	}

	proxy->callback = NULL;

	if (login_proxy_ipc_server == NULL) {
		login_proxy_ipc_server =
			ipc_server_init(LOGIN_PROXY_IPC_PATH,
					LOGIN_PROXY_IPC_NAME,
					login_proxy_ipc_cmd);
	}

	DLLIST_REMOVE(&login_proxies_pending, proxy);
	DLLIST_PREPEND(&login_proxies, proxy);

	client->fd = -1;
	client->login_proxy = NULL;
}