Fix piped producer error (Unable to read message: ..)

Author: edenhill

input.c

@@ -328,7 +328,7 @@ int inbuf_read_to_delimeter (struct inbuf *inbuf, FILE *fp,
         if (!inbuf->buf)
                 return 0;  /* Previous EOF encountered, see below. */
 
-        while (conf.run && 1) {
+        while (conf.run) {
                 ssize_t r;
                 size_t dof;
                 int delim_found;

kcat.c

@@ -416,12 +416,13 @@ static void producer_run (FILE *fp, char **paths, int pathcnt) {
         } else {
                 struct inbuf inbuf;
                 struct buf *b;
+                int at_eof = 0;
 
                 inbuf_init(&inbuf, conf.msg_size, conf.delim, conf.delim_size);
 
                 /* Read messages from input, delimited by conf.delim */
                 while (conf.run &&
-                       inbuf_read_to_delimeter(&inbuf, fp, &b)) {
+                       !(at_eof = !inbuf_read_to_delimeter(&inbuf, fp, &b))) {
                         int msgflags = 0;
                         char *buf = b->buf;
                         char *key = NULL;
@@ -486,11 +487,9 @@ static void producer_run (FILE *fp, char **paths, int pathcnt) {
                                 conf.run = 0;
                 }
 
-                if (conf.run) {
-                        if (!feof(fp))
-                                KC_FATAL("Unable to read message: %s",
-                                         strerror(errno));
-                }
+                if (conf.run && !at_eof)
+                        KC_FATAL("Unable to read message: %s",
+                                 strerror(errno));
         }
 
 #if ENABLE_TXNS

tests/0004-piped_producer.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+#
+
+set -e
+source helpers.sh
+
+
+#
+# Verify that piping messages to the producer works.
+#
+
+
+topic=$(make_topic_name)
+
+
+
+echo "msg1" | $KCAT -P -t $topic