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_istream_tee_blocks(const char *str)
{
	struct istream *test_input, *child_input[CHILD_COUNT];
	struct tee_istream *tee;
	unsigned int i, j;

	test_input = test_istream_create(str);
	test_istream_set_max_buffer_size(test_input, TEST_BUF_SIZE);

	test_begin("istream tee blocks");
	tee = tee_i_stream_create(test_input);
	for (i = 0; i < CHILD_COUNT; i++)
		child_input[i] = tee_i_stream_create_child(tee);

	test_istream_set_allow_eof(test_input, FALSE);
	for (j = 1; j <= 3; j++) {
		test_istream_set_size(test_input, TEST_BUF_SIZE*j);
		for (i = 0; i < CHILD_COUNT; i++) {
			test_assert(i_stream_read(child_input[i]) == TEST_BUF_SIZE);
			i_stream_skip(child_input[i], TEST_BUF_SIZE);
		}
	}
	test_istream_set_allow_eof(test_input, TRUE);
	for (i = 0; i < CHILD_COUNT; i++) {
		test_assert(i_stream_read(child_input[i]) == -1);
		i_stream_unref(&child_input[i]);
	}
	i_stream_unref(&test_input);

	test_end();
}

static void test_istream_dot_error(const char *input_str, bool test_bufsize)
{
	struct istream *test_input, *input;
	unsigned int i;
	size_t outsize, input_len;
	uoff_t offset;
	int ret;

	test_input = test_istream_create(input_str);
	input = i_stream_create_dot(test_input, FALSE);

	input_len = strlen(input_str);

	if (!test_bufsize) {
		outsize = 1; i = 0;
		i_stream_set_max_buffer_size(input, outsize);
		test_istream_set_size(test_input, 1);
		while ((ret = i_stream_read(input)) != -1) {
			switch (ret) {
			case -2:
				i_stream_set_max_buffer_size(input, ++outsize);
				offset = test_input->v_offset;
				/* seek one byte backwards so stream gets
				   reset */
				i_stream_seek(test_input, offset - 1);
				/* go back to original position */
				test_istream_set_size(test_input, offset);
				i_stream_skip(test_input, 1);
				/* and finally allow reading one more byte */
				test_istream_set_size(test_input, offset + 1);
				break;
			case 0:
				test_istream_set_size(test_input, ++i);
				break;
			default:
				test_assert(ret > 0);
			}
		}
		test_istream_set_size(test_input, input_len);
		(void)i_stream_read(test_input);
	} else {
		test_istream_set_size(test_input, input_len);
		for (i = 1; i <= input_len; i++) {
			i_stream_set_max_buffer_size(input, i);
			(void)i_stream_read(input);
			(void)i_stream_read(input);
		}
		i_stream_set_max_buffer_size(input, i+1);
		(void)i_stream_read(input);
	}
	test_assert(input->stream_errno == EPIPE);

	i_stream_unref(&test_input);
	i_stream_unref(&input);
}

static int
mail_storage_try_copy(struct mail_save_context **_ctx, struct mail *mail)
{
	struct mail_save_context *ctx = *_ctx;
	struct mail_private *pmail = (struct mail_private *)mail;
	struct istream *input;

	ctx->copying_via_save = TRUE;

	/* we need to open the file in any case. caching metadata is unlikely
	   to help anything. */
	pmail->v.set_uid_cache_updates(mail, TRUE);

	if (mail_get_stream(mail, NULL, NULL, &input) < 0) {
		mail_copy_set_failed(ctx, mail, "stream");
		return -1;
	}
	if (mail_save_copy_default_metadata(ctx, mail) < 0)
		return -1;

	if (mailbox_save_begin(_ctx, input) < 0)
		return -1;

	do {
		if (mailbox_save_continue(ctx) < 0)
			break;
	} while (i_stream_read(input) != -1);

	if (input->stream_errno != 0) {
		mail_storage_set_critical(ctx->transaction->box->storage,
					  "copy: i_stream_read() failed: %m");
		return -1;
	}
	return 0;
}

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;
}

void i_stream_default_seek_nonseekable(struct istream_private *stream,
				       uoff_t v_offset, bool mark ATTR_UNUSED)
{
	size_t available;

	if (stream->istream.v_offset > v_offset)
		i_panic("stream %s doesn't support seeking backwards",
			i_stream_get_name(&stream->istream));

	while (stream->istream.v_offset < v_offset) {
		(void)i_stream_read(&stream->istream);

		available = stream->pos - stream->skip;
		if (available == 0) {
			if (stream->istream.stream_errno != 0) {
				/* read failed */
				return;
			}
			io_stream_set_error(&stream->iostream,
				"Can't seek to offset %"PRIuUOFF_T
				", because we have data only up to offset %"
				PRIuUOFF_T" (eof=%d)", v_offset,
				stream->istream.v_offset, stream->istream.eof ? 1 : 0);
			stream->istream.stream_errno = ESPIPE;
			return;
		}
		if (available <= v_offset - stream->istream.v_offset)
			i_stream_skip(&stream->istream, available);
		else {
			i_stream_skip(&stream->istream,
				      v_offset - stream->istream.v_offset);
		}
	}
}

char *i_stream_read_next_line(struct istream *stream)
{
	char *line;

	for (;;) {
		line = i_stream_next_line(stream);
		if (line != NULL)
			break;

		switch (i_stream_read(stream)) {
		case -2:
			io_stream_set_error(&stream->real_stream->iostream,
				"Line is too long (over %"PRIuSIZE_T
				" bytes at offset %"PRIuUOFF_T")",
				i_stream_get_data_size(stream), stream->v_offset);
			stream->stream_errno = errno = ENOBUFS;
			stream->eof = TRUE;
			return NULL;
		case -1:
			return i_stream_last_line(stream->real_stream);
		case 0:
			return NULL;
		}
	}
	return line;
}

int dbox_save_continue(struct mail_save_context *_ctx)
{
	struct dbox_save_context *ctx = (struct dbox_save_context *)_ctx;
	struct mail_storage *storage = _ctx->transaction->box->storage;

	if (ctx->failed)
		return -1;

	if (_ctx->data.attach != NULL)
		return index_attachment_save_continue(_ctx);

	do {
		if (o_stream_send_istream(_ctx->data.output, ctx->input) < 0) {
			if (!mail_storage_set_error_from_errno(storage)) {
				mail_storage_set_critical(storage,
					"write(%s) failed: %m",
					o_stream_get_name(_ctx->data.output));
			}
			ctx->failed = TRUE;
			return -1;
		}
		index_mail_cache_parse_continue(_ctx->dest_mail);

		/* both tee input readers may consume data from our primary
		   input stream. we'll have to make sure we don't return with
		   one of the streams still having data in them. */
	} while (i_stream_read(ctx->input) > 0);
	return 0;
}

static ssize_t i_stream_metawrap_read(struct istream_private *stream)
{
	struct metawrap_istream *mstream = (struct metawrap_istream *)stream;
	int ret;

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

	if (mstream->in_metadata) {
		ret = metadata_header_read(mstream);
		i_assert(stream->istream.v_offset == 0);
		mstream->start_offset = stream->parent->v_offset;
		if (ret <= 0)
			return ret;
		/* this stream is kind of silently skipping over the metadata */
		stream->abs_start_offset += mstream->start_offset;
		mstream->in_metadata = FALSE;
		if (mstream->pending_seek != 0) {
			i_stream_seek(&stream->istream, mstream->pending_seek);
			return i_stream_read(&stream->istream);
		}
	}
	/* after metadata header it's all just passthrough */
	return i_stream_read_copy_from_parent(&stream->istream);
}

static void client_input(struct client *client)
{
	const char *const *args, *error;
	int ret;

	if (client->to_pending != NULL)
		timeout_remove(&client->to_pending);

	switch (i_stream_read(client->input)) {
	case -2:
		i_error("BUG: Stats client sent too much data");
		client_destroy(&client);
		return;
	case -1:
		client_destroy(&client);
		return;
	}

	o_stream_cork(client->output);
	while ((args = client_read_next_line(client)) != NULL) {
		ret = client_handle_request(client, args, &error);
		if (ret < 0) {
			i_error("Stats client input error: %s", error);
			client_destroy(&client);
			return;
		}
		if (ret == 0) {
			o_stream_set_flush_pending(client->output, TRUE);
			io_remove(&client->io);
			break;
		}
		client->cmd_more = NULL;
	}
	o_stream_uncork(client->output);
}

int maildir_save_continue(struct mail_save_context *_ctx)
{
	struct maildir_save_context *ctx = (struct maildir_save_context *)_ctx;
	struct mail_storage *storage = &ctx->mbox->storage->storage;

	if (ctx->failed)
		return -1;

	do {
		if (o_stream_send_istream(_ctx->data.output, ctx->input) < 0) {
			if (!mail_storage_set_error_from_errno(storage)) {
				mail_storage_set_critical(storage,
					"o_stream_send_istream(%s/%s) "
					"failed: %m",
					ctx->tmpdir, ctx->file_last->tmp_name);
			}
			ctx->failed = TRUE;
			return -1;
		}
		if (ctx->cur_dest_mail != NULL)
			index_mail_cache_parse_continue(ctx->cur_dest_mail);

		/* both tee input readers may consume data from our primary
		   input stream. we'll have to make sure we don't return with
		   one of the streams still having data in them. */
	} while (i_stream_read(ctx->input) > 0);
	return 0;
}

static int i_stream_dot_read_some(struct dot_istream *dstream)
{
	struct istream_private *stream = &dstream->istream;
	size_t size, avail;
	ssize_t ret;

	size = i_stream_get_data_size(stream->parent);
	if (size == 0) {
		ret = i_stream_read(stream->parent);
		if (ret <= 0 && (ret != -2 || stream->skip == 0)) {
			if (stream->parent->stream_errno != 0) {
				stream->istream.stream_errno =
					stream->parent->stream_errno;
			} else if (ret < 0 && stream->parent->eof) {
				/* we didn't see "." line */
				io_stream_set_error(&stream->iostream,
					"dot-input stream ends without '.' line");
				stream->istream.stream_errno = EPIPE;
			}
			return ret;
		}
		size = i_stream_get_data_size(stream->parent);
		i_assert(size != 0);
	}

	if (!i_stream_try_alloc(stream, size, &avail))
		return -2;
	return 1;
}

static void notify_input(struct notify_connection *conn)
{
    const char *line;
    int ret;

    switch (i_stream_read(conn->input)) {
    case -2:
        /* buffer full */
        i_error("Client sent too long line");
        (void)notify_input_error(conn);
        return;
    case -1:
        /* disconnected */
        notify_connection_destroy(conn);
        return;
    }

    while ((line = i_stream_next_line(conn->input)) != NULL) {
        T_BEGIN {
            ret = notify_input_line(conn, line);
        } T_END;
        if (ret < 0) {
            if (!notify_input_error(conn))
                return;
        }
    }
}

uoff_t istream_raw_mbox_get_body_offset(struct istream *stream)
{
	struct raw_mbox_istream *rstream =
		(struct raw_mbox_istream *)stream->real_stream;
	uoff_t offset;

	i_assert(rstream->seeked);

	if (rstream->body_offset != (uoff_t)-1)
		return rstream->body_offset;

	offset = stream->v_offset;
	i_stream_seek(stream, rstream->hdr_offset);
	while (rstream->body_offset == (uoff_t)-1) {
		i_stream_skip(stream, i_stream_get_data_size(stream));

		if (i_stream_read(stream) < 0) {
			if (rstream->corrupted) {
				i_error("Unexpectedly lost From-line from mbox file "
					"%s at %"PRIuUOFF_T,
					i_stream_get_name(stream),
					rstream->from_offset);
			} else {
				i_assert(rstream->body_offset != (uoff_t)-1);
			}
			break;
		}
	}

	i_stream_seek(stream, offset);
	return rstream->body_offset;
}

static void postfix_input(struct auth_postfix_connection *conn)
{
 	char *line;
	bool ret;

	switch (i_stream_read(conn->input)) {
	case 0:
		return;
	case -1:
		/* disconnected */
                auth_postfix_connection_destroy(&conn);
		return;
	case -2:
		/* buffer full */
		i_error("BUG: Postfix sent us more than %d bytes",
			(int)MAX_INBUF_SIZE);
                auth_postfix_connection_destroy(&conn);
		return;
	}

	while ((line = i_stream_next_line(conn->input)) != NULL) {
		T_BEGIN {
			ret = auth_postfix_input_line(conn, line);
		} T_END;
		if (!ret) {
			auth_postfix_connection_destroy(&conn);
			return;
		}
	}
}

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 i_stream_metawrap_stat(struct istream_private *stream, bool exact)
{
	struct metawrap_istream *mstream = (struct metawrap_istream *)stream;
	const struct stat *st;
	int ret;

	if (i_stream_stat(stream->parent, exact, &st) < 0) {
		stream->istream.stream_errno = stream->parent->stream_errno;
		return -1;
	}
	stream->statbuf = *st;

	if (mstream->in_metadata) {
		ret = i_stream_read(&stream->istream);
		if (ret < 0 && stream->istream.stream_errno != 0)
			return -1;
		if (ret == 0) {
			stream->statbuf.st_size = -1;
			return 0;
		}
	}
	i_assert((uoff_t)stream->statbuf.st_size >= mstream->start_offset);
	stream->statbuf.st_size -= mstream->start_offset;
	return 0;
}

static const char *next_line(struct mailbox_list *list, const char *path,
			     struct istream *input, bool *failed_r,
			     bool ignore_estale)
{
	const char *line;

	*failed_r = FALSE;

	while ((line = i_stream_next_line(input)) == NULL) {
                switch (i_stream_read(input)) {
		case -1:
                        if (input->stream_errno != 0 &&
                            (input->stream_errno != ESTALE || !ignore_estale)) {
                                subswrite_set_syscall_error(list,
							    "read()", path);
                                *failed_r = TRUE;
                        }
			return NULL;
		case -2:
			/* mailbox name too large */
			mailbox_list_set_critical(list,
				"Subscription file %s contains lines longer "
				"than %u characters", path,
				(unsigned int)list->mailbox_name_max_length);
			*failed_r = TRUE;
			return NULL;
		}
	}

	return line;
}

static int
i_stream_lzma_stat(struct istream_private *stream, bool exact)
{
	struct lzma_istream *zstream = (struct lzma_istream *) stream;
	const struct stat *st;
	size_t size;

	if (i_stream_stat(stream->parent, exact, &st) < 0) {
		stream->istream.stream_errno = stream->parent->stream_errno;
		return -1;
	}
	stream->statbuf = *st;

	/* when exact=FALSE always return the parent stat's size, even if we
	   know the exact value. this is necessary because otherwise e.g. mbox
	   code can see two different values and think that a compressed mbox
	   file keeps changing. */
	if (!exact)
		return 0;

	if (zstream->stream_size == (uoff_t)-1) {
		uoff_t old_offset = stream->istream.v_offset;

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

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

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;
}