Mercurial > hg > ucx / changeset

--- a/src/buffer.c	Mon Jul 25 14:16:49 2022 +0200
+++ b/src/buffer.c	Mon Jul 25 15:29:56 2022 +0200
@@ -248,7 +248,8 @@
                 // try again with the remaining stuff
                 unsigned char const *new_ptr = ptr;
                 new_ptr += items_flush * size;
-                return cxBufferWrite(new_ptr, size, items_keep, buffer);
+                // report the directly flushed items as written plus the remaining stuff
+                return items_flush + cxBufferWrite(new_ptr, size, items_keep, buffer);
             } else {
                 // all items have been flushed - report them as written
                 return nitems;
@@ -257,7 +258,7 @@
             // nothing could be flushed at all, we immediately give up without writing any data
             return 0;
         } else {
-            // we were partially successful, we have shift left and try again
+            // we were partially successful, we shift left and try again
             cxBufferShiftLeft(buffer, flush_pos);
             return cxBufferWrite(ptr, size, nitems, buffer);
         }

--- a/src/cx/buffer.h	Mon Jul 25 14:16:49 2022 +0200
+++ b/src/cx/buffer.h	Mon Jul 25 15:29:56 2022 +0200
@@ -310,7 +310,20 @@
 /**
  * Writes data to a CxBuffer.
  *
- * The position of the buffer is increased by the number of bytes written.
+ * If flushing is enabled and the buffer needs to flush, the data is flushed to
+ * the target until the target signals that it cannot take more data by
+ * returning zero via the respective write function. In that case, the remaining
+ * data in this buffer is shifted to the beginning of this buffer so that the
+ * newly available space can be used to append as much data as possible. This
+ * function only stops writing more elements, when the flush target and this
+ * buffer are both incapable of taking more data or all data has been written.
+ * The number returned by this function is the total number of elements that
+ * could be written during the process. It does not necessarily mean that those
+ * elements are still in this buffer, because some of them could have also be
+ * flushed already.
+ *
+ * If automatic flushing is not enabled, the position of the buffer is increased
+ * by the number of bytes written.
  *
  * \note The signature is compatible with the fwrite() family of functions.
  *

--- a/test/test_buffer.cpp	Mon Jul 25 14:16:49 2022 +0200
+++ b/test/test_buffer.cpp	Mon Jul 25 15:29:56 2022 +0200
@@ -351,6 +351,23 @@
     }
 };
 
+static size_t mock_write_limited_rate(
+        void const *ptr,
+        size_t size,
+        __attribute__((unused)) size_t nitems,
+        CxBuffer *buffer
+) {
+    // simulate limited target drain capacity
+    static bool full = false;
+    if (full) {
+        full = false;
+        return 0;
+    } else {
+        full = true;
+        return cxBufferWrite(ptr, size, nitems > 2 ? 2 : nitems, buffer);
+    }
+}
+
 TEST_F(BufferWrite, SizeOneFit) {
     const char *data = "test";
     ASSERT_EQ(buf.capacity, 8);
@@ -595,6 +612,33 @@
     EXPECT_EQ(memcmp(target.space, "prepfoobarhello", 15), 0);
 }
 
+TEST_F(BufferWrite, FlushRateLimited) {
+    enableFlushing();
+    // limit the rate of the flush function and the capacity of the target
+    target.capacity = 16;
+    target.flags &= ~CX_BUFFER_AUTO_EXTEND;
+    buf.flush_func = (cx_write_func) mock_write_limited_rate;
+    ASSERT_EQ(buf.capacity, 8);
+    ASSERT_EQ(buf.pos, 4);
+    size_t written = cxBufferWrite("foo", 1, 3, &buf);
+    EXPECT_EQ(written, 3);
+    ASSERT_EQ(buf.pos, 7);
+    ASSERT_EQ(buf.size, 7);
+    ASSERT_EQ(target.pos, 0);
+    ASSERT_EQ(target.size, 0);
+    written = cxBufferWrite("hello, world!", 1, 13, &buf);
+    // " world!" fits into this buffer, the remaining stuff is flushed out
+    EXPECT_EQ(written, 13);
+    EXPECT_EQ(buf.pos, 7);
+    EXPECT_EQ(buf.size, 7);
+    EXPECT_EQ(buf.capacity, 8);
+    EXPECT_EQ(memcmp(buf.space, " world!", 7), 0);
+    EXPECT_EQ(target.pos, 13);
+    ASSERT_EQ(target.size, 13);
+    EXPECT_EQ(target.capacity, 16);
+    EXPECT_EQ(memcmp(target.space, "prepfoohello,", 13), 0);
+}
+
 class BufferSeek : public BufferFixture {
 };