static int index_mail_parse_body(struct index_mail *mail,
				 enum index_cache_field field)
{
	struct index_mail_data *data = &mail->data;
	uoff_t old_offset;
	int ret;

	i_assert(data->parser_ctx != NULL);

	old_offset = data->stream->v_offset;
	i_stream_seek(data->stream, data->hdr_size.physical_size);

	if (data->save_bodystructure_body) {
		/* bodystructure header is parsed, we want the body's mime
		   headers too */
		i_assert(!data->save_bodystructure_header);
		message_parser_parse_body(data->parser_ctx,
					  parse_bodystructure_part_header,
					  mail->data_pool);
		data->save_bodystructure_body = FALSE;
		data->parsed_bodystructure = TRUE;
	} else {
		message_parser_parse_body(data->parser_ctx,
			null_message_part_header_callback, NULL);
	}
	ret = index_mail_stream_check_failure(mail);
	if (index_mail_parse_body_finish(mail, field) < 0)
		ret = -1;

	i_stream_seek(data->stream, old_offset);
	return ret;
}

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 test_message_parser_small_blocks(void)
{
	struct message_parser_ctx *parser;
	struct istream *input;
	struct message_part *parts, *parts2;
	struct message_block block;
	unsigned int i, end_of_headers_idx;
	pool_t pool;
	int ret;

	test_begin("message parser in small blocks");
	pool = pool_alloconly_create("message parser", 10240);
	input = test_istream_create(test_msg);

	/* full parsing */
	parser = message_parser_init(pool, input, 0, 0);
	while ((ret = message_parser_parse_next_block(parser, &block)) > 0) ;
	test_assert(ret < 0);
	test_assert(message_parser_deinit(&parser, &parts) == 0);

	/* parsing in small blocks */
	i_stream_seek(input, 0);
	test_istream_set_allow_eof(input, FALSE);

	parser = message_parser_init(pool, input, 0, 0);
	for (i = 1; i <= TEST_MSG_LEN*2+1; i++) {
		test_istream_set_size(input, i/2);
		if (i > TEST_MSG_LEN*2)
			test_istream_set_allow_eof(input, TRUE);
		while ((ret = message_parser_parse_next_block(parser,
							      &block)) > 0) ;
		test_assert((ret == 0 && i <= TEST_MSG_LEN*2) ||
			    (ret < 0 && i > TEST_MSG_LEN*2));
	}
	test_assert(message_parser_deinit(&parser, &parts2) == 0);
	test_assert(msg_parts_cmp(parts, parts2));

	/* parsing in small blocks from preparsed parts */
	i_stream_seek(input, 0);
	test_istream_set_allow_eof(input, FALSE);

	end_of_headers_idx = (strstr(test_msg, "\n-----") - test_msg);
	parser = message_parser_init_from_parts(parts, input, 0,
					MESSAGE_PARSER_FLAG_SKIP_BODY_BLOCK);
	for (i = 1; i <= TEST_MSG_LEN*2+1; i++) {
		test_istream_set_size(input, i/2);
		if (i > TEST_MSG_LEN*2)
			test_istream_set_allow_eof(input, TRUE);
		while ((ret = message_parser_parse_next_block(parser,
							      &block)) > 0) ;
		test_assert((ret == 0 && i/2 <= end_of_headers_idx) ||
			    (ret < 0 && i/2 > end_of_headers_idx));
	}
	test_assert(message_parser_deinit(&parser, &parts2) == 0);
	test_assert(msg_parts_cmp(parts, parts2));

	i_stream_unref(&input);
	pool_unref(&pool);
	test_end();
}

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_raw_mbox_read(&rstream->istream) < 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;
}

uoff_t istream_raw_mbox_get_body_size(struct istream *stream,
				      uoff_t expected_body_size)
{
	struct raw_mbox_istream *rstream =
		(struct raw_mbox_istream *)stream->real_stream;
	const unsigned char *data;
	size_t size;
	uoff_t old_offset, body_size, next_body_offset;

	i_assert(rstream->seeked);
	i_assert(rstream->hdr_offset != (uoff_t)-1);

	(void)istream_raw_mbox_get_body_offset(stream);
	body_size = rstream->mail_size == (uoff_t)-1 ? (uoff_t)-1 :
		rstream->mail_size - (rstream->body_offset -
				      rstream->hdr_offset);
	old_offset = stream->v_offset;
	if (expected_body_size != (uoff_t)-1) {
		/* if we already have the existing body size, use it as long as
		   it's >= expected body_size. otherwise the previous parsing
		   may have stopped at a From_-line that belongs to the body. */
		if (body_size != (uoff_t)-1 && body_size >= expected_body_size)
			return body_size;

		next_body_offset = rstream->body_offset + expected_body_size;
		/* If header_missing_eoh is set, the message body begins with
		   a From_-line and the body_offset is pointing to the line
		   *before* the first line of the body, i.e. the empty line
		   separating the headers from the body. If that is the case,
		   we'll have to skip over the empty line to get the correct
		   next_body_offset. */
		if (rstream->header_missing_eoh) {
			i_assert(body_size == 0);
			next_body_offset += rstream->crlf_ending ? 2 : 1;
		}

		i_stream_seek(rstream->istream.parent, next_body_offset);
		if (istream_raw_mbox_is_valid_from(rstream) > 0) {
			rstream->mail_size =
				next_body_offset - rstream->hdr_offset;
			i_stream_seek(stream, old_offset);
			return expected_body_size;
		}
		/* invalid expected_body_size */
	}
	if (body_size != (uoff_t)-1)
		return body_size;

	/* have to read through the message body */
	while (i_stream_read_data(stream, &data, &size, 0) > 0)
		i_stream_skip(stream, size);
	i_stream_seek(stream, old_offset);

	i_assert(rstream->mail_size != (uoff_t)-1);
	return rstream->mail_size -
		(rstream->body_offset - rstream->hdr_offset);
}

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 int seek_partial(unsigned int select_counter, unsigned int uid,
			struct partial_cache *partial, struct istream *stream,
			uoff_t virtual_skip, bool *cr_skipped_r)
{
	if (select_counter == partial->select_counter && uid == partial->uid &&
	    stream->v_offset == partial->physical_start &&
	    virtual_skip >= partial->pos.virtual_size) {
		/* we can use the cache */
		virtual_skip -= partial->pos.virtual_size;
	} else {
		partial->select_counter = select_counter;
		partial->uid = uid;
		partial->physical_start = stream->v_offset;
		partial->cr_skipped = FALSE;
		memset(&partial->pos, 0, sizeof(partial->pos));
	}

	i_stream_seek(stream, partial->physical_start +
		      partial->pos.physical_size);
	if (message_skip_virtual(stream, virtual_skip, &partial->pos,
				 partial->cr_skipped, cr_skipped_r) < 0)
		return -1;

	partial->cr_skipped = FALSE;
	return 0;
}

static int o_stream_temp_dup_istream(struct temp_ostream *outstream,
				     struct istream *instream)
{
	uoff_t in_size;
	off_t ret;

	if (!instream->readable_fd || i_stream_get_fd(instream) == -1)
		return 0;

	if (i_stream_get_size(instream, TRUE, &in_size) <= 0) {
		if (outstream->dupstream != NULL)
			return o_stream_temp_dup_cancel(outstream);
		return 0;
	}

	if (outstream->dupstream == NULL) {
		outstream->dupstream = instream;
		outstream->dupstream_start_offset = instream->v_offset;
		i_stream_ref(outstream->dupstream);
	} else {
		if (outstream->dupstream != instream ||
		    outstream->dupstream_offset != instream->v_offset ||
		    outstream->dupstream_offset > in_size)
			return o_stream_temp_dup_cancel(outstream);
	}
	ret = in_size - instream->v_offset;
	i_stream_seek(instream, in_size);
	outstream->dupstream_offset = instream->v_offset;
	return ret;
}

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 test_message_header_parser(void)
{
	static enum message_header_parser_flags max_hdr_flags =
		MESSAGE_HEADER_PARSER_FLAG_SKIP_INITIAL_LWSP |
		MESSAGE_HEADER_PARSER_FLAG_DROP_CR |
		MESSAGE_HEADER_PARSER_FLAG_CLEAN_ONELINE;
	enum message_header_parser_flags hdr_flags;
	struct message_header_parser_ctx *parser;
	struct message_size hdr_size;
	struct istream *input;

	test_begin("message header parser");
	input = test_istream_create(test1_msg);

	for (hdr_flags = 0; hdr_flags <= max_hdr_flags; hdr_flags++) {
		i_stream_seek(input, 0);
		parser = message_parse_header_init(input, &hdr_size, hdr_flags);
		test_message_header_parser_one(parser, hdr_flags);
		message_parse_header_deinit(&parser);
	}
	test_assert(hdr_size.physical_size == strlen(test1_msg)-TEST1_MSG_BODY_LEN);
	test_assert(hdr_size.virtual_size == strlen(test1_msg) - TEST1_MSG_BODY_LEN + 4);

	i_stream_unref(&input);
	test_end();
}

int imap_msgpart_url_read_part(struct imap_msgpart_url *mpurl,
			       struct imap_msgpart_open_result *result_r,
			       const char **error_r)
{
	struct mail *mail;
	int ret;

	if (mpurl->result.input != NULL) {
		i_stream_seek(mpurl->result.input, 0);
		*result_r = mpurl->result;
		return 1;
	}

	/* open mail if it is not yet open */
	ret = imap_msgpart_url_open_mail(mpurl, &mail, error_r);
	if (ret <= 0)
		return ret;

	/* open the referenced part as a stream */
	ret = imap_msgpart_open(mail, mpurl->part, result_r);
	if (ret < 0) {
		*error_r = mailbox_get_last_error(mpurl->box, NULL);
		return ret;
	}

	mpurl->result = *result_r;
	return 1;
}

int dbox_attachment_file_get_stream(struct dbox_file *file,
				    struct istream **stream)
{
	const char *ext_refs, *error;
	int ret;

	/* need to read metadata in case there are external references */
	if ((ret = dbox_file_metadata_read(file)) <= 0)
		return ret;

	i_stream_seek(file->input, file->cur_offset + file->msg_header_size);

	ext_refs = dbox_file_metadata_get(file, DBOX_METADATA_EXT_REF);
	if (ext_refs == NULL)
		return 1;

	/* we have external references. */
	T_BEGIN {
		ret = dbox_attachment_file_get_stream_from(file, ext_refs,
							   stream, &error);
		if (ret == 0) {
			dbox_file_set_corrupted(file,
				"Corrupted ext-refs metadata %s: %s",
				ext_refs, error);
		}
	} T_END;
	return ret;
}

static int imapc_mail_get_hdr_hash(struct index_mail *imail)
{
	struct istream *input;
	const unsigned char *data;
	size_t size;
	uoff_t old_offset;
	struct sha1_ctxt sha1_ctx;
	unsigned char sha1_output[SHA1_RESULTLEN];
	const char *sha1_str;

	sha1_init(&sha1_ctx);
	old_offset = imail->data.stream == NULL ? 0 :
		imail->data.stream->v_offset;
	if (mail_get_hdr_stream(&imail->mail.mail, NULL, &input) < 0)
		return -1;
	while (i_stream_read_data(input, &data, &size, 0) > 0) {
		sha1_loop(&sha1_ctx, data, size);
		i_stream_skip(input, size);
	}
	i_stream_seek(imail->data.stream, old_offset);
	sha1_result(&sha1_ctx, sha1_output);

	sha1_str = binary_to_hex(sha1_output, sizeof(sha1_output));
	imail->data.guid = p_strdup(imail->mail.data_pool, sha1_str);
	return 0;
}

static int
mail_crypt_istream_opened(struct mail *_mail, struct istream **stream)
{
	struct mail_private *mail = (struct mail_private *)_mail;
	struct mail_user *user = _mail->box->storage->user;
	struct mail_crypt_user *muser = MAIL_CRYPT_USER_CONTEXT(user);
	struct mail_crypt_cache *cache = &muser->cache;
	union mail_module_context *mmail = MAIL_CRYPT_MAIL_CONTEXT(mail);
	struct istream *input;

	if (_mail->uid > 0 && cache->uid == _mail->uid && cache->box == _mail->box) {
		/* use the cached stream. when doing partial reads it should
		   already be seeked into the wanted offset. */
		i_stream_unref(stream);
		i_stream_seek(cache->input, 0);
		*stream = i_stream_create_limit(cache->input, (uoff_t)-1);
		return mmail->super.istream_opened(_mail, stream);
	}

	/* decryption is the outmost stream, so add it before others
	   (e.g. zlib) */
	if (!mail_crypt_is_stream_encrypted(*stream))
		return mmail->super.istream_opened(_mail, stream);

	input = *stream;
	*stream = i_stream_create_decrypt_callback(input,
				mail_crypt_istream_get_private_key, _mail);
	i_stream_unref(&input);

	*stream = mail_crypt_cache_open(muser, _mail, *stream);
	return mmail->super.istream_opened(_mail, stream);
}

static int dbox_file_read_header(struct dbox_file *file)
{
	const char *line;
	unsigned int hdr_size;
	int ret;

	i_stream_seek(file->input, 0);
	line = i_stream_read_next_line(file->input);
	if (line == NULL) {
		if (file->input->stream_errno == 0) {
			dbox_file_set_corrupted(file,
				"EOF while reading file header");
			return 0;
		}

		dbox_file_set_syscall_error(file, "read()");
		return -1;
	}
	hdr_size = file->input->v_offset;
	T_BEGIN {
		ret = dbox_file_parse_header(file, line) < 0 ? 0 : 1;
	} T_END;
	if (ret > 0)
		file->file_header_size = hdr_size;
	return ret;
}

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

void http_client_request_resubmit(struct http_client_request *req)
{
	i_assert(!req->payload_wait);

	http_client_request_debug(req, "Resubmitting request");

	/* rewind payload stream */
	if (req->payload_input != NULL && req->payload_size > 0) {
		if (req->payload_input->v_offset != req->payload_offset &&
			!req->payload_input->seekable) {
			http_client_request_error(&req,
				HTTP_CLIENT_REQUEST_ERROR_ABORTED,
				"Resubmission failed: Cannot resend payload; stream is not seekable");
			return;
		} else {
			i_stream_seek(req->payload_input, req->payload_offset);
		}
	}

	/* drop payload output stream from previous attempt */
	if (req->payload_output != NULL)
		o_stream_unref(&req->payload_output);

	req->conn = NULL;
	req->peer = NULL;
	req->state = HTTP_REQUEST_STATE_QUEUED;
	http_client_host_submit_request(req->host, req);
}

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
dbox_file_copy_metadata(struct dbox_file *file, struct ostream *output,
			bool *have_guid_r)
{
	const char *line;
	uoff_t prev_offset = file->input->v_offset;

	*have_guid_r = FALSE;
	while ((line = i_stream_read_next_line(file->input)) != NULL) {
		if (*line == DBOX_METADATA_OLDV1_SPACE || *line == '\0') {
			/* end of metadata */
			return;
		}
		if (*line < 32) {
			/* broken - possibly a new pre-magic block */
			i_stream_seek(file->input, prev_offset);
			return;
		}
		if (*line == DBOX_METADATA_VIRTUAL_SIZE) {
			/* it may be wrong - recreate it */
			continue;
		}
		if (*line == DBOX_METADATA_GUID)
			*have_guid_r = TRUE;
		o_stream_send_str(output, line);
		o_stream_send_str(output, "\n");
	}
}

ssize_t i_stream_read(struct istream *stream)
{
	struct istream_private *_stream = stream->real_stream;
	size_t old_size;
	ssize_t ret;

	if (unlikely(stream->closed || stream->stream_errno != 0)) {
		stream->eof = TRUE;
		errno = stream->stream_errno;
		return -1;
	}

	stream->eof = FALSE;

	if (_stream->parent != NULL)
		i_stream_seek(_stream->parent, _stream->parent_expected_offset);

	old_size = _stream->pos - _stream->skip;
	ret = _stream->read(_stream);
	i_assert(old_size <= _stream->pos - _stream->skip);
	switch (ret) {
	case -2:
		i_assert(_stream->skip != _stream->pos);
		break;
	case -1:
		if (stream->stream_errno != 0) {
			/* error handling should be easier if we now just
			   assume the stream is now at EOF */
			stream->eof = TRUE;
			errno = stream->stream_errno;
		} else {
			i_assert(stream->eof);
			i_assert(old_size == _stream->pos - _stream->skip);
		}
		break;
	case 0:
		i_assert(!stream->blocking);
		break;
	default:
		i_assert(ret > 0);
		i_assert(_stream->skip < _stream->pos);
		i_assert((size_t)ret+old_size == _stream->pos - _stream->skip);
		break;
	}

	if (stream->stream_errno != 0) {
		/* error handling should be easier if we now just
		   assume the stream is now at EOF. Note that we could get here
		   even if read() didn't return -1, although that's a little
		   bit sloppy istream implementation. */
		stream->eof = TRUE;
	}

	i_stream_update(_stream);
	/* verify that parents' access_counters are valid. the parent's
	   i_stream_read() should guarantee this. */
	i_assert(!i_stream_is_buffer_invalid(_stream));
	return ret;
}