.. _stream: ***************** Stream processing ***************** .. highlight:: c :: #include Stream producers ---------------- A *producer* of binary data should take in a pointer to a :c:type:`cork_stream_consumer` instance. Any data that is produced by the stream is then sent into the consumer instance for processing. Once the stream has been exhausted (for instance, by reaching the end of a file), you signal this to the consumer. During both of these steps, the consumer is able to signal error conditions; for instance, a stream consumer that parses a particular file format might return an error condition if the stream of data is malformed. If possible, the stream producer can try to recover from the error condition, but more often, the stream producer will simply pass the error back up to its caller. .. function:: int cork_stream_consumer_data(struct cork_stream_consumer \*consumer, const void \*buf, size_t size, bool is_first_chunk) Send the next chunk of data into a stream consumer. You only have to ensure that *buf* is valid for the duration of this function call; the stream consumer is responsible for saving a copy of the data if it needs to be processed later. In particular, this means that it's perfectly safe for *buf* to refer to a stack-allocated memory region. .. function:: int cork_stream_consumer_eof(struct cork_stream_consumer \*consumer) Notify the stream consumer that the end of the stream has been reached. The stream consumer might perform some final validation and error detection at this point. .. function:: void cork_stream_consumer_free(struct cork_stream_consumer \*consumer) Finalize and deallocate a stream consumer. Built-in stream producers ~~~~~~~~~~~~~~~~~~~~~~~~~ We provide several built-in stream producers: .. function:: int cork_consume_fd(struct cork_stream_consumer \*consumer, int fd) int cork_consume_file(struct cork_stream_consumer \*consumer, FILE \*fp) int cork_consume_file_from_path(struct cork_stream_consumer \*consumer, const char \*path, int flags) Read in a file, passing its contents into the given stream consumer. The ``_fd`` and ``_file`` variants consume a file that you've already opened; you are responsible for closing the file after its been consumed. The ``_file_from_path`` variant will open the file for you, using the standard ``open(2)`` function with the given *flags*. This variant will close the file before returning, regardless of whether the file was successfully consumed or not. File stream producer example ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ As an example, we could implement the :c:func:`cork_consume_file` stream producer as follows:: #include #include #include #include #define BUFFER_SIZE 65536 int cork_consume_file(struct cork_stream_consumer *consumer, FILE *fp) { char buf[BUFFER_SIZE]; size_t bytes_read; bool first = true; while ((bytes_read = fread(buf, 1, BUFFER_SIZE, fp)) > 0) { rii_check(cork_stream_consumer_data(consumer, buf, bytes_read, first)); first = false; } if (feof(fp)) { return cork_stream_consumer_eof(consumer); } else { cork_system_error_set(); return -1; } } Note that this stream producer does not take care of opening or closing the ``FILE`` object, nor does it take care of freeing the consumer. (Our actual implementation of :c:func:`cork_consume_file` also correctly handles ``EINTR`` errors, and so is a bit more complex. But this example still works as an illustration of how to pass data into a stream consumer.) .. _stream-consumers: Stream consumers ---------------- To consume data from a stream, you must create a type that implements the :c:type:`cork_stream_consumer` interface. .. type:: struct cork_stream_consumer An interface for consumer a stream of binary data. The producer of the stream will call the :c:func:`cork_stream_consumer_data()` function repeatedly, once for each successive chunk of data in the stream. Once the stream has been exhausted, the producer will call :c:func:`cork_stream_consumer_eof()` to signal the end of the stream. .. member:: int (\*data)(struct cork_stream_consumer \*consumer, const void \*buf, size_t size, bool is_first_chunk) Process the next chunk of data in the stream. *buf* is only guaranteed to be valid for the duration of this function call. If you need to access the contents of the slice later, you must save it somewhere yourself. If there is an error processing this chunk of data, you should return ``-1`` and fill in the current error condition using :c:func:`cork_error_set`. .. member:: int (\*eof)(struct cork_stream_consumer \*consumer) Handle the end of the stream. This allows you to defer any final validation or error detection until all of the data has been processed. If there is an error detected at this point, you should return ``-1`` and fill in the current error condition using :c:func:`cork_error_set`. .. member:: void (\*free)(struct cork_stream_consumer \*consumer) Free the consumer object. Built-in stream consumers ~~~~~~~~~~~~~~~~~~~~~~~~~ We provide several built-in stream consumers: .. function:: struct cork_stream_consumer \*cork_fd_consumer_new(int fd) struct cork_stream_consumer \*cork_file_consumer_new(FILE \*fp) struct cork_stream_consumer \*cork_file_from_path_consumer_new(const char \*path, int flags) Create a stream consumer that appends any data that it receives to a file. The ``_fd`` and ``_file`` variants append to a file that you've already opened; you are responsible for closing the file after the consumer has finished processing data. The ``_file_from_path`` variant will open the file for you, using the standard ``open(2)`` function with the given *flags*. This variant will close the file before returning, regardless of whether the stream consumer successfully processed the data or not. File stream consumer example ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ As an example, we could implement a stream consumer for the :c:func:`cork_file_consumer_new` function as follows:: #include #include #include #include struct cork_file_consumer { /* cork_file_consumer implements the cork_stream_consumer interface */ struct cork_stream_consumer parent; /* the file to write the data into */ FILE *fp; }; static int cork_file_consumer__data(struct cork_stream_consumer *vself, const void *buf, size_t size, bool is_first) { struct file_consumer *self = cork_container_of(vself, struct cork_file_consumer, parent); size_t bytes_written = fwrite(buf, 1, size, self->fp); /* If there was an error writing to the file, then signal this to * the producer */ if (bytes_written == size) { return 0; } else { cork_system_error_set(); return -1; } } static int cork_file_consumer__eof(struct cork_stream_consumer *vself) { /* We don't close the file, so there's nothing special to do at * end-of-stream. */ return 0; } static void cork_file_consumer__free(struct cork_stream_consumer *vself) { struct file_consumer *self = cork_container_of(vself, struct cork_file_consumer, parent); free(self); } struct cork_stream_consumer * cork_file_consumer_new(FILE *fp) { struct cork_file_consumer *self = cork_new(struct cork_file_consumer); self->parent.data = cork_file_consumer__data; self->parent.eof = cork_file_consumer__eof; self->parent.free = cork_file_consumer__free; self->fp = fp; return &self->parent; } Note that this stream consumer does not take care of opening or closing the ``FILE`` object.