libsubprocess: support fbuf credit callback

Problem: It would be convenient to be notified when data has
been read from a buffer, so that more can be written into it.

Support a credit callback.  The callback will be called after some
data has been read out of the buffer and inform the caller of
the number of bytes read.

Add unit tests.

src/common/libsubprocess/fbuf.c

@@ -32,6 +32,8 @@ struct fbuf {
     int buflen;
     fbuf_notify_f notify_cb;
     void *notify_cb_arg;
+    fbuf_credit_f credit_cb;
+    void *credit_cb_arg;
 };
 
 struct fbuf *fbuf_create (int size)
@@ -82,6 +84,14 @@ void fbuf_set_notify (struct fbuf *fb, fbuf_notify_f notify_cb, void *arg)
     }
 }
 
+void fbuf_set_credit (struct fbuf *fb, fbuf_credit_f credit_cb, void *arg)
+{
+    if (fb) {
+        fb->credit_cb = credit_cb;
+        fb->credit_cb_arg = arg;
+    }
+}
+
 void fbuf_destroy (void *data)
 {
     struct fbuf *fb = data;
@@ -161,6 +171,12 @@ static void nonfull_transition_check (struct fbuf *fb, bool was_full)
     }
 }
 
+static void credit_check (struct fbuf *fb, int bytes)
+{
+    if (fb->credit_cb && bytes)
+        fb->credit_cb (fb, bytes, fb->credit_cb_arg);
+}
+
 /* check if internal buffer can hold data from user */
 static int return_buffer_check (struct fbuf *fb)
 {
@@ -219,6 +235,7 @@ const void *fbuf_read (struct fbuf *fb, int len, int *lenp)
         (*lenp) = ret;
 
     nonfull_transition_check (fb, full);
+    credit_check (fb, ret);
 
     return fb->buf;
 }
@@ -278,6 +295,7 @@ const void *fbuf_read_line (struct fbuf *fb, int *lenp)
         (*lenp) = ret;
 
     nonfull_transition_check (fb, full);
+    credit_check (fb, ret);
 
     return fb->buf;
 }
@@ -316,6 +334,7 @@ int fbuf_read_to_fd (struct fbuf *fb, int fd, int len)
         return -1;
 
     nonfull_transition_check (fb, full);
+    credit_check (fb, ret);
 
     return ret;
 }

src/common/libsubprocess/fbuf.h

@@ -90,6 +90,14 @@ typedef void (*fbuf_notify_f) (struct fbuf *fb, void *arg);
  */
 void fbuf_set_notify (struct fbuf *fb, fbuf_notify_f notify_cb, void *arg);
 
+typedef void (*fbuf_credit_f) (struct fbuf *fb, int len, void *arg);
+
+/* Set credit callback for internal use by fbuf watchers.  The callback
+ * is invoked when buffer reads happen, giving "credit" to the buffer and
+ * allowing more writes to happen.
+ */
+void fbuf_set_credit (struct fbuf *fb, fbuf_credit_f credit_cb, void *arg);
+
 #endif /* !_LIBSUBPROCESS_FBUF_H */
 
 // vi: ts=4 sw=4 expandtab

src/common/libsubprocess/test/fbuf.c

@@ -219,6 +219,51 @@ void notify_callback (void)
     fbuf_destroy (fb);
 }
 
+struct credit_count {
+    int count;
+    int last_bytes;
+};
+
+void credit_cb (struct fbuf *fb, int bytes, void *arg)
+{
+    struct credit_count *cp = arg;
+    cp->count++;
+    cp->last_bytes = bytes;
+}
+
+void credit_callback (void)
+{
+    struct fbuf *fb;
+    struct credit_count credit = {0};
+    int len;
+
+    ok ((fb = fbuf_create (16)) != NULL,
+        "fbuf_create 16 byte buffer works");
+    fbuf_set_credit (fb, credit_cb, &credit);
+
+    ok (fbuf_write (fb, "foobar", 6) == 6,
+        "fbuf_write 6 bytes");
+
+    ok (credit.count == 0,
+        "read callback not called");
+
+    ok (fbuf_read (fb, 1, &len) != NULL && len == 1,
+        "fbuf_read read one byte");
+
+    ok (credit.count == 1
+        && credit.last_bytes == 1,
+        "credit callback with 1 byte");
+
+    ok (fbuf_read (fb, -1, &len) != NULL && len == 5,
+        "fbuf_read cleared all data");
+
+    ok (credit.count == 2
+        && credit.last_bytes == 5,
+        "credit callback with 5 bytes");
+
+    fbuf_destroy (fb);
+}
+
 void corner_case (void)
 {
     struct fbuf *fb;
@@ -431,6 +476,7 @@ int main (int argc, char *argv[])
 
     basic ();
     notify_callback ();
+    credit_callback ();
     corner_case ();
     full_buffer ();
     readonly_buffer ();