diff options
Diffstat (limited to '')
| -rw-r--r-- | src/lib/istream-chain.c | 349 |
1 files changed, 349 insertions, 0 deletions
diff --git a/src/lib/istream-chain.c b/src/lib/istream-chain.c new file mode 100644 index 0000000..3b9a87b --- /dev/null +++ b/src/lib/istream-chain.c @@ -0,0 +1,349 @@ +/* Copyright (c) 2003-2018 Dovecot authors, see the included COPYING file */ + +#include "lib.h" +#include "llist.h" +#include "memarea.h" +#include "istream-private.h" +#include "istream-chain.h" + +struct chain_istream; + +struct istream_chain_link { + struct istream_chain_link *prev, *next; + + struct istream *stream; + bool eof; +}; + +struct istream_chain { + struct istream_chain_link *head, *tail; + + struct chain_istream *stream; +}; + +struct chain_istream { + struct istream_private istream; + + /* how much of the previous link's stream still exists at the + beginning of our buffer. skipping through this should point to + the beginning of the current link's stream. */ + size_t prev_stream_left; + size_t prev_skip; + + struct istream_chain chain; +}; + +static void ATTR_NULL(2) +i_stream_chain_append_internal(struct istream_chain *chain, + struct istream *stream) +{ + struct istream_chain_link *link; + + if (stream == NULL && chain->tail != NULL && chain->tail->stream == NULL) + return; + + link = i_new(struct istream_chain_link, 1); + link->stream = stream; + link->eof = stream == NULL; + + if (stream != NULL) + i_stream_ref(stream); + + if (chain->head == NULL && stream != NULL) { + i_stream_set_max_buffer_size(stream, + chain->stream->istream.max_buffer_size); + } + DLLIST2_APPEND(&chain->head, &chain->tail, link); + /* if io_add_istream() has been added to this chain stream, notify + the callback that we have more data available. */ + if (stream != NULL) + i_stream_set_input_pending(stream, TRUE); +} + +void i_stream_chain_append(struct istream_chain *chain, struct istream *stream) +{ + i_stream_chain_append_internal(chain, stream); +} + +void i_stream_chain_append_eof(struct istream_chain *chain) +{ + i_stream_chain_append_internal(chain, NULL); +} + +static void +i_stream_chain_set_max_buffer_size(struct iostream_private *stream, + size_t max_size) +{ + struct chain_istream *cstream = + container_of(stream, struct chain_istream, istream.iostream); + struct istream_chain_link *link = cstream->chain.head; + + cstream->istream.max_buffer_size = max_size; + while (link != NULL) { + if (link->stream != NULL) + i_stream_set_max_buffer_size(link->stream, max_size); + link = link->next; + } +} + +static void i_stream_chain_destroy(struct iostream_private *stream) +{ + struct chain_istream *cstream = + container_of(stream, struct chain_istream, istream.iostream); + struct istream_chain_link *link = cstream->chain.head; + + while (link != NULL) { + struct istream_chain_link *next = link->next; + + i_stream_unref(&link->stream); + i_free(link); + link = next; + } + i_stream_free_buffer(&cstream->istream); +} + +static void i_stream_chain_read_next(struct chain_istream *cstream) +{ + struct istream_chain_link *link = cstream->chain.head; + struct istream *prev_input; + const unsigned char *data; + size_t data_size, cur_data_pos; + + i_assert(link != NULL && link->stream != NULL); + i_assert(link->stream->eof); + + prev_input = link->stream; + data = i_stream_get_data(prev_input, &data_size); + + DLLIST2_REMOVE(&cstream->chain.head, &cstream->chain.tail, link); + i_free(link); + + /* a) we have more streams, b) we have EOF, c) we need to wait + for more streams */ + link = cstream->chain.head; + if (link != NULL && link->stream != NULL) + i_stream_seek(link->stream, 0); + + if (cstream->prev_stream_left > 0) { + /* we've already buffered some of the prev_input. continue + appending the rest to it. if it's already at EOF, there's + nothing more to append. */ + cur_data_pos = cstream->istream.pos - + (cstream->istream.skip + cstream->prev_stream_left); + i_assert(cur_data_pos <= data_size); + data += cur_data_pos; + data_size -= cur_data_pos; + /* the stream has now become "previous", so its contents in + buffer are now part of prev_stream_left. */ + cstream->prev_stream_left += cur_data_pos; + } else { + cstream->istream.pos = 0; + cstream->istream.skip = 0; + cstream->prev_stream_left = 0; + } + + if (data_size > 0) { + if (cstream->istream.memarea != NULL && + memarea_get_refcount(cstream->istream.memarea) > 1) + i_stream_memarea_detach(&cstream->istream); + memcpy(i_stream_alloc(&cstream->istream, data_size), + data, data_size); + cstream->istream.pos += data_size; + cstream->prev_stream_left += data_size; + } + + i_stream_skip(prev_input, i_stream_get_data_size(prev_input)); + i_stream_unref(&prev_input); +} + +static bool i_stream_chain_skip(struct chain_istream *cstream) +{ + struct istream_private *stream = &cstream->istream; + struct istream_chain_link *link = cstream->chain.head; + size_t bytes_skipped; + + i_assert(stream->skip >= cstream->prev_skip); + bytes_skipped = stream->skip - cstream->prev_skip; + + if (cstream->prev_stream_left == 0) { + /* no need to worry about buffers, skip everything */ + } else if (bytes_skipped < cstream->prev_stream_left) { + /* we're still skipping inside buffer */ + cstream->prev_stream_left -= bytes_skipped; + bytes_skipped = 0; + } else { + /* done with the buffer */ + bytes_skipped -= cstream->prev_stream_left; + cstream->prev_stream_left = 0; + } + if (bytes_skipped > 0) { + i_assert(stream->buffer != NULL); + stream->pos -= bytes_skipped; + stream->skip -= bytes_skipped; + stream->buffer += bytes_skipped; + } + cstream->prev_skip = stream->skip; + if (link == NULL || link->eof) { + i_assert(bytes_skipped == 0); + return FALSE; + } + i_stream_skip(link->stream, bytes_skipped); + return TRUE; +} + +static ssize_t i_stream_chain_read(struct istream_private *stream) +{ + struct chain_istream *cstream = + container_of(stream, struct chain_istream, istream); + struct istream_chain_link *link = cstream->chain.head; + const unsigned char *data; + size_t data_size, cur_data_pos, new_pos; + size_t new_bytes_count; + ssize_t ret; + + if (link != NULL && link->eof) { + stream->istream.eof = TRUE; + return -1; + } + + if (!i_stream_chain_skip(cstream)) + return 0; + i_assert(link != NULL); + + i_assert(stream->pos >= stream->skip + cstream->prev_stream_left); + cur_data_pos = stream->pos - (stream->skip + cstream->prev_stream_left); + + data = i_stream_get_data(link->stream, &data_size); + if (data_size > cur_data_pos) + ret = 0; + else { + /* need to read more */ + i_assert(cur_data_pos == data_size); + ret = i_stream_read_memarea(link->stream); + if (ret == -2 || ret == 0) + return ret; + + if (ret == -1) { + if (link->stream->stream_errno != 0) { + io_stream_set_error(&stream->iostream, + "read(%s) failed: %s", + i_stream_get_name(link->stream), + i_stream_get_error(link->stream)); + stream->istream.stream_errno = + link->stream->stream_errno; + return -1; + } + /* EOF of this stream, go to next stream */ + i_stream_chain_read_next(cstream); + cstream->prev_skip = stream->skip; + return i_stream_chain_read(stream); + } + /* we read something */ + data = i_stream_get_data(link->stream, &data_size); + } + + if (data_size == cur_data_pos) { + /* nothing new read - preserve the buffer as it was */ + i_assert(ret == 0 || ret == -1); + return ret; + } + if (cstream->prev_stream_left == 0) { + /* we can point directly to the current stream's buffers */ + stream->buffer = data; + stream->pos -= stream->skip; + stream->skip = 0; + new_pos = data_size; + } else { + /* we still have some of the previous stream left. merge the + new data with it. */ + i_assert(data_size > cur_data_pos); + new_bytes_count = data_size - cur_data_pos; + memcpy(i_stream_alloc(stream, new_bytes_count), + data + cur_data_pos, new_bytes_count); + stream->buffer = stream->w_buffer; + new_pos = stream->pos + new_bytes_count; + } + + i_assert(new_pos > stream->pos); + ret = (ssize_t)(new_pos - stream->pos); + stream->pos = new_pos; + cstream->prev_skip = stream->skip; + return ret; +} + +static void i_stream_chain_close(struct iostream_private *stream, + bool close_parent) +{ + struct chain_istream *cstream = + container_of(stream, struct chain_istream, istream.iostream); + + /* seek to the correct position in parent stream in case it didn't + end with EOF */ + (void)i_stream_chain_skip(cstream); + + if (close_parent) { + struct istream_chain_link *link = cstream->chain.head; + while (link != NULL) { + i_stream_close(link->stream); + link = link->next; + } + } +} + +static struct istream_snapshot * +i_stream_chain_snapshot(struct istream_private *stream, + struct istream_snapshot *prev_snapshot) +{ + if (stream->buffer == stream->w_buffer) { + /* Two or more istreams have been combined. Snapshot the + w_buffer's contents that contains their data. */ + i_assert(stream->memarea != NULL); + return i_stream_default_snapshot(stream, prev_snapshot); + } + /* Individual istreams are being read. Snapshot the istream directly. */ + struct chain_istream *cstream = + container_of(stream, struct chain_istream, istream); + struct istream_chain_link *link = cstream->chain.head; + if (link == NULL || link->stream == NULL) + return prev_snapshot; + + struct istream_private *_link_stream = link->stream->real_stream; + struct istream_snapshot *snapshot = i_new(struct istream_snapshot, 1); + snapshot->prev_snapshot = + _link_stream->snapshot(_link_stream, prev_snapshot); + if (snapshot->prev_snapshot == prev_snapshot) { + /* The link stream didn't implement snapshotting in any way. + This could cause trouble if the link stream is freed while + it's still referred to in this snapshot. Fix this by + referencing the link istream. Normally avoid doing this, + since the extra references can cause unexpected problems. */ + snapshot->istream = link->stream; + i_stream_ref(snapshot->istream); + } + return snapshot; +} + +struct istream *i_stream_create_chain(struct istream_chain **chain_r, + size_t max_buffer_size) +{ + struct chain_istream *cstream; + + cstream = i_new(struct chain_istream, 1); + cstream->chain.stream = cstream; + cstream->istream.max_buffer_size = max_buffer_size; + + cstream->istream.iostream.close = i_stream_chain_close; + cstream->istream.iostream.destroy = i_stream_chain_destroy; + cstream->istream.iostream.set_max_buffer_size = + i_stream_chain_set_max_buffer_size; + + cstream->istream.read = i_stream_chain_read; + cstream->istream.snapshot = i_stream_chain_snapshot; + + cstream->istream.istream.readable_fd = FALSE; + cstream->istream.istream.blocking = FALSE; + cstream->istream.istream.seekable = FALSE; + + *chain_r = &cstream->chain; + return i_stream_create(&cstream->istream, NULL, -1, 0); +} |