til_stream: introduce til_stream_{end,active}()

rkt needs a way to signal the end of a sequence, this will
probably get more work in the future but something simple is fine
for now

2 files changed, 19 insertions, 0 deletions

diff --git a/src/til_stream.c b/src/til_stream.c
index c24f239..a28e138 100644
--- a/src/til_stream.c
+++ b/src/til_stream.c
@@ -91,6 +91,7 @@ struct til_stream_module_context_t {
 
 typedef struct til_stream_t {
 	pthread_mutex_t			mutex;
+	volatile int			ended;
 	const til_stream_hooks_t	*hooks;
 	void				*hooks_context;
 	til_stream_pipe_t		*pipe_buckets[TIL_STREAM_PIPE_BUCKETS_COUNT];
@@ -112,6 +113,22 @@ til_stream_t * til_stream_new(void)
 }
 
 
+void til_stream_end(til_stream_t *stream)
+{
+	assert(stream);
+
+	stream->ended = 1;
+}
+
+
+int til_stream_active(til_stream_t *stream)
+{
+	assert(stream);
+
+	return !stream->ended;
+}
+
+
 til_stream_t * til_stream_free(til_stream_t *stream)
 {
 	if (!stream)
diff --git a/src/til_stream.h b/src/til_stream.h
index cb3c98e..e6346fd 100644
--- a/src/til_stream.h
+++ b/src/til_stream.h
@@ -45,6 +45,8 @@ typedef struct til_stream_hooks_t {
 
 til_stream_t * til_stream_new(void);
 til_stream_t * til_stream_free(til_stream_t *stream);
+void til_stream_end(til_stream_t *stream);
+int til_stream_active(til_stream_t *stream);
 int til_stream_set_hooks(til_stream_t *stream, const til_stream_hooks_t *hooks, void *context);
 int til_stream_unset_hooks(til_stream_t *stream, const til_stream_hooks_t *hooks);