static void test_compress_file(const char *in_path, const char *out_path)
{
	const struct compression_handler *handler;
	struct istream *input, *file_input;
	struct ostream *output, *file_output;
	int fd_in, fd_out;
	struct sha1_ctxt sha1;
	unsigned char output_sha1[SHA1_RESULTLEN], input_sha1[SHA1_RESULTLEN];
	const unsigned char *data;
	size_t size;
	ssize_t ret;

	handler = compression_lookup_handler_from_ext(out_path);
	if (handler == NULL)
		i_fatal("Can't detect compression algorithm from path %s", out_path);
	if (handler->create_ostream == NULL)
		i_fatal("Support not compiled in for %s", handler->name);

	/* write the compressed output file */
	fd_in = open(in_path, O_RDONLY);
	if (fd_in == -1)
		i_fatal("open(%s) failed: %m", in_path);
	fd_out = open(out_path, O_TRUNC | O_CREAT | O_RDWR, 0600);
	if (fd_out == -1)
		i_fatal("creat(%s) failed: %m", out_path);

	sha1_init(&sha1);
	file_output = o_stream_create_fd_file(fd_out, 0, FALSE);
	output = handler->create_ostream(file_output, 1);
	input = i_stream_create_fd_autoclose(&fd_in, IO_BLOCK_SIZE);
	while (i_stream_read_data(input, &data, &size, 0) > 0) {
		sha1_loop(&sha1, data, size);
		o_stream_nsend(output, data, size);
		i_stream_skip(input, size);
	}
	if (o_stream_nfinish(output) < 0) {
		i_fatal("write(%s) failed: %s",
			out_path, o_stream_get_error(output));
	}
	i_stream_destroy(&input);
	o_stream_destroy(&output);
	o_stream_destroy(&file_output);
	sha1_result(&sha1, output_sha1);

	/* verify that we can read the compressed file */
	sha1_init(&sha1);
	file_input = i_stream_create_fd(fd_out, IO_BLOCK_SIZE, FALSE);
	input = handler->create_istream(file_input, FALSE);
	while ((ret = i_stream_read_data(input, &data, &size, 0)) > 0) {
		sha1_loop(&sha1, data, size);
		i_stream_skip(input, size);
	}
	i_stream_destroy(&input);
	i_stream_destroy(&file_input);
	sha1_result(&sha1, input_sha1);

	if (memcmp(input_sha1, output_sha1, sizeof(input_sha1)) != 0)
		i_fatal("Decompression couldn't get the original input");
	i_close_fd(&fd_out);
}

static struct istream *mail_raw_create_stream
(struct mail_user *ruser, int fd, time_t *mtime_r, const char **sender)
{
	struct istream *input, *input2, *input_list[2];
	const unsigned char *data;
	size_t i, size;
	int ret, tz;
	char *env_sender = NULL;

	*mtime_r = (time_t)-1;
	fd_set_nonblock(fd, FALSE);

	input = i_stream_create_fd(fd, 4096, FALSE);
	input->blocking = TRUE;
	/* If input begins with a From-line, drop it */
	ret = i_stream_read_data(input, &data, &size, 5);
	if (ret > 0 && size >= 5 && memcmp(data, "From ", 5) == 0) {
		/* skip until the first LF */
		i_stream_skip(input, 5);
		while ( i_stream_read_data(input, &data, &size, 0) > 0 ) {
			for (i = 0; i < size; i++) {
				if (data[i] == '\n')
					break;
			}
			if (i != size) {
				(void)mbox_from_parse(data, i, mtime_r, &tz, &env_sender);
				i_stream_skip(input, i + 1);
				break;
			}
			i_stream_skip(input, size);
		}
	}

	if (env_sender != NULL && sender != NULL) {
		*sender = t_strdup(env_sender);
	}
	i_free(env_sender);

	if (input->v_offset == 0) {
		input2 = input;
		i_stream_ref(input2);
	} else {
		input2 = i_stream_create_limit(input, (uoff_t)-1);
	}
	i_stream_unref(&input);

	input_list[0] = input2; input_list[1] = NULL;
	input = i_stream_create_seekable(input_list, MAIL_MAX_MEMORY_BUFFER,
		seekable_fd_callback, (void*)ruser);
	i_stream_unref(&input2);
	return input;
}

int message_get_body_size(struct istream *input, struct message_size *body,
			  bool *has_nuls_r)
{
	const unsigned char *msg;
	size_t i, size, missing_cr_count;
	int ret;

	memset(body, 0, sizeof(struct message_size));
	*has_nuls_r = FALSE;

	missing_cr_count = 0;
	if ((ret = i_stream_read_data(input, &msg, &size, 0)) <= 0)
		return ret < 0 && input->stream_errno != 0 ? -1 : 0;

	if (msg[0] == '\n')
		missing_cr_count++;

	do {
		for (i = 1; i < size; i++) {
			if (msg[i] > '\n')
				continue;

			if (msg[i] == '\n') {
				if (msg[i-1] != '\r') {
					/* missing CR */
					missing_cr_count++;
				}

				/* increase after making sure we didn't break
				   at virtual \r */
				body->lines++;
			} else if (msg[i] == '\0') {
				*has_nuls_r = TRUE;
			}
		}

		/* leave the last character, it may be \r */
		i_stream_skip(input, i - 1);
		body->physical_size += i - 1;
	} while (i_stream_read_data(input, &msg, &size, 1) > 0);

	ret = input->stream_errno != 0 ? -1 : 0;

	i_stream_skip(input, 1);
	body->physical_size++;

	body->virtual_size = body->physical_size + missing_cr_count;
	i_assert(body->virtual_size >= body->physical_size);
	return ret;
}

off_t io_stream_copy(struct ostream *outstream, struct istream *instream)
{
	uoff_t start_offset;
	struct const_iovec iov;
	const unsigned char *data;
	ssize_t ret;

	start_offset = instream->v_offset;
	do {
		(void)i_stream_read_data(instream, &data, &iov.iov_len, 0);
		if (iov.iov_len == 0) {
			/* all sent */
			if (instream->stream_errno != 0)
				return -1;
			break;
		}

		iov.iov_base = data;
		ret = o_stream_sendv(outstream, &iov, 1);
		if (ret <= 0) {
			if (ret == 0)
				break;
			return -1;
		}
		i_stream_skip(instream, ret);
	} while ((size_t)ret == iov.iov_len);

	return (off_t)(instream->v_offset - start_offset);
}

static int i_stream_zlib_read_trailer(struct zlib_istream *zstream)
{
	struct istream_private *stream = &zstream->istream;
	const unsigned char *data;
	size_t size;
	int ret;

	ret = i_stream_read_data(stream->parent, &data, &size,
				 GZ_TRAILER_SIZE-1);
	if (size == zstream->prev_size) {
		if (ret == -1) {
			if (zstream->log_errors)
				zlib_read_error(zstream, "missing gz trailer");
			stream->istream.stream_errno = EINVAL;
		}
		return ret;
	}
	zstream->prev_size = size;

	if (size < GZ_TRAILER_SIZE)
		return 0;

	if (data_get_uint32(data) != zstream->crc32) {
		if (zstream->log_errors) {
			zlib_read_error(zstream,
					"gz trailer has wrong CRC value");
		}
		stream->istream.stream_errno = EINVAL;
		return -1;
	}
	i_stream_skip(stream->parent, GZ_TRAILER_SIZE);
	zstream->prev_size = 0;
	zstream->trailer_read = TRUE;
	return 1;
}

static void cmd_dump_imapzlib(int argc ATTR_UNUSED, char *argv[])
{
	struct istream *input, *input2;
	const unsigned char *data;
	size_t size;
	const char *line;
	int fd;

	fd = open(argv[1], O_RDONLY);
	if (fd < 0)
		i_fatal("open(%s) failed: %m", argv[1]);
	input = i_stream_create_fd(fd, 1024*32, TRUE);
	while ((line = i_stream_read_next_line(input)) != NULL) {
		/* skip tag */
		printf("%s\r\n", line);
		while (*line != ' ' && *line != '\0') line++;
		if (*line == '\0')
			continue;
		line++;

		if (strcmp(line, "OK Begin compression.") == 0 ||
		    strcasecmp(line, "COMPRESS DEFLATE") == 0)
			break;
	}

	input2 = i_stream_create_deflate(input, TRUE);
	i_stream_unref(&input);

	while (i_stream_read_data(input2, &data, &size, 0) != -1) {
		fwrite(data, 1, size, stdout);
		i_stream_skip(input2, size);
	}
	i_stream_unref(&input2);
	fflush(stdout);
}

int dsync_mail_get_hdr_hash(struct mail *mail, const char **hdr_hash_r)
{
	struct istream *hdr_input, *input;
	struct mailbox_header_lookup_ctx *hdr_ctx;
	struct md5_context md5_ctx;
	unsigned char md5_result[MD5_RESULTLEN];
	const unsigned char *data;
	size_t size;
	int ret = 0;

	hdr_ctx = mailbox_header_lookup_init(mail->box, hashed_headers);
	ret = mail_get_header_stream(mail, hdr_ctx, &hdr_input);
	mailbox_header_lookup_unref(&hdr_ctx);
	if (ret < 0)
		return -1;

	input = i_stream_create_lf(hdr_input);

	md5_init(&md5_ctx);
	while (!i_stream_is_eof(input)) {
		if (i_stream_read_data(input, &data, &size, 0) == -1)
			break;
		if (size == 0)
			break;
		md5_update(&md5_ctx, data, size);
		i_stream_skip(input, size);
	}
	if (input->stream_errno != 0)
		ret = -1;
	i_stream_unref(&input);

	md5_final(&md5_ctx, md5_result);
	*hdr_hash_r = binary_to_hex(md5_result, sizeof(md5_result));
	return ret;
}

static int
mdbox_file_read_metadata_hdr(struct dbox_file *file,
			     struct dbox_metadata_header *meta_hdr_r)
{
	const unsigned char *data;
	size_t size;
	int ret;

	ret = i_stream_read_data(file->input, &data, &size,
				 sizeof(*meta_hdr_r));
	if (ret <= 0) {
		i_assert(ret == -1);
		if (file->input->stream_errno == 0) {
			dbox_file_set_corrupted(file, "missing metadata");
			return 0;
		}
		mail_storage_set_critical(&file->storage->storage,
			"read(%s) failed: %m", file->cur_path);
		return -1;
	}

	memcpy(meta_hdr_r, data, sizeof(*meta_hdr_r));
	if (memcmp(meta_hdr_r->magic_post, DBOX_MAGIC_POST,
		   sizeof(meta_hdr_r->magic_post)) != 0) {
		dbox_file_set_corrupted(file, "invalid metadata magic");
		return 0;
	}
	i_stream_skip(file->input, sizeof(*meta_hdr_r));
	return 1;
}

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 client_run_url(struct client *client)
{
	const unsigned char *data;
	size_t size;
	ssize_t ret = 0;

	while (i_stream_read_data(client->msg_part_input, &data, &size, 0) > 0) {
		if ((ret = o_stream_send(client->output, data, size)) < 0)
			break;
		i_stream_skip(client->msg_part_input, ret);

		if (o_stream_get_buffer_used_size(client->output) >= 4096) {
			if ((ret = o_stream_flush(client->output)) < 0)
				break;
			if (ret == 0)
				return 0;
		}
	}

	if (client->output->closed || ret < 0) {
		imap_msgpart_url_free(&client->url);
		return -1;
	}

	if (client->msg_part_input->eof) {
		o_stream_send(client->output, "\n", 1);
		imap_msgpart_url_free(&client->url);
		return 1;
	}
	return 0;
}

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

	if ( input != NULL ) {
		while ( (ret=i_stream_read_data(input, &data, &size, 0)) > 0 ) {
			if ( output != NULL ) {
				ssize_t sent;

				if ( (sent=o_stream_send(output, data, size)) < 0 ) {
					program_client_fail(pclient, PROGRAM_CLIENT_ERROR_IO);
					return;
				}
				size = (size_t)sent;
			}

			i_stream_skip(input, size);
		}

		if ( ret < 0 ) {
			if ( i_stream_is_eof(input) ) {
				if ( !program_client_input_pending(pclient) )
					program_client_disconnect(pclient, FALSE);
				return;
			}
			program_client_fail(pclient, PROGRAM_CLIENT_ERROR_IO);
		}
	}
}

static int
sieve_attribute_set_default(struct mail_storage *storage,
			    struct sieve_storage *svstorage,
			    const struct mail_attribute_value *value)
{
	const unsigned char *data;
	size_t size;
	ssize_t ret;
	char type;

	if (value->value != NULL) {
		type = value->value[0];
	} else if (value->value_stream != NULL) {
		ret = i_stream_read_data(value->value_stream, &data, &size, 0);
		if (ret == -1) {
			mail_storage_set_critical(storage, "read(%s) failed: %m",
				i_stream_get_name(value->value_stream));
			return -1;
		}
		i_assert(ret > 0);
		type = data[0];
	} else {
		type = MAILBOX_ATTRIBUTE_SIEVE_DEFAULT_SCRIPT;
	}
	if (type == MAILBOX_ATTRIBUTE_SIEVE_DEFAULT_LINK)
		return sieve_attribute_set_active(storage, svstorage, value);
	if (type == MAILBOX_ATTRIBUTE_SIEVE_DEFAULT_SCRIPT)
		return sieve_attribute_set_active_script(storage, svstorage, value);
	mail_storage_set_error(storage, MAIL_ERROR_PARAMS,
			       "Invalid value for default sieve attribute");
	return -1;
}

int client_auth_read_line(struct client *client)
{
	const unsigned char *data;
	size_t i, size;
	unsigned int len;

	if (i_stream_read_data(client->input, &data, &size, 0) == -1) {
		client_destroy(client, "Disconnected");
		return -1;
	}

	/* see if we have a full line */
	for (i = 0; i < size; i++) {
		if (data[i] == '\n')
			break;
	}
	if (client->auth_response == NULL)
		client->auth_response = str_new(default_pool, I_MAX(i+1, 256));
	if (str_len(client->auth_response) + i > LOGIN_MAX_AUTH_BUF_SIZE) {
		client_destroy(client, "Authentication response too large");
		return -1;
	}
	str_append_n(client->auth_response, data, i);
	i_stream_skip(client->input, i == size ? size : i+1);

	/* drop trailing \r */
	len = str_len(client->auth_response);
	if (len > 0 && str_c(client->auth_response)[len-1] == '\r')
		str_truncate(client->auth_response, len-1);

	return i < size;
}

static void script_client_script_input(struct script_client *sclient)
{
	struct istream *input = sclient->script_input;
	struct ostream *output = sclient->output;
	const unsigned char *data;
	size_t size;
	int ret = 0;

	if ( input != NULL ) {
		while ( (ret=i_stream_read_data(input, &data, &size, 0)) > 0 ) {
			if ( output != NULL ) {
				ssize_t sent;

				if ( (sent=o_stream_send(output, data, size)) < 0 ) {
					script_client_fail(sclient, SCRIPT_CLIENT_ERROR_IO);
					return;
				}
				size = (size_t)sent;
			}

			i_stream_skip(input, size);
		}

		if ( ret < 0 ) {
			if ( input->eof ) {
				script_client_disconnect(sclient, FALSE);
				return;
			}
			script_client_fail(sclient, SCRIPT_CLIENT_ERROR_IO);
		}
	}
}

static void proxy_client_input(struct login_proxy *proxy)
{
	const unsigned char *data;
	size_t size;
	ssize_t ret;

	proxy->last_io = ioloop_time;
	if (o_stream_get_buffer_used_size(proxy->server_output) >
	    OUTBUF_THRESHOLD) {
		/* proxy's output buffer is already quite full.
		   don't send more until we're below threshold. */
		io_remove(&proxy->client_io);
		return;
	}

	if (i_stream_read_data(proxy->client_input, &data, &size, 0) < 0) {
		const char *errstr = i_stream_get_error(proxy->client_input);
		login_proxy_free_errstr(&proxy, errstr, FALSE);
		return;
	}
	o_stream_cork(proxy->server_output);
	ret = o_stream_send(proxy->server_output, data, size);
	o_stream_uncork(proxy->server_output);
	if (ret != (ssize_t)size)
		login_proxy_free_ostream(&proxy, proxy->server_output, TRUE);
	else
		i_stream_skip(proxy->client_input, ret);
}

static int test_input_file(const char *path)
{
	struct istream *file_input, *input, *input2;
	const unsigned char *data;
	size_t size;
	struct sha1_ctxt hash;
	unsigned char hash_file[SHA1_RESULTLEN], hash_converter[SHA1_RESULTLEN];
	int ret = 0;

	lib_init();

	file_input = i_stream_create_file(path, 64);
	
	/* get hash when directly reading input */
	input = i_stream_create_crlf(file_input);
	sha1_init(&hash);
	while (i_stream_read_data(input, &data, &size, 0) > 0) {
		sha1_loop(&hash, data, size);
		i_stream_skip(input, size);
	}
	sha1_result(&hash, hash_file);
	i_stream_unref(&input);

	/* get hash when going through converter */
	i_stream_seek(file_input, 0);
	input = i_stream_create_crlf(file_input);
	input2 = i_stream_create_binary_converter(input);
	sha1_init(&hash);
	while (i_stream_read_data(input2, &data, &size, 0) > 0) {
		sha1_loop(&hash, data, size);
		i_stream_skip(input2, size);
	}
	sha1_result(&hash, hash_converter);
	i_stream_unref(&input2);
	i_stream_unref(&input);

	if (memcmp(hash_file, hash_converter, SHA1_RESULTLEN) != 0) {
		fprintf(stderr, "istream-binary-converter: mismatch on file %s\n",
			path);
		ret = 1;
	}

	i_stream_unref(&file_input);
	lib_deinit();
	return ret;
}

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

int message_get_header_size(struct istream *input, struct message_size *hdr,
			    bool *has_nuls_r)
{
	const unsigned char *msg;
	size_t i, size, startpos, missing_cr_count;
	int ret;

	memset(hdr, 0, sizeof(struct message_size));
	*has_nuls_r = FALSE;

	missing_cr_count = 0; startpos = 0;
	while (i_stream_read_data(input, &msg, &size, startpos) > 0) {
		for (i = startpos; i < size; i++) {
			if (msg[i] != '\n') {
				if (msg[i] == '\0')
					*has_nuls_r = TRUE;
				continue;
			}

			hdr->lines++;
			if (i == 0 || msg[i-1] != '\r') {
				/* missing CR */
				missing_cr_count++;
			}

			if (i == 0 || (i == 1 && msg[i-1] == '\r')) {
				/* no headers at all */
				break;
			}

			if ((i > 0 && msg[i-1] == '\n') ||
			    (i > 1 && msg[i-2] == '\n' && msg[i-1] == '\r')) {
				/* \n\n or \n\r\n - end of headers */
				break;
			}
		}

		if (i < size) {
			/* end of header */
			startpos = i+1;
			break;
		}

		/* leave the last two characters, they may be \r\n */
		startpos = size == 1 ? 1 : 2;
		i_stream_skip(input, i - startpos);

		hdr->physical_size += i - startpos;
	}
	ret = input->stream_errno != 0 ? -1 : 0;
	i_stream_skip(input, startpos);
	hdr->physical_size += startpos;

	hdr->virtual_size = hdr->physical_size + missing_cr_count;
	i_assert(hdr->virtual_size >= hdr->physical_size);
	return ret;
}

static bool is_compressed_xz(struct istream *input)
{
	const unsigned char *data;
	size_t size;

	if (i_stream_read_data(input, &data, &size, 6 - 1) <= 0)
		return FALSE;
	return memcmp(data, "\xfd\x37\x7a\x58\x5a", 6) == 0;
}

static bool is_compressed_lz4(struct istream *input)
{
	const unsigned char *data;
	size_t size;

	if (i_stream_read_data(input, &data, &size, 6 - 1) <= 0)
		return FALSE;
	/* there is no standard LZ4 header, so we've created our own */
	return memcmp(data, IOSTREAM_LZ4_MAGIC, IOSTREAM_LZ4_MAGIC_LEN) == 0;
}