src/lib/ostream.c

	ostream.c revision 46631c1d903c409444b1b1c4a1d41a033c09ee37
5e0ce63bb65db34d7f48b34bbb5545fa791781c4Timo Sirainen/* Copyright (c) 2002-2014 Dovecot authors, see the included COPYING file */
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen#include "lib.h"
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen#include "istream.h"
5dd05e966ffd69181ab3067f6939b03ced68ebc3Timo Sirainen#include "ostream-private.h"
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenvoid o_stream_set_name(struct ostream *stream, const char *name)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    i_free(stream->real_stream->iostream.name);
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen    stream->real_stream->iostream.name = i_strdup(name);
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
4b058f90f9e8a2c6b2eed275de4eb8cc5195a71dTimo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenconst char *o_stream_get_name(struct ostream *stream)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen    while (stream->real_stream->iostream.name == NULL) {
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen        stream = stream->real_stream->parent;
cd466fe7b84b0223735a6469c7f7bc225f65996dTimo Sirainen        if (stream == NULL)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen            return "";
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    }
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen    return stream->real_stream->iostream.name;
d56384d5226c8860079d0d0b08b83404e8c42986Timo Sirainen}
0f66f12eb4cdbf47670975044c88d8f388bf92dfTimo Sirainen
d56384d5226c8860079d0d0b08b83404e8c42986Timo Sirainenint o_stream_get_fd(struct ostream *stream)
d56384d5226c8860079d0d0b08b83404e8c42986Timo Sirainen{
4b058f90f9e8a2c6b2eed275de4eb8cc5195a71dTimo Sirainen    return stream->real_stream->fd;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenconst char *o_stream_get_error(struct ostream *stream)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
4b058f90f9e8a2c6b2eed275de4eb8cc5195a71dTimo Sirainen    struct ostream *s;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    /* we'll only return errors for streams that have stream_errno set.
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen       we might be returning unintended error otherwise. */
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    if (stream->stream_errno == 0)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen        return "<no error>";
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen
6a9f9a5101b665fd2ef80c9e048a5eace78e01efTimo Sirainen    for (s = stream; s != NULL; s = s->real_stream->parent) {
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen        if (s->stream_errno == 0)
0f66f12eb4cdbf47670975044c88d8f388bf92dfTimo Sirainen            break;
6a9f9a5101b665fd2ef80c9e048a5eace78e01efTimo Sirainen        if (s->real_stream->iostream.error != NULL)
6a9f9a5101b665fd2ef80c9e048a5eace78e01efTimo Sirainen            return s->real_stream->iostream.error;
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen    }
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    return strerror(stream->stream_errno);
4b058f90f9e8a2c6b2eed275de4eb8cc5195a71dTimo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenstatic void o_stream_close_full(struct ostream *stream, bool close_parents)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen    if (!stream->closed && !stream->real_stream->closing) {
a3ea111cfdbfd4f32baeb0bd7f1d72568c60a023Timo Sirainen        /* first mark the stream as being closed so the
a3ea111cfdbfd4f32baeb0bd7f1d72568c60a023Timo Sirainen           o_stream_copy_error_from_parent() won't recurse us back
a3ea111cfdbfd4f32baeb0bd7f1d72568c60a023Timo Sirainen           here. but don't immediately mark the stream closed, because
a3ea111cfdbfd4f32baeb0bd7f1d72568c60a023Timo Sirainen           we may still want to write something to it. */
8d80659e504ffb34bb0c6a633184fece35751b18Timo Sirainen        stream->real_stream->closing = TRUE;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen        io_stream_close(&stream->real_stream->iostream, close_parents);
4b058f90f9e8a2c6b2eed275de4eb8cc5195a71dTimo Sirainen        stream->closed = TRUE;
8d80659e504ffb34bb0c6a633184fece35751b18Timo Sirainen    }
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    if (stream->stream_errno == 0)
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen        stream->stream_errno = EPIPE;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenvoid o_stream_destroy(struct ostream **stream)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen    o_stream_close_full(*stream, FALSE);
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    o_stream_unref(stream);
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenvoid o_stream_ref(struct ostream *stream)
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen{
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    io_stream_ref(&stream->real_stream->iostream);
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenvoid o_stream_unref(struct ostream **_stream)
a3ea111cfdbfd4f32baeb0bd7f1d72568c60a023Timo Sirainen{
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    struct ostream *stream = *_stream;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    if (stream->real_stream->last_errors_not_checked &&
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen        !stream->real_stream->error_handling_disabled &&
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen        stream->real_stream->iostream.refcount == 1) {
a29a5b7520f7b8d6cdaf97e66d184b6a9e4f4ecfTimo Sirainen        i_panic("output stream %s is missing error handling",
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen            o_stream_get_name(stream));
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    }
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    io_stream_unref(&stream->real_stream->iostream);
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen    *_stream = NULL;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainenvoid o_stream_close(struct ostream *stream)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    o_stream_close_full(stream, TRUE);
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen#undef o_stream_set_flush_callback
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainenvoid o_stream_set_flush_callback(struct ostream *stream,
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen                 stream_flush_callback_t *callback,
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen                 void *context)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    struct ostream_private *_stream = stream->real_stream;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    _stream->set_flush_callback(_stream, callback, context);
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainenvoid o_stream_unset_flush_callback(struct ostream *stream)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    struct ostream_private *_stream = stream->real_stream;
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    _stream->set_flush_callback(_stream, NULL, NULL);
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen}
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainenvoid o_stream_set_max_buffer_size(struct ostream *stream, size_t max_size)
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen{
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    io_stream_set_max_buffer_size(&stream->real_stream->iostream, max_size);
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen}
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainensize_t o_stream_get_max_buffer_size(struct ostream *stream)
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen{
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    return stream->real_stream->max_buffer_size;
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen}
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainenvoid o_stream_cork(struct ostream *stream)
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen{
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    struct ostream_private *_stream = stream->real_stream;
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    if (unlikely(stream->closed))
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen        return;
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    _stream->cork(_stream, TRUE);
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen}
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainenvoid o_stream_uncork(struct ostream *stream)
5dd05e966ffd69181ab3067f6939b03ced68ebc3Timo Sirainen{
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    struct ostream_private *_stream = stream->real_stream;
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen
b09be485e9373be4288f5615bbce6ebed65a425aTimo Sirainen    if (unlikely(stream->closed))
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen        return;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
9453e8d75cfd8fab2232cf772e9b120f308fb3eeTimo Sirainen    _stream->cork(_stream, FALSE);
9453e8d75cfd8fab2232cf772e9b120f308fb3eeTimo Sirainen}
9453e8d75cfd8fab2232cf772e9b120f308fb3eeTimo Sirainen
9453e8d75cfd8fab2232cf772e9b120f308fb3eeTimo Sirainenbool o_stream_is_corked(struct ostream *stream)
9453e8d75cfd8fab2232cf772e9b120f308fb3eeTimo Sirainen{
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen    struct ostream_private *_stream = stream->real_stream;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    return _stream->corked;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen}
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainenstatic void o_stream_clear_error(struct ostream *stream)
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen{
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    stream->stream_errno = 0;
33ca6b017b6ebbd048651b5e3d16915001dbc291Timo Sirainen    i_free_and_null(stream->real_stream->iostream.error);
}

int o_stream_flush(struct ostream *stream)
{
    struct ostream_private *_stream = stream->real_stream;
    int ret = 1;

    if (unlikely(stream->closed)) {
        errno = stream->stream_errno;
        return -1;
    }

    o_stream_clear_error(stream);
    if (unlikely((ret = _stream->flush(_stream)) < 0)) {
        i_assert(stream->stream_errno != 0);
        stream->last_failed_errno = stream->stream_errno;
        errno = stream->stream_errno;
    }
    return ret;
}

void o_stream_set_flush_pending(struct ostream *stream, bool set)
{
    struct ostream_private *_stream = stream->real_stream;

    if (unlikely(stream->closed))
        return;

    _stream->flush_pending(_stream, set);
}

size_t o_stream_get_buffer_used_size(const struct ostream *stream)
{
    const struct ostream_private *_stream = stream->real_stream;

    return _stream->get_used_size(_stream);
}

size_t o_stream_get_buffer_avail_size(const struct ostream *stream)
{
    size_t used = o_stream_get_buffer_used_size(stream);

    return stream->real_stream->max_buffer_size <= used ? 0 :
        stream->real_stream->max_buffer_size - used;
}

int o_stream_seek(struct ostream *stream, uoff_t offset)
{
    struct ostream_private *_stream = stream->real_stream;

    if (unlikely(stream->closed)) {
        errno = stream->stream_errno;
        return -1;
    }

    o_stream_clear_error(stream);
    if (unlikely(_stream->seek(_stream, offset) < 0)) {
        i_assert(stream->stream_errno != 0);
        stream->last_failed_errno = stream->stream_errno;
        errno = stream->stream_errno;
        return -1;
    }
    return 1;
}

ssize_t o_stream_send(struct ostream *stream, const void *data, size_t size)
{
    struct const_iovec iov;

    memset(&iov, 0, sizeof(iov));
    iov.iov_base = data;
    iov.iov_len = size;

    return o_stream_sendv(stream, &iov, 1);
}

ssize_t o_stream_sendv(struct ostream *stream, const struct const_iovec *iov,
               unsigned int iov_count)
{
    struct ostream_private *_stream = stream->real_stream;
    unsigned int i;
    size_t total_size;
    ssize_t ret;

    if (unlikely(stream->closed)) {
        errno = stream->stream_errno;
        return -1;
    }

    o_stream_clear_error(stream);
    for (i = 0, total_size = 0; i < iov_count; i++)
        total_size += iov[i].iov_len;
    if (total_size == 0)
        return 0;

    ret = _stream->sendv(_stream, iov, iov_count);
    if (unlikely(ret != (ssize_t)total_size)) {
        if (ret < 0) {
            i_assert(stream->stream_errno != 0);
            stream->last_failed_errno = stream->stream_errno;
            errno = stream->stream_errno;
        } else {
            stream->overflow = TRUE;
        }
    }
    return ret;
}

ssize_t o_stream_send_str(struct ostream *stream, const char *str)
{
    return o_stream_send(stream, str, strlen(str));
}

void o_stream_nsend(struct ostream *stream, const void *data, size_t size)
{
    struct const_iovec iov;

    memset(&iov, 0, sizeof(iov));
    iov.iov_base = data;
    iov.iov_len = size;

    o_stream_nsendv(stream, &iov, 1);
}

void o_stream_nsendv(struct ostream *stream, const struct const_iovec *iov,
             unsigned int iov_count)
{
    if (unlikely(stream->closed))
        return;
    (void)o_stream_sendv(stream, iov, iov_count);
    stream->real_stream->last_errors_not_checked = TRUE;
}

void o_stream_nsend_str(struct ostream *stream, const char *str)
{
    o_stream_nsend(stream, str, strlen(str));
}

void o_stream_nflush(struct ostream *stream)
{
    if (unlikely(stream->closed))
        return;
    (void)o_stream_flush(stream);
    stream->real_stream->last_errors_not_checked = TRUE;
}

int o_stream_nfinish(struct ostream *stream)
{
    o_stream_nflush(stream);
    o_stream_ignore_last_errors(stream);
    errno = stream->last_failed_errno;
    return stream->last_failed_errno != 0 ? -1 : 0;
}

void o_stream_ignore_last_errors(struct ostream *stream)
{
    while (stream != NULL) {
        stream->real_stream->last_errors_not_checked = FALSE;
        stream = stream->real_stream->parent;
    }
}

void o_stream_set_no_error_handling(struct ostream *stream, bool set)
{
    stream->real_stream->error_handling_disabled = set;
}

off_t o_stream_send_istream(struct ostream *outstream,
                struct istream *instream)
{
    struct ostream_private *_outstream = outstream->real_stream;
    off_t ret;

    if (unlikely(outstream->closed || instream->closed)) {
        errno = outstream->stream_errno;
        return -1;
    }

    o_stream_clear_error(outstream);
    ret = _outstream->send_istream(_outstream, instream);
    if (unlikely(ret < 0)) {
        i_assert(outstream->stream_errno != 0);
        outstream->last_failed_errno = outstream->stream_errno;
        errno = outstream->stream_errno;
    }
    return ret;
}

int o_stream_pwrite(struct ostream *stream, const void *data, size_t size,
            uoff_t offset)
{
    int ret;

    if (unlikely(stream->closed)) {
        errno = stream->stream_errno;
        return -1;
    }

    o_stream_clear_error(stream);
    ret = stream->real_stream->write_at(stream->real_stream,
                        data, size, offset);
    if (unlikely(ret < 0)) {
        i_assert(stream->stream_errno != 0);
        stream->last_failed_errno = stream->stream_errno;
        errno = stream->stream_errno;
    }
    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 */
            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);
}

void o_stream_switch_ioloop(struct ostream *stream)
{
    struct ostream_private *_stream = stream->real_stream;

    _stream->switch_ioloop(_stream);
}

static void o_stream_default_close(struct iostream_private *stream,
                   bool close_parent)
{
    struct ostream_private *_stream = (struct ostream_private *)stream;

    (void)o_stream_flush(&_stream->ostream);
    if (close_parent && _stream->parent != NULL)
        o_stream_close(_stream->parent);
}

static void o_stream_default_destroy(struct iostream_private *stream)
{
    struct ostream_private *_stream = (struct ostream_private *)stream;

    if (_stream->parent != NULL)
        o_stream_unref(&_stream->parent);
}

static void
o_stream_default_set_max_buffer_size(struct iostream_private *stream,
                     size_t max_size)
{
    struct ostream_private *_stream = (struct ostream_private *)stream;

    if (_stream->parent != NULL)
        o_stream_set_max_buffer_size(_stream->parent, max_size);
    _stream->max_buffer_size = max_size;
}

static void o_stream_default_cork(struct ostream_private *_stream, bool set)
{
    _stream->corked = set;
    if (set) {
        if (_stream->parent != NULL)
            o_stream_cork(_stream->parent);
    } else {
        (void)o_stream_flush(&_stream->ostream);
        if (_stream->parent != NULL)
            o_stream_uncork(_stream->parent);
    }
}

void o_stream_copy_error_from_parent(struct ostream_private *_stream)
{
    struct ostream *src = _stream->parent;
    struct ostream *dest = &_stream->ostream;

    dest->stream_errno = src->stream_errno;
    dest->last_failed_errno = src->last_failed_errno;
    dest->overflow = src->overflow;
    if (src->closed)
        o_stream_close(dest);
}

int o_stream_flush_parent_if_needed(struct ostream_private *_stream)
{
    if (o_stream_get_buffer_used_size(_stream->parent) >= IO_BLOCK_SIZE) {
        /* we already have quite a lot of data in parent stream.
           unless we can flush it, don't add any more to it or we
           could keep wasting memory by just increasing the buffer
           size all the time. */
        if (o_stream_flush(_stream->parent) < 0) {
            o_stream_copy_error_from_parent(_stream);
            return -1;
        }
        if (o_stream_get_buffer_used_size(_stream->parent) >= IO_BLOCK_SIZE)
            return 0;
    }
    return 1;
}

static int o_stream_default_flush(struct ostream_private *_stream)
{
    int ret;

    if (_stream->parent == NULL)
        return 1;

    if ((ret = o_stream_flush(_stream->parent)) < 0)
        o_stream_copy_error_from_parent(_stream);
    return ret;
}

static void
o_stream_default_set_flush_callback(struct ostream_private *_stream,
                    stream_flush_callback_t *callback,
                    void *context)
{
    if (_stream->parent != NULL)
        o_stream_set_flush_callback(_stream->parent, callback, context);

    _stream->callback = callback;
    _stream->context = context;
}

static void
o_stream_default_set_flush_pending(struct ostream_private *_stream, bool set)
{
    if (_stream->parent != NULL)
        o_stream_set_flush_pending(_stream->parent, set);
}

static size_t
o_stream_default_get_used_size(const struct ostream_private *_stream)
{
    if (_stream->parent == NULL)
        return 0;
    else
        return o_stream_get_buffer_used_size(_stream->parent);
}

static int
o_stream_default_seek(struct ostream_private *_stream,
              uoff_t offset ATTR_UNUSED)
{
    _stream->ostream.stream_errno = ESPIPE;
    return -1;
}

static int
o_stream_default_write_at(struct ostream_private *_stream,
              const void *data ATTR_UNUSED,
              size_t size ATTR_UNUSED, uoff_t offset ATTR_UNUSED)
{
    _stream->ostream.stream_errno = ESPIPE;
    return -1;
}

static off_t o_stream_default_send_istream(struct ostream_private *outstream,
                       struct istream *instream)
{
    return io_stream_copy(&outstream->ostream, instream);
}

static void o_stream_default_switch_ioloop(struct ostream_private *_stream)
{
    if (_stream->parent != NULL)
        o_stream_switch_ioloop(_stream->parent);
}

struct ostream *
o_stream_create(struct ostream_private *_stream, struct ostream *parent, int fd)
{
    _stream->fd = fd;
    _stream->ostream.real_stream = _stream;
    if (parent != NULL) {
        _stream->parent = parent;
        o_stream_ref(parent);

        _stream->callback = parent->real_stream->callback;
        _stream->context = parent->real_stream->context;
        _stream->max_buffer_size = parent->real_stream->max_buffer_size;
        _stream->error_handling_disabled =
            parent->real_stream->error_handling_disabled;
    }

    if (_stream->iostream.close == NULL)
        _stream->iostream.close = o_stream_default_close;
    if (_stream->iostream.destroy == NULL)
        _stream->iostream.destroy = o_stream_default_destroy;
    if (_stream->iostream.set_max_buffer_size == NULL) {
        _stream->iostream.set_max_buffer_size =
            o_stream_default_set_max_buffer_size;
    }

    if (_stream->cork == NULL)
        _stream->cork = o_stream_default_cork;
    if (_stream->flush == NULL)
        _stream->flush = o_stream_default_flush;
    if (_stream->set_flush_callback == NULL) {
        _stream->set_flush_callback =
            o_stream_default_set_flush_callback;
    }
    if (_stream->flush_pending == NULL)
        _stream->flush_pending = o_stream_default_set_flush_pending;
    if (_stream->get_used_size == NULL)
        _stream->get_used_size = o_stream_default_get_used_size;
    if (_stream->seek == NULL)
        _stream->seek = o_stream_default_seek;
    if (_stream->write_at == NULL)
        _stream->write_at = o_stream_default_write_at;
    if (_stream->send_istream == NULL)
        _stream->send_istream = o_stream_default_send_istream;
    if (_stream->switch_ioloop == NULL)
        _stream->switch_ioloop = o_stream_default_switch_ioloop;

    io_stream_init(&_stream->iostream);
    return &_stream->ostream;
}

struct ostream *o_stream_create_error(int stream_errno)
{
    struct ostream_private *stream;
    struct ostream *output;

    stream = i_new(struct ostream_private, 1);
    stream->ostream.closed = TRUE;
    stream->ostream.stream_errno = stream_errno;
    stream->ostream.last_failed_errno = stream_errno;

    output = o_stream_create(stream, NULL, -1);
    o_stream_set_no_error_handling(output, TRUE);
    o_stream_set_name(output, "(error)");
    return output;
}

struct ostream *
o_stream_create_error_str(int stream_errno, const char *fmt, ...)
{
    struct ostream *output;
    va_list args;

    va_start(args, fmt);
    output = o_stream_create_error(stream_errno);
    io_stream_set_verror(&output->real_stream->iostream, fmt, args);
    va_end(args);
    return output;
}