static int copy_to_temp_file(struct seekable_istream *sstream)
{
	struct istream_private *stream = &sstream->istream;
	const char *path;
	const unsigned char *buffer;
	size_t size;
	int fd;

	fd = sstream->fd_callback(&path, sstream->context);
	if (fd == -1)
		return -1;

	/* copy our currently read buffer to it */
	i_assert(stream->pos <= sstream->buffer_peak);
	if (write_full(fd, stream->buffer, sstream->buffer_peak) < 0) {
		if (!ENOSPACE(errno))
			i_error("istream-seekable: write_full(%s) failed: %m", path);
		i_close_fd(&fd);
		return -1;
	}
	sstream->temp_path = i_strdup(path);
	sstream->write_peak = sstream->buffer_peak;

	sstream->fd = fd;
	sstream->fd_input = i_stream_create_fd_autoclose(&fd,
		I_MAX(stream->pos, sstream->istream.max_buffer_size));
	i_stream_set_name(sstream->fd_input, t_strdup_printf(
		"(seekable temp-istream for: %s)", i_stream_get_name(&stream->istream)));

	/* read back the data we just had in our buffer */
	for (;;) {
		buffer = i_stream_get_data(sstream->fd_input, &size);
		if (size >= stream->pos)
			break;

		ssize_t ret;
		if ((ret = i_stream_read_memarea(sstream->fd_input)) <= 0) {
			i_assert(ret != 0);
			i_assert(ret != -2);
			i_error("istream-seekable: Couldn't read back "
				"in-memory input %s: %s",
				i_stream_get_name(&stream->istream),
				i_stream_get_error(sstream->fd_input));
			i_stream_destroy(&sstream->fd_input);
			i_close_fd(&sstream->fd);
			return -1;
		}
	}
	/* Set the max buffer size only after we've already read everything
	   into memory. For example with istream-data it's possible that
	   more data exists in buffer than max_buffer_size. */
	i_stream_set_max_buffer_size(sstream->fd_input,
				     sstream->istream.max_buffer_size);
	stream->buffer = buffer;
	i_stream_free_buffer(&sstream->istream);
	return 0;
}

int istream_attachment_connector_add(struct istream_attachment_connector *conn,
				     struct istream *decoded_input,
				     uoff_t start_offset, uoff_t encoded_size,
				     unsigned int base64_blocks_per_line,
				     bool base64_have_crlf,
				     const char **error_r)
{
	struct istream *input, *input2;
	uoff_t base_prefix_size;

	if (start_offset < conn->encoded_offset) {
		*error_r = t_strdup_printf(
			"Attachment %s points before the previous attachment "
			"(%"PRIuUOFF_T" < %"PRIuUOFF_T")",
			i_stream_get_name(decoded_input),
			start_offset, conn->encoded_offset);
		return -1;
	}
	base_prefix_size = start_offset - conn->encoded_offset;
	if (start_offset + encoded_size > conn->msg_size) {
		*error_r = t_strdup_printf(
			"Attachment %s points outside message "
			"(%"PRIuUOFF_T" + %"PRIuUOFF_T" > %"PRIuUOFF_T")",
			i_stream_get_name(decoded_input),
			start_offset, encoded_size,
			conn->msg_size);
		return -1;
	}

	if (base_prefix_size > 0) {
		/* add a part of the base message before the attachment */
		input = i_stream_create_range(conn->base_input,
					      conn->base_input_offset,
					      base_prefix_size);
		array_append(&conn->streams, &input, 1);
		conn->base_input_offset += base_prefix_size;
		conn->encoded_offset += base_prefix_size;
	}
	conn->encoded_offset += encoded_size;

	if (base64_blocks_per_line == 0) {
		input = decoded_input;
		i_stream_ref(input);
	} else {
		input = i_stream_create_base64_encoder(decoded_input,
						       base64_blocks_per_line*4,
						       base64_have_crlf);
	}
	input2 = i_stream_create_sized(input, encoded_size);
	array_append(&conn->streams, &input2, 1);
	i_stream_unref(&input);
	return 0;
}

static int copy_to_temp_file(struct seekable_istream *sstream)
{
    struct istream_private *stream = &sstream->istream;
    const char *path;
    const unsigned char *buffer;
    size_t size;
    int fd;

    fd = sstream->fd_callback(&path, sstream->context);
    if (fd == -1)
        return -1;

    /* copy our currently read buffer to it */
    if (write_full(fd, sstream->membuf->data, sstream->membuf->used) < 0) {
        if (!ENOSPACE(errno))
            i_error("istream-seekable: write_full(%s) failed: %m", path);
        i_close_fd(&fd);
        return -1;
    }
    sstream->temp_path = i_strdup(path);
    sstream->write_peak = sstream->membuf->used;

    sstream->fd = fd;
    sstream->fd_input =
        i_stream_create_fd_autoclose(&fd, sstream->istream.max_buffer_size);
    i_stream_set_name(sstream->fd_input, t_strdup_printf(
                          "(seekable temp-istream for: %s)", i_stream_get_name(&stream->istream)));

    /* read back the data we just had in our buffer */
    i_stream_seek(sstream->fd_input, stream->istream.v_offset);
    for (;;) {
        buffer = i_stream_get_data(sstream->fd_input, &size);
        if (size >= stream->pos)
            break;

        if (i_stream_read(sstream->fd_input) <= 0) {
            i_error("istream-seekable: Couldn't read back "
                    "in-memory input %s",
                    i_stream_get_name(&stream->istream));
            i_stream_destroy(&sstream->fd_input);
            return -1;
        }
    }
    stream->buffer = buffer;
    stream->pos = size;
    buffer_free(&sstream->membuf);
    return 0;
}

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

int sieve_file_storage_save_continue
(struct sieve_storage_save_context *sctx)
{
	struct sieve_file_save_context *fsctx =
		(struct sieve_file_save_context *)sctx;

	switch (o_stream_send_istream(fsctx->output, sctx->input)) {
	case OSTREAM_SEND_ISTREAM_RESULT_FINISHED:
	case OSTREAM_SEND_ISTREAM_RESULT_WAIT_INPUT:
		return 0;
	case OSTREAM_SEND_ISTREAM_RESULT_WAIT_OUTPUT:
		i_unreached();
	case OSTREAM_SEND_ISTREAM_RESULT_ERROR_INPUT:
		sieve_storage_set_critical(sctx->storage,
			"save: read(%s) failed: %s",
			i_stream_get_name(sctx->input),
			i_stream_get_error(sctx->input));
		return -1;
	case OSTREAM_SEND_ISTREAM_RESULT_ERROR_OUTPUT:
		sieve_storage_set_critical(sctx->storage,
			"save: write(%s) failed: %s", fsctx->tmp_path,
			o_stream_get_error(fsctx->output));
		return -1;
	}
	return 0;
}

static int server_connection_send_cmd_input_more(struct server_connection *conn)
{
    off_t ret;

    /* ostream-dot writes only up to max buffer size, so keep it non-zero */
    o_stream_set_max_buffer_size(conn->cmd_output, IO_BLOCK_SIZE);
    ret = o_stream_send_istream(conn->cmd_output, conn->cmd_input);
    o_stream_set_max_buffer_size(conn->cmd_output, (size_t)-1);

    if (ret >= 0 && i_stream_have_bytes_left(conn->cmd_input)) {
        o_stream_set_flush_pending(conn->cmd_output, TRUE);
        return 0;
    }
    if (conn->cmd_input->stream_errno != 0) {
        i_error("read(%s) failed: %s",
                i_stream_get_name(conn->cmd_input),
                i_stream_get_error(conn->cmd_input));
    } else if (conn->cmd_output->stream_errno != 0 ||
               o_stream_flush(conn->cmd_output) < 0) {
        i_error("write(%s) failed: %s",
                o_stream_get_name(conn->cmd_output),
                o_stream_get_error(conn->cmd_output));
    }

    i_stream_destroy(&conn->cmd_input);
    o_stream_destroy(&conn->cmd_output);
    return ret < 0 ? -1 : 1;
}

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

static void
i_stream_mail_set_size_corrupted(struct mail_istream *mstream, size_t size)
{
	uoff_t cur_size = mstream->istream.istream.v_offset + size;
	const char *str, *mail_id;
	char chr;

	if (mstream->expected_size < cur_size) {
		str = "smaller";
		chr = '<';
	} else {
		str = "larger";
		chr = '>';
	}

	mail_id = i_stream_mail_get_cached_mail_id(mstream);
	if (mail_id[0] != '\0')
		mail_id = t_strconcat(", cached ", mail_id, NULL);
	io_stream_set_error(&mstream->istream.iostream,
		"Cached message size %s than expected "
		"(%"PRIuUOFF_T" %c %"PRIuUOFF_T", box=%s, UID=%u%s)", str,
		mstream->expected_size, chr, cur_size,
		mailbox_get_vname(mstream->mail->box),
		mstream->mail->uid, mail_id);
	mail_set_cache_corrupted_reason(mstream->mail, MAIL_FETCH_PHYSICAL_SIZE,
		t_strdup_printf("read(%s) failed: %s",
				i_stream_get_name(&mstream->istream.istream),
				mstream->istream.iostream.error));
	mstream->istream.istream.stream_errno = EINVAL;
}

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

int json_parser_deinit(struct json_parser **_parser, const char **error_r)
{
	struct json_parser *parser = *_parser;

	*_parser = NULL;

	if (parser->error != NULL) {
		/* actual parser error */
		*error_r = parser->error;
	} else if (parser->input->stream_errno != 0) {
		*error_r = t_strdup_printf("read(%s) failed: %m",
					   i_stream_get_name(parser->input));
	} else if (parser->data == parser->end &&
		   !i_stream_have_bytes_left(parser->input) &&
		   parser->state != JSON_STATE_DONE) {
		*error_r = "Missing '}'";
	} else {
		*error_r = NULL;
	}
	
	i_stream_unref(&parser->input);
	array_free(&parser->nesting);
	str_free(&parser->value);
	i_free(parser);
	return *error_r != NULL ? -1 : 0;
}

void http_client_request_set_payload(struct http_client_request *req,
				     struct istream *input, bool sync)
{
	int ret;

	i_assert(req->state == HTTP_REQUEST_STATE_NEW);
	i_assert(req->payload_input == NULL);

	i_stream_ref(input);
	req->payload_input = input;
	if ((ret = i_stream_get_size(input, TRUE, &req->payload_size)) <= 0) {
		if (ret < 0) {
			i_error("i_stream_get_size(%s) failed: %s",
				i_stream_get_name(input),
				i_stream_get_error(input));
		}
		req->payload_size = 0;
		req->payload_chunked = TRUE;
	}
	req->payload_offset = input->v_offset;

	/* prepare request payload sync using 100 Continue response from server */
	if ((req->payload_chunked || req->payload_size > 0) && sync)
		req->payload_sync = TRUE;
}

static int raw_mail_stat(struct mail *mail)
{
	struct raw_mailbox *mbox = (struct raw_mailbox *)mail->box;
	const struct stat *st;

	if (mail->lookup_abort == MAIL_LOOKUP_ABORT_NOT_IN_CACHE) {
		mail_set_aborted(mail);
		return -1;
	}

	mail->transaction->stats.fstat_lookup_count++;
	if (i_stream_stat(mail->box->input, TRUE, &st) < 0) {
		mail_storage_set_critical(mail->box->storage,
					  "stat(%s) failed: %m",
					  i_stream_get_name(mail->box->input));
		return -1;
	}

	if (mbox->mtime != (time_t)-1)
		mbox->mtime = st->st_mtime;
	if (mbox->ctime != (time_t)-1)
		mbox->ctime = st->st_ctime;
	mbox->size = st->st_size;
	return 0;
}

static void bzlib_read_error(struct bzlib_istream *zstream, const char *error)
{
	i_error("bzlib.read(%s): %s at %"PRIuUOFF_T,
		i_stream_get_name(&zstream->istream.istream), error,
		zstream->istream.abs_start_offset +
		zstream->istream.istream.v_offset);
}

raw_mailbox_alloc_common(struct mail_user *user, struct istream *input,
			 const char *path, time_t received_time,
			 const char *envelope_sender, struct mailbox **box_r)
{
	struct mail_namespace *ns = user->namespaces;
	struct mailbox *box;
	struct raw_mailbox *raw_box;
	const char *name;

	name = path != NULL ? path : i_stream_get_name(input);
	box = *box_r = mailbox_alloc(ns->list, name,
				     MAILBOX_FLAG_NO_INDEX_FILES);
	if (input != NULL) {
		if (mailbox_open_stream(box, input) < 0)
			return -1;
	} else {
		if (mailbox_open(box) < 0)
			return -1;
	}
	if (mailbox_sync(box, 0) < 0)
		return -1;

	i_assert(strcmp(box->storage->name, RAW_STORAGE_NAME) == 0);
	raw_box = (struct raw_mailbox *)box;
	raw_box->envelope_sender = envelope_sender;
	raw_box->mtime = received_time;
	return 0;
}

struct istream *tee_i_stream_create_child(struct tee_istream *tee)
{
	struct tee_child_istream *tstream;
	struct istream *ret, *input = tee->input;

	tstream = i_new(struct tee_child_istream, 1);
	tstream->tee = tee;

	tstream->istream.max_buffer_size = input->real_stream->max_buffer_size;
	tstream->istream.iostream.close = i_stream_tee_close;
	tstream->istream.iostream.destroy = i_stream_tee_destroy;
	tstream->istream.iostream.set_max_buffer_size =
		i_stream_tee_set_max_buffer_size;

	tstream->istream.read = i_stream_tee_read;
	tstream->istream.stat = i_stream_tee_stat;
	tstream->istream.sync = i_stream_tee_sync;

	tstream->next = tee->children;
	tee->children = tstream;

	ret = i_stream_create(&tstream->istream, input, i_stream_get_fd(input));
	i_stream_set_name(&tstream->istream.istream, i_stream_get_name(input));
	/* we keep the reference in tee stream, no need for extra references */
	i_stream_unref(&input);
	return ret;
}

static int
i_stream_file_stat(struct istream_private *stream, bool exact ATTR_UNUSED)
{
	struct file_istream *fstream = (struct file_istream *) stream;
	const char *name = i_stream_get_name(&stream->istream);

	if (!fstream->file) {
		/* return defaults */
	} else if (stream->fd != -1) {
		if (fstat(stream->fd, &stream->statbuf) < 0) {
			stream->istream.stream_errno = errno;
			io_stream_set_error(&stream->iostream,
				"file_istream.fstat(%s) failed: %m", name);
			i_error("%s", i_stream_get_error(&stream->istream));
			return -1;
		}
	} else {
		if (stat(name, &stream->statbuf) < 0) {
			stream->istream.stream_errno = errno;
			io_stream_set_error(&stream->iostream,
				"file_istream.stat(%s) failed: %m", name);
			i_error("%s", i_stream_get_error(&stream->istream));
			return -1;
		}
	}
	return 0;
}

static void fetch_read_error(struct imap_fetch_context *ctx)
{
	errno = ctx->cur_input->stream_errno;
	mail_storage_set_critical(ctx->box->storage,
		"read(%s) failed: %m (FETCH for mailbox %s UID %u)",
		i_stream_get_name(ctx->cur_input),
		mailbox_get_vname(ctx->mail->box), ctx->mail->uid);
}

static void
i_stream_zlib_seek(struct istream_private *stream, uoff_t v_offset, bool mark)
{
	struct zlib_istream *zstream = (struct zlib_istream *) stream;
	uoff_t start_offset = stream->istream.v_offset - stream->skip;

	if (v_offset < start_offset) {
		/* have to seek backwards */
		i_stream_zlib_reset(zstream);
		start_offset = 0;
	} else if (zstream->high_pos != 0) {
		stream->pos = zstream->high_pos;
		zstream->high_pos = 0;
	}

	if (v_offset <= start_offset + stream->pos) {
		/* seeking backwards within what's already cached */
		stream->skip = v_offset - start_offset;
		stream->istream.v_offset = v_offset;
		zstream->high_pos = stream->pos;
		stream->pos = stream->skip;
	} else {
		/* read and cache forward */
		ssize_t ret;

		do {
			size_t avail = stream->pos - stream->skip;

			if (stream->istream.v_offset + avail >= v_offset) {
				i_stream_skip(&stream->istream,
					      v_offset -
					      stream->istream.v_offset);
				ret = -1;
				break;
			}

			i_stream_skip(&stream->istream, avail);
		} while ((ret = i_stream_read(&stream->istream)) > 0);
		i_assert(ret == -1);

		if (stream->istream.v_offset != v_offset) {
			/* some failure, we've broken it */
			if (stream->istream.stream_errno != 0) {
				i_error("zlib_istream.seek(%s) failed: %s",
					i_stream_get_name(&stream->istream),
					strerror(stream->istream.stream_errno));
				i_stream_close(&stream->istream);
			} else {
				/* unexpected EOF. allow it since we may just
				   want to check if there's anything.. */
				i_assert(stream->istream.eof);
			}
		}
	}

	if (mark)
		zstream->marked = TRUE;
}

static int
cmd_dsync_server_run(struct doveadm_mail_cmd_context *_ctx,
		     struct mail_user *user)
{
	struct dsync_cmd_context *ctx = (struct dsync_cmd_context *)_ctx;
	struct dsync_ibc *ibc;
	struct dsync_brain *brain;
	string_t *temp_prefix, *state_str = NULL;
	enum dsync_brain_sync_type sync_type;
	const char *name;

	if (_ctx->conn != NULL) {
		/* doveadm-server connection. start with a success reply.
		   after that follows the regular dsync protocol. */
		ctx->fd_in = ctx->fd_out = -1;
		ctx->input = _ctx->conn->input;
		ctx->output = _ctx->conn->output;
		o_stream_nsend(ctx->output, "\n+\n", 3);
		i_set_failure_prefix("dsync-server(%s): ", user->username);
		name = i_stream_get_name(ctx->input);
	} else {
		/* the log messages go via stderr to the remote dsync,
		   so the names are reversed */
		i_set_failure_prefix("dsync-remote(%s): ", user->username);
		name = "local";
	}

	doveadm_user_init_dsync(user);

	temp_prefix = t_str_new(64);
	mail_user_set_get_temp_prefix(temp_prefix, user->set);

	ibc = cmd_dsync_icb_stream_init(ctx, name, str_c(temp_prefix));
	brain = dsync_brain_slave_init(user, ibc, FALSE);

	io_loop_run(current_ioloop);

	if (ctx->replicator_notify) {
		state_str = t_str_new(128);
		dsync_brain_get_state(brain, state_str);
	}
	sync_type = dsync_brain_get_sync_type(brain);

	if (dsync_brain_deinit(&brain) < 0)
		_ctx->exit_code = EX_TEMPFAIL;
	dsync_ibc_deinit(&ibc);

	if (_ctx->conn != NULL) {
		/* make sure nothing more is written by the generic doveadm
		   connection code */
		o_stream_close(_ctx->conn->output);
	}

	if (ctx->replicator_notify && _ctx->exit_code == 0)
		dsync_replicator_notify(ctx, sync_type, str_c(state_str));
	return _ctx->exit_code == 0 ? 0 : -1;
}

static void lzma_read_error(struct lzma_istream *zstream, const char *error)
{
	io_stream_set_error(&zstream->istream.iostream,
			    "lzma.read(%s): %s at %"PRIuUOFF_T,
			    i_stream_get_name(&zstream->istream.istream), error,
			    i_stream_get_absolute_offset(&zstream->istream.istream));
	if (zstream->log_errors)
		i_error("%s", zstream->istream.iostream.error);
}