rbuffer: Reimplement as a ring buffer and decouple from rstream

Extract the RBuffer class from rstream.c and reimplement it as a ring buffer,
a more efficient version that doesn't need to relocate memory.

The old rbuffer_read/rbuffer_write interfaces are kept for simple
reading/writing, and the RBUFFER_UNTIL_{FULL,EMPTY} macros are introduced to
hide wrapping logic when more control is required(such as passing the buffer
pointer to a library function that writes directly to the pointer)

Also add a basic infrastructure for writing helper C files that are only
compiled in the unit test library, and use this to write unit tests for RBuffer
which contains some macros that can't be accessed directly by luajit.

Helped-by: oni-link <knil.ino@gmail.com>
Reviewed-by: oni-link <knil.ino@gmail.com>
Reviewed-by: Scott Prager <splinterofchaos@gmail.com>
Reviewed-by: Justin M. Keyes <justinkz@gmail.com>
Reviewed-by: Michael Reed <m.reed@mykolab.com>

16 files changed, 789 insertions, 271 deletions

diff --git a/src/nvim/CMakeLists.txt b/src/nvim/CMakeLists.txt
index 2e45b5e9d6..880ff42ba5 100644
--- a/src/nvim/CMakeLists.txt
+++ b/src/nvim/CMakeLists.txt
@@ -40,6 +40,7 @@ file(MAKE_DIRECTORY ${GENERATED_INCLUDES_DIR}/tui)
 file(GLOB NEOVIM_SOURCES *.c os/*.c api/*.c api/private/*.c msgpack_rpc/*.c
   tui/*.c)
 file(GLOB_RECURSE NEOVIM_HEADERS *.h)
+file(GLOB UNIT_TEST_FIXTURES ${PROJECT_SOURCE_DIR}/test/unit/fixtures/*.c)
 
 foreach(sfile ${NEOVIM_SOURCES})
   get_filename_component(f ${sfile} NAME)
@@ -205,7 +206,7 @@ set_target_properties(libnvim PROPERTIES
 set_property(TARGET libnvim APPEND_STRING PROPERTY COMPILE_FLAGS " -DMAKE_LIB ")
 
 add_library(nvim-test MODULE EXCLUDE_FROM_ALL ${NEOVIM_GENERATED_SOURCES}
-    ${NEOVIM_SOURCES} ${NEOVIM_HEADERS})
+    ${NEOVIM_SOURCES} ${UNIT_TEST_FIXTURES} ${NEOVIM_HEADERS})
 target_link_libraries(nvim-test ${NVIM_LINK_LIBRARIES})
 set_property(TARGET nvim-test APPEND_STRING PROPERTY COMPILE_FLAGS -DUNIT_TESTING)
 
diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index 93590445c9..d3ab47a505 100644
--- a/src/nvim/eval.c
+++ b/src/nvim/eval.c
@@ -20341,19 +20341,19 @@ static inline void push_job_event(Job *job, ufunc_T *callback,
   }, !disable_job_defer);
 }
 
-static void on_job_stdout(RStream *rstream, void *job, bool eof)
+static void on_job_stdout(RStream *rstream, RBuffer *buf, void *job, bool eof)
 {
   TerminalJobData *data = job_data(job);
-  on_job_output(rstream, job, eof, data->on_stdout, "stdout");
+  on_job_output(rstream, job, buf, eof, data->on_stdout, "stdout");
 }
 
-static void on_job_stderr(RStream *rstream, void *job, bool eof)
+static void on_job_stderr(RStream *rstream, RBuffer *buf, void *job, bool eof)
 {
   TerminalJobData *data = job_data(job);
-  on_job_output(rstream, job, eof, data->on_stderr, "stderr");
+  on_job_output(rstream, job, buf, eof, data->on_stderr, "stderr");
 }
 
-static void on_job_output(RStream *rstream, Job *job, bool eof,
+static void on_job_output(RStream *rstream, Job *job, RBuffer *buf, bool eof,
     ufunc_T *callback, const char *type)
 {
   if (eof) {
@@ -20361,20 +20361,19 @@ static void on_job_output(RStream *rstream, Job *job, bool eof,
   }
 
   TerminalJobData *data = job_data(job);
-  char *ptr = rstream_read_ptr(rstream);
-  size_t len = rstream_pending(rstream);
+  RBUFFER_UNTIL_EMPTY(buf, ptr, len) {
+    // The order here matters, the terminal must receive the data first because
+    // push_job_event will modify the read buffer(convert NULs into NLs)
+    if (data->term) {
+      terminal_receive(data->term, ptr, len);
+    }
 
-  // The order here matters, the terminal must receive the data first because
-  // push_job_event will modify the read buffer(convert NULs into NLs)
-  if (data->term) {
-    terminal_receive(data->term, ptr, len);
-  }
+    if (callback) {
+      push_job_event(job, callback, type, ptr, len, 0);
+    }
 
-  if (callback) {
-    push_job_event(job, callback, type, ptr, len, 0);
+    rbuffer_consumed(buf, len);
   }
-
-  rbuffer_consumed(rstream_buffer(rstream), len);
 }
 
 static void on_job_exit(Job *job, int status, void *d)
diff --git a/src/nvim/memory.h b/src/nvim/memory.h
index 4ff31ff732..7b477da2f5 100644
--- a/src/nvim/memory.h
+++ b/src/nvim/memory.h
@@ -1,6 +1,7 @@
 #ifndef NVIM_MEMORY_H
 #define NVIM_MEMORY_H
 
+#include <stdint.h>  // for uint8_t
 #include <stddef.h>  // for size_t
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index df78f822d6..2a81b4f160 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -328,19 +328,17 @@ static void channel_from_stdio(void)
   channel->data.streams.uv = NULL;
 }
 
-static void job_out(RStream *rstream, void *data, bool eof)
+static void job_out(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
   Job *job = data;
-  parse_msgpack(rstream, job_data(job), eof);
+  parse_msgpack(rstream, buf, job_data(job), eof);
 }
 
-static void job_err(RStream *rstream, void *data, bool eof)
+static void job_err(RStream *rstream, RBuffer *rbuf, void *data, bool eof)
 {
-  size_t count;
-  char buf[256];
-
-  while ((count = rstream_pending(rstream))) {
-    size_t read = rstream_read(rstream, buf, sizeof(buf) - 1);
+  while (rbuffer_size(rbuf)) {
+    char buf[256];
+    size_t read = rbuffer_read(rbuf, buf, sizeof(buf) - 1);
     buf[read] = NUL;
     ELOG("Channel %" PRIu64 " stderr: %s",
          ((Channel *)job_data(data))->id, buf);
@@ -352,7 +350,7 @@ static void job_exit(Job *job, int status, void *data)
   decref(data);
 }
 
-static void parse_msgpack(RStream *rstream, void *data, bool eof)
+static void parse_msgpack(RStream *rstream, RBuffer *rbuf, void *data, bool eof)
 {
   Channel *channel = data;
   incref(channel);
@@ -363,14 +361,14 @@ static void parse_msgpack(RStream *rstream, void *data, bool eof)
     goto end;
   }
 
-  size_t count = rstream_pending(rstream);
+  size_t count = rbuffer_size(rbuf);
   DLOG("Feeding the msgpack parser with %u bytes of data from RStream(%p)",
        count,
        rstream);
 
   // Feed the unpacker with data
   msgpack_unpacker_reserve_buffer(channel->unpacker, count);
-  rstream_read(rstream, msgpack_unpacker_buffer(channel->unpacker), count);
+  rbuffer_read(rbuf, msgpack_unpacker_buffer(channel->unpacker), count);
   msgpack_unpacker_buffer_consumed(channel->unpacker, count);
 
   msgpack_unpacked unpacked;
diff --git a/src/nvim/os/input.c b/src/nvim/os/input.c
index 74a5d3bc2e..726335bd9a 100644
--- a/src/nvim/os/input.c
+++ b/src/nvim/os/input.c
@@ -79,7 +79,7 @@ void input_stop(void)
 // Low level input function
 int os_inchar(uint8_t *buf, int maxlen, int ms, int tb_change_cnt)
 {
-  if (rbuffer_pending(input_buffer)) {
+  if (rbuffer_size(input_buffer)) {
     return (int)rbuffer_read(input_buffer, (char *)buf, (size_t)maxlen);
   }
 
@@ -108,7 +108,7 @@ int os_inchar(uint8_t *buf, int maxlen, int ms, int tb_change_cnt)
     return 0;
   }
 
-  if (rbuffer_pending(input_buffer)) {
+  if (rbuffer_size(input_buffer)) {
     // Safe to convert rbuffer_read to int, it will never overflow since we use
     // relatively small buffers.
     return (int)rbuffer_read(input_buffer, (char *)buf, (size_t)maxlen);
@@ -153,7 +153,7 @@ size_t input_enqueue(String keys)
 {
   char *ptr = keys.data, *end = ptr + keys.size;
 
-  while (rbuffer_available(input_buffer) >= 6 && ptr < end) {
+  while (rbuffer_space(input_buffer) >= 6 && ptr < end) {
     uint8_t buf[6] = {0};
     unsigned int new_size = trans_special((uint8_t **)&ptr, buf, true);
 
@@ -309,16 +309,17 @@ static InbufPollResult inbuf_poll(int ms)
   return input_eof ? kInputEof : kInputNone;
 }
 
-static void read_cb(RStream *rstream, void *data, bool at_eof)
+static void read_cb(RStream *rstream, RBuffer *buf, void *data, bool at_eof)
 {
   if (at_eof) {
     input_eof = true;
   }
 
-  char *buf = rbuffer_read_ptr(read_buffer);
-  size_t buf_size = rbuffer_pending(read_buffer);
-  (void)rbuffer_write(input_buffer, buf, buf_size);
-  rbuffer_consumed(read_buffer, buf_size);
+  assert(rbuffer_space(input_buffer) >= rbuffer_size(read_buffer));
+  RBUFFER_UNTIL_EMPTY(read_buffer, ptr, len) {
+    (void)rbuffer_write(input_buffer, ptr, len);
+    rbuffer_consumed(read_buffer, len);
+  }
 }
 
 static void process_interrupts(void)
@@ -327,18 +328,16 @@ static void process_interrupts(void)
     return;
   }
 
-  char *inbuf = rbuffer_read_ptr(input_buffer);
-  size_t count = rbuffer_pending(input_buffer), consume_count = 0;
-
-  for (int i = (int)count - 1; i >= 0; i--) {
-    if (inbuf[i] == 3) {
+  size_t consume_count = 0;
+  RBUFFER_EACH_REVERSE(input_buffer, c, i) {
+    if ((uint8_t)c == 3) {
       got_int = true;
-      consume_count = (size_t)i;
+      consume_count = i;
       break;
     }
   }
 
-  if (got_int) {
+  if (got_int && consume_count) {
     // Remove everything typed before the CTRL-C
     rbuffer_consumed(input_buffer, consume_count);
   }
@@ -362,7 +361,7 @@ static int push_event_key(uint8_t *buf, int maxlen)
 static bool input_ready(void)
 {
   return typebuf_was_filled ||                 // API call filled typeahead
-         rbuffer_pending(input_buffer) > 0 ||  // Input buffer filled
+         rbuffer_size(input_buffer) ||         // Input buffer filled
          event_has_deferred();                 // Events must be processed
 }
 
diff --git a/src/nvim/os/job.c b/src/nvim/os/job.c
index 038d0e3c26..4769ee4d2f 100644
--- a/src/nvim/os/job.c
+++ b/src/nvim/os/job.c
@@ -404,17 +404,17 @@ static void job_stop_timer_cb(uv_timer_t *handle)
 }
 
 // Wraps the call to std{out,err}_cb and emits a JobExit event if necessary.
-static void read_cb(RStream *rstream, void *data, bool eof)
+static void read_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
   Job *job = data;
 
   if (rstream == job->out) {
-    job->opts.stdout_cb(rstream, data, eof);
+    job->opts.stdout_cb(rstream, buf, data, eof);
     if (eof) {
       close_job_out(job);
     }
   } else {
-    job->opts.stderr_cb(rstream, data, eof);
+    job->opts.stderr_cb(rstream, buf, data, eof);
     if (eof) {
       close_job_err(job);
     }
diff --git a/src/nvim/os/rstream.c b/src/nvim/os/rstream.c
index a99745f068..af84288f0f 100644
--- a/src/nvim/os/rstream.c
+++ b/src/nvim/os/rstream.c
@@ -14,12 +14,6 @@
 #include "nvim/log.h"
 #include "nvim/misc1.h"
 
-struct rbuffer {
-  char *data;
-  size_t capacity, rpos, wpos;
-  RStream *rstream;
-};
-
 struct rstream {
   void *data;
   uv_buf_t uvbuf;
@@ -37,135 +31,6 @@ struct rstream {
 # include "os/rstream.c.generated.h"
 #endif
 
-/// Creates a new `RBuffer` instance.
-RBuffer *rbuffer_new(size_t capacity)
-{
-  RBuffer *rv = xmalloc(sizeof(RBuffer));
-  rv->data = xmalloc(capacity);
-  rv->capacity = capacity;
-  rv->rpos = rv->wpos = 0;
-  rv->rstream = NULL;
-  return rv;
-}
-
-/// Advances `rbuffer` read pointers to consume data. If the associated
-/// RStream had stopped because the buffer was full, this will restart it.
-///
-/// This is called automatically by rbuffer_read, but when using
-/// `rbuffer_read_ptr` directly, this needs to called after the data was
-/// consumed.
-void rbuffer_consumed(RBuffer *rbuffer, size_t count)
-{
-  rbuffer->rpos += count;
-  if (count && rbuffer->wpos == rbuffer->capacity) {
-    // `wpos` is at the end of the buffer, so free some space by moving unread
-    // data...
-    rbuffer_relocate(rbuffer);
-    if (rbuffer->rstream) {
-      // restart the associated RStream
-      rstream_start(rbuffer->rstream);
-    }
-  }
-}
-
-/// Advances `rbuffer` write pointers. If the internal buffer becomes full,
-/// this will stop the associated RStream instance.
-void rbuffer_produced(RBuffer *rbuffer, size_t count)
-{
-  rbuffer->wpos += count;
-  DLOG("Received %u bytes from RStream(%p)", (size_t)count, rbuffer->rstream);
-
-  rbuffer_relocate(rbuffer);
-  if (rbuffer->rstream && rbuffer->wpos == rbuffer->capacity) {
-    // The last read filled the buffer, stop reading for now
-    //
-    rstream_stop(rbuffer->rstream);
-    DLOG("Buffer for RStream(%p) is full, stopping it", rbuffer->rstream);
-  }
-}
-
-/// Reads data from a `RBuffer` instance into a raw buffer.
-///
-/// @param rbuffer The `RBuffer` instance
-/// @param buffer The buffer which will receive the data
-/// @param count Number of bytes that `buffer` can accept
-/// @return The number of bytes copied into `buffer`
-size_t rbuffer_read(RBuffer *rbuffer, char *buffer, size_t count)
-{
-  size_t read_count = rbuffer_pending(rbuffer);
-
-  if (count < read_count) {
-    read_count = count;
-  }
-
-  if (read_count > 0) {
-    memcpy(buffer, rbuffer_read_ptr(rbuffer), read_count);
-    rbuffer_consumed(rbuffer, read_count);
-  }
-
-  return read_count;
-}
-
-/// Copies data to `rbuffer` read queue.
-///
-/// @param rbuffer the `RBuffer` instance
-/// @param buffer The buffer containing data to be copied
-/// @param count Number of bytes that should be copied
-/// @return The number of bytes actually copied
-size_t rbuffer_write(RBuffer *rbuffer, char *buffer, size_t count)
-{
-  size_t write_count = rbuffer_available(rbuffer);
-
-  if (count < write_count) {
-    write_count = count;
-  }
-
-  if (write_count > 0) {
-    memcpy(rbuffer_write_ptr(rbuffer), buffer, write_count);
-    rbuffer_produced(rbuffer, write_count);
-  }
-
-  return write_count;
-}
-
-/// Returns a pointer to a raw buffer containing the first byte available for
-/// reading.
-char *rbuffer_read_ptr(RBuffer *rbuffer)
-{
-  return rbuffer->data + rbuffer->rpos;
-}
-
-/// Returns a pointer to a raw buffer containing the first byte available for
-/// write.
-char *rbuffer_write_ptr(RBuffer *rbuffer)
-{
-  return rbuffer->data + rbuffer->wpos;
-}
-
-/// Returns the number of bytes ready for consumption in `rbuffer`
-///
-/// @param rbuffer The `RBuffer` instance
-/// @return The number of bytes ready for consumption
-size_t rbuffer_pending(RBuffer *rbuffer)
-{
-  return rbuffer->wpos - rbuffer->rpos;
-}
-
-/// Returns available space in `rbuffer`
-///
-/// @param rbuffer The `RBuffer` instance
-/// @return The space available in number of bytes
-size_t rbuffer_available(RBuffer *rbuffer)
-{
-  return rbuffer->capacity - rbuffer->wpos;
-}
-
-void rbuffer_free(RBuffer *rbuffer)
-{
-  xfree(rbuffer->data);
-  xfree(rbuffer);
-}
-
 /// Creates a new RStream instance. A RStream encapsulates all the boilerplate
 /// necessary for reading from a libuv stream.
 ///
@@ -177,8 +42,10 @@ void rbuffer_free(RBuffer *rbuffer)
 RStream * rstream_new(rstream_cb cb, RBuffer *buffer, void *data)
 {
   RStream *rv = xmalloc(sizeof(RStream));
+  buffer->data = rv;
+  buffer->full_cb = on_rbuffer_full;
+  buffer->nonfull_cb = on_rbuffer_nonfull;
   rv->buffer = buffer;
-  rv->buffer->rstream = rv;
   rv->fpos = 0;
   rv->data = data;
   rv->cb = cb;
@@ -190,16 +57,14 @@ RStream * rstream_new(rstream_cb cb, RBuffer *buffer, void *data)
   return rv;
 }
 
-/// Returns the read pointer used by the rstream.
-char *rstream_read_ptr(RStream *rstream)
+static void on_rbuffer_full(RBuffer *buf, void *data)
 {
-  return rbuffer_read_ptr(rstream->buffer);
+  rstream_stop(data);
 }
 
-/// Returns the number of bytes before the rstream is full.
-size_t rstream_available(RStream *rstream)
+static void on_rbuffer_nonfull(RBuffer *buf, void *data)
 {
-  return rbuffer_available(rstream->buffer);
+  rstream_start(data);
 }
 
 /// Frees all memory allocated for a RStream instance
@@ -297,37 +162,13 @@ void rstream_stop(RStream *rstream)
   }
 }
 
-/// Returns the number of bytes ready for consumption in `rstream`
-size_t rstream_pending(RStream *rstream)
-{
-  return rbuffer_pending(rstream->buffer);
-}
-
-/// Reads data from a `RStream` instance into a buffer.
-///
-/// @param rstream The `RStream` instance
-/// @param buffer The buffer which will receive the data
-/// @param count Number of bytes that `buffer` can accept
-/// @return The number of bytes copied into `buffer`
-size_t rstream_read(RStream *rstream, char *buffer, size_t count)
-{
-  return rbuffer_read(rstream->buffer, buffer, count);
-}
-
-RBuffer *rstream_buffer(RStream *rstream)
-{
-  return rstream->buffer;
-}
-
 // Callbacks used by libuv
 
 // Called by libuv to allocate memory for reading.
 static void alloc_cb(uv_handle_t *handle, size_t suggested, uv_buf_t *buf)
 {
   RStream *rstream = handle_get_rstream(handle);
-
-  buf->len = rbuffer_available(rstream->buffer);
-  buf->base = rbuffer_write_ptr(rstream->buffer);
+  buf->base = rbuffer_write_ptr(rstream->buffer, &buf->len);
 }
 
 // Callback invoked by libuv after it copies the data into the buffer provided
@@ -341,7 +182,7 @@ static void read_cb(uv_stream_t *stream, ssize_t cnt, const uv_buf_t *buf)
     if (cnt != UV_ENOBUFS
         // cnt == 0 means libuv asked for a buffer and decided it wasn't needed:
         // http://docs.libuv.org/en/latest/stream.html#c.uv_read_start.
-        // 
+        //
         // We don't need to do anything with the RBuffer because the next call
         // to `alloc_cb` will return the same unused pointer(`rbuffer_produced`
         // won't be called)
@@ -351,18 +192,17 @@ static void read_cb(uv_stream_t *stream, ssize_t cnt, const uv_buf_t *buf)
       // Read error or EOF, either way stop the stream and invoke the callback
       // with eof == true
       uv_read_stop(stream);
-      rstream->cb(rstream, rstream->data, true);
+      rstream->cb(rstream, rstream->buffer, rstream->data, true);
     }
     return;
   }
 
   // at this point we're sure that cnt is positive, no error occurred
-  size_t nread = (size_t) cnt;
-
+  size_t nread = (size_t)cnt;
   // Data was already written, so all we need is to update 'wpos' to reflect
   // the space actually used in the buffer.
   rbuffer_produced(rstream->buffer, nread);
-  rstream->cb(rstream, rstream->data, false);
+  rstream->cb(rstream, rstream->buffer, rstream->data, false);
 }
 
 // Called by the by the 'idle' handle to emulate a reading event
@@ -371,8 +211,7 @@ static void fread_idle_cb(uv_idle_t *handle)
   uv_fs_t req;
   RStream *rstream = handle_get_rstream((uv_handle_t *)handle);
 
-  rstream->uvbuf.len = rbuffer_available(rstream->buffer);
-  rstream->uvbuf.base = rbuffer_write_ptr(rstream->buffer);
+  rstream->uvbuf.base = rbuffer_write_ptr(rstream->buffer, &rstream->uvbuf.len);
 
   // the offset argument to uv_fs_read is int64_t, could someone really try
   // to read more than 9 quintillion (9e18) bytes?
@@ -397,7 +236,7 @@ static void fread_idle_cb(uv_idle_t *handle)
 
   if (req.result <= 0) {
     uv_idle_stop(rstream->fread_idle);
-    rstream->cb(rstream, rstream->data, true);
+    rstream->cb(rstream, rstream->buffer, rstream->data, true);
     return;
   }
 
@@ -412,15 +251,3 @@ static void close_cb(uv_handle_t *handle)
   xfree(handle->data);
   xfree(handle);
 }
-
-static void rbuffer_relocate(RBuffer *rbuffer)
-{
-  assert(rbuffer->rpos <= rbuffer->wpos);
-  // Move data ...
-  memmove(
-      rbuffer->data,  // ...to the beginning of the buffer(rpos 0)
-      rbuffer->data + rbuffer->rpos,  // ...From the first unread position
-      rbuffer->wpos - rbuffer->rpos);  // ...By the number of unread bytes
-  rbuffer->wpos -= rbuffer->rpos;
-  rbuffer->rpos = 0;
-}
diff --git a/src/nvim/os/rstream.h b/src/nvim/os/rstream.h
index 713d1e77e6..3e24724573 100644
--- a/src/nvim/os/rstream.h
+++ b/src/nvim/os/rstream.h
@@ -5,7 +5,6 @@
 #include <stdint.h>
 #include <uv.h>
 #include "nvim/os/event_defs.h"
-
 #include "nvim/os/rstream_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/os/rstream_defs.h b/src/nvim/os/rstream_defs.h
index 1d71160963..45dced0b62 100644
--- a/src/nvim/os/rstream_defs.h
+++ b/src/nvim/os/rstream_defs.h
@@ -3,15 +3,18 @@
 
 #include <stdbool.h>
 
-typedef struct rbuffer RBuffer;
+#include "nvim/rbuffer.h"
+
 typedef struct rstream RStream;
 
 /// Type of function called when the RStream receives data
 ///
 /// @param rstream The RStream instance
+/// @param rbuffer The associated RBuffer instance
 /// @param data State associated with the RStream instance
 /// @param eof If the stream reached EOF.
-typedef void (*rstream_cb)(RStream *rstream, void *data, bool eof);
+typedef void (*rstream_cb)(RStream *rstream, RBuffer *buf, void *data,
+    bool eof);
 
 #endif  // NVIM_OS_RSTREAM_DEFS_H
 
diff --git a/src/nvim/os/shell.c b/src/nvim/os/shell.c
index 2de3b1aeed..48174533a6 100644
--- a/src/nvim/os/shell.c
+++ b/src/nvim/os/shell.c
@@ -283,25 +283,28 @@ static void dynamic_buffer_ensure(DynamicBuffer *buf, size_t desired)
   buf->data = xrealloc(buf->data, buf->cap);
 }
 
-static void system_data_cb(RStream *rstream, void *data, bool eof)
+static void system_data_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
   Job *job = data;
-  DynamicBuffer *buf = job_data(job);
+  DynamicBuffer *dbuf = job_data(job);
 
-  size_t nread = rstream_pending(rstream);
-
-  dynamic_buffer_ensure(buf, buf->len + nread + 1);
-  rstream_read(rstream, buf->data + buf->len, nread);
-
-  buf->len += nread;
+  size_t nread = buf->size;
+  dynamic_buffer_ensure(dbuf, dbuf->len + nread + 1);
+  rbuffer_read(buf, dbuf->data + dbuf->len, nread);
+  dbuf->len += nread;
 }
 
-static void out_data_cb(RStream *rstream, void *data, bool eof)
+static void out_data_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
-  RBuffer *rbuffer = rstream_buffer(rstream);
-  size_t written = write_output(rbuffer_read_ptr(rbuffer),
-                                rbuffer_pending(rbuffer), false, eof);
-  rbuffer_consumed(rbuffer, written);
+  RBUFFER_UNTIL_EMPTY(buf, ptr, len) {
+    size_t written = write_output(ptr, len, false,
+        eof && len <= rbuffer_size(buf));
+    if (written) {
+      rbuffer_consumed(buf, written);
+    } else {
+      break;
+    }
+  }
 }
 
 /// Parses a command string into a sequence of words, taking quotes into
diff --git a/src/nvim/rbuffer.c b/src/nvim/rbuffer.c
new file mode 100644
index 0000000000..9cf681585b
--- /dev/null
+++ b/src/nvim/rbuffer.c
@@ -0,0 +1,205 @@
+#include <assert.h>
+#include <stddef.h>
+#include <string.h>
+
+#include "nvim/memory.h"
+#include "nvim/vim.h"
+#include "nvim/rbuffer.h"
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "rbuffer.c.generated.h"
+#endif
+
+/// Creates a new `RBuffer` instance.
+RBuffer *rbuffer_new(size_t capacity)
+  FUNC_ATTR_WARN_UNUSED_RESULT FUNC_ATTR_NONNULL_RET
+{
+  if (!capacity) {
+    capacity = 0xffff;
+  }
+
+  RBuffer *rv = xmalloc(sizeof(RBuffer) + capacity);
+  rv->full_cb = rv->nonfull_cb = NULL;
+  rv->data = NULL;
+  rv->size = 0;
+  rv->write_ptr = rv->read_ptr = rv->start_ptr;
+  rv->end_ptr = rv->start_ptr + capacity;
+  return rv;
+}
+
+void rbuffer_free(RBuffer *buf)
+{
+  xfree(buf);
+}
+
+size_t rbuffer_size(RBuffer *buf) FUNC_ATTR_NONNULL_ALL
+{
+  return buf->size;
+}
+
+size_t rbuffer_capacity(RBuffer *buf) FUNC_ATTR_NONNULL_ALL
+{
+  return (size_t)(buf->end_ptr - buf->start_ptr);
+}
+
+size_t rbuffer_space(RBuffer *buf) FUNC_ATTR_NONNULL_ALL
+{
+  return rbuffer_capacity(buf) - buf->size;
+}
+
+/// Return a pointer to a raw buffer containing the first empty slot available
+/// for writing. The second argument is a pointer to the maximum number of
+/// bytes that could be written.
+///
+/// It is necessary to call this function twice to ensure all empty space was
+/// used. See RBUFFER_UNTIL_FULL for a macro that simplifies this task.
+char *rbuffer_write_ptr(RBuffer *buf, size_t *write_count) FUNC_ATTR_NONNULL_ALL
+{
+  if (buf->size == rbuffer_capacity(buf)) {
+    *write_count = 0;
+    return NULL;
+  }
+
+  if (buf->write_ptr >= buf->read_ptr) {
+    *write_count = (size_t)(buf->end_ptr - buf->write_ptr);
+  } else {
+    *write_count = (size_t)(buf->read_ptr - buf->write_ptr);
+  }
+
+  return buf->write_ptr;
+}
+
+/// Adjust `rbuffer` write pointer to reflect produced data. This is called
+/// automatically by `rbuffer_write`, but when using `rbuffer_write_ptr`
+/// directly, this needs to called after the data was copied to the internal
+/// buffer. The write pointer will be wrapped if required.
+void rbuffer_produced(RBuffer *buf, size_t count) FUNC_ATTR_NONNULL_ALL
+{
+  assert(count && count <= rbuffer_space(buf));
+
+  buf->write_ptr += count;
+  if (buf->write_ptr >= buf->end_ptr) {
+    // wrap around
+    buf->write_ptr -= rbuffer_capacity(buf);
+  }
+
+  buf->size += count;
+  if (buf->full_cb && !rbuffer_space(buf)) {
+    buf->full_cb(buf, buf->data);
+  }
+}
+
+/// Return a pointer to a raw buffer containing the first byte available
+/// for reading. The second argument is a pointer to the maximum number of
+/// bytes that could be read.
+///
+/// It is necessary to call this function twice to ensure all available bytes
+/// were read. See RBUFFER_UNTIL_EMPTY for a macro that simplifies this task.
+char *rbuffer_read_ptr(RBuffer *buf, size_t *read_count) FUNC_ATTR_NONNULL_ALL
+{
+  if (!buf->size) {
+    *read_count = 0;
+    return NULL;
+  }
+
+  if (buf->read_ptr < buf->write_ptr) {
+    *read_count = (size_t)(buf->write_ptr - buf->read_ptr);
+  } else {
+    *read_count = (size_t)(buf->end_ptr - buf->read_ptr);
+  }
+
+  return buf->read_ptr;
+}
+
+/// Adjust `rbuffer` read pointer to reflect consumed data. This is called
+/// automatically by `rbuffer_read`, but when using `rbuffer_read_ptr`
+/// directly, this needs to called after the data was copied from the internal
+/// buffer. The read pointer will be wrapped if required.
+void rbuffer_consumed(RBuffer *buf, size_t count)
+  FUNC_ATTR_NONNULL_ALL
+{
+  assert(count && count <= buf->size);
+
+  buf->read_ptr += count;
+  if (buf->read_ptr >= buf->end_ptr) {
+      buf->read_ptr -= rbuffer_capacity(buf);
+  }
+
+  bool was_full = buf->size == rbuffer_capacity(buf);
+  buf->size -= count;
+  if (buf->nonfull_cb && was_full) {
+    buf->nonfull_cb(buf, buf->data);
+  }
+}
+
+// Higher level functions for copying from/to RBuffer instances and data
+// pointers
+size_t rbuffer_write(RBuffer *buf, char *src, size_t src_size)
+  FUNC_ATTR_NONNULL_ALL
+{
+  size_t size = src_size;
+
+  RBUFFER_UNTIL_FULL(buf, wptr, wcnt) {
+    size_t copy_count = MIN(src_size, wcnt);
+    memcpy(wptr, src, copy_count);
+    rbuffer_produced(buf, copy_count);
+
+    if (!(src_size -= copy_count)) {
+      return size;
+    }
+
+    src += copy_count;
+  }
+
+  return size - src_size;
+}
+
+size_t rbuffer_read(RBuffer *buf, char *dst, size_t dst_size)
+  FUNC_ATTR_NONNULL_ALL
+{
+  size_t size = dst_size;
+
+  RBUFFER_UNTIL_EMPTY(buf, rptr, rcnt) {
+    size_t copy_count = MIN(dst_size, rcnt);
+    memcpy(dst, rptr, copy_count);
+    rbuffer_consumed(buf, copy_count);
+
+    if (!(dst_size -= copy_count)) {
+      return size;
+    }
+
+    dst += copy_count;
+  }
+
+  return size - dst_size;
+}
+
+char *rbuffer_get(RBuffer *buf, size_t index)
+    FUNC_ATTR_NONNULL_ALL FUNC_ATTR_NONNULL_RET
+{
+  assert(index < buf->size);
+  char *rptr = buf->read_ptr + index;
+  if (rptr >= buf->end_ptr) {
+    rptr -= rbuffer_capacity(buf);
+  }
+  return rptr;
+}
+
+int rbuffer_cmp(RBuffer *buf, const char *str, size_t count)
+  FUNC_ATTR_NONNULL_ALL
+{
+  assert(count <= buf->size);
+  size_t rcnt;
+  (void)rbuffer_read_ptr(buf, &rcnt);
+  size_t n = MIN(count, rcnt);
+  int rv = memcmp(str, buf->read_ptr, n);
+  count -= n;
+  size_t remaining = buf->size - rcnt;
+
+  if (rv || !count || !remaining) {
+    return rv;
+  }
+
+  return memcmp(str + n, buf->start_ptr, count);
+}
+
diff --git a/src/nvim/rbuffer.h b/src/nvim/rbuffer.h
new file mode 100644
index 0000000000..b205db0b5a
--- /dev/null
+++ b/src/nvim/rbuffer.h
@@ -0,0 +1,83 @@
+// Ring buffer implementation. This is basically an array that wraps read/write
+// pointers around the memory region. It should be more efficient than the old
+// RBuffer which required memmove() calls to relocate read/write positions.
+//
+// The main purpose of RBuffer is simplify memory management when reading from
+// uv_stream_t instances:
+//
+// - The event loop writes data to a RBuffer, advancing the write pointer
+// - The main loop reads data, advancing the read pointer
+// - If the buffer becomes full(size == capacity) the rstream is temporarily
+//   stopped(automatic backpressure handling)
+//
+// Reference: http://en.wikipedia.org/wiki/Circular_buffer
+#ifndef NVIM_RBUFFER_H
+#define NVIM_RBUFFER_H
+
+#include <stddef.h>
+#include <stdint.h>
+
+// Macros that simplify working with the read/write pointers directly by hiding
+// ring buffer wrap logic. Some examples:
+//
+// - Pass the write pointer to a function(write_data) that incrementally
+//   produces data, returning the number of bytes actually written to the
+//   ring buffer:
+//
+//       RBUFFER_UNTIL_FULL(rbuf, ptr, cnt)
+//         rbuffer_produced(rbuf, write_data(state, ptr, cnt));
+//
+// - Pass the read pointer to a function(read_data) that incrementally
+//   consumes data, returning the number of bytes actually read from the
+//   ring buffer:
+//
+//       RBUFFER_UNTIL_EMPTY(rbuf, ptr, cnt)
+//         rbuffer_consumed(rbuf, read_data(state, ptr, cnt));
+//
+// Note that the rbuffer_{produced,consumed} calls are necessary or these macros
+// create infinite loops
+#define RBUFFER_UNTIL_EMPTY(buf, rptr, rcnt)                                 \
+  for (size_t rcnt = 0, _r = 1; _r; _r = 0)                                  \
+    for (char *rptr = rbuffer_read_ptr(buf, &rcnt);                          \
+         buf->size;                                                          \
+         rptr = rbuffer_read_ptr(buf, &rcnt))
+
+#define RBUFFER_UNTIL_FULL(buf, wptr, wcnt)                                  \
+  for (size_t wcnt = 0, _r = 1; _r; _r = 0)                                  \
+    for (char *wptr = rbuffer_write_ptr(buf, &wcnt);                         \
+         rbuffer_space(buf);                                                 \
+         wptr = rbuffer_write_ptr(buf, &wcnt))
+
+
+// Iteration
+#define RBUFFER_EACH(buf, c, i)                                              \
+  for (size_t i = 0; i < buf->size; i = buf->size)                           \
+      for (char c = 0;                                                       \
+           i < buf->size ? ((int)(c = *rbuffer_get(buf, i))) || 1 : 0;       \
+           i++)
+
+#define RBUFFER_EACH_REVERSE(buf, c, i)                                      \
+  for (size_t i = buf->size; i != SIZE_MAX; i = SIZE_MAX)                    \
+      for (char c = 0;                                                       \
+           i-- > 0 ? ((int)(c = *rbuffer_get(buf, i))) || 1 : 0;             \
+           )
+
+typedef struct rbuffer RBuffer;
+/// Type of function invoked during certain events:
+///   - When the RBuffer switches to the full state
+///   - When the RBuffer switches to the non-full state
+typedef void(*rbuffer_callback)(RBuffer *buf, void *data);
+
+struct rbuffer {
+  rbuffer_callback full_cb, nonfull_cb;
+  void *data;
+  size_t size;
+  char *end_ptr, *read_ptr, *write_ptr;
+  char start_ptr[];
+};
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "rbuffer.h.generated.h"
+#endif
+
+#endif  // NVIM_RBUFFER_H
diff --git a/src/nvim/tui/term_input.inl b/src/nvim/tui/term_input.inl
index d25cbb7ba1..451ac195f2 100644
--- a/src/nvim/tui/term_input.inl
+++ b/src/nvim/tui/term_input.inl
@@ -162,11 +162,10 @@ static void timer_cb(uv_timer_t *handle)
 
 static bool handle_bracketed_paste(TermInput *input)
 {
-  char *ptr = rbuffer_read_ptr(input->read_buffer);
-  size_t len = rbuffer_pending(input->read_buffer);
-  if (len > 5 && (!strncmp(ptr, "\x1b[200~", 6)
-        || !strncmp(ptr, "\x1b[201~", 6))) {
-    bool enable = ptr[4] == '0';
+  if (rbuffer_size(input->read_buffer) > 5 &&
+      (!rbuffer_cmp(input->read_buffer, "\x1b[200~", 6) ||
+       !rbuffer_cmp(input->read_buffer, "\x1b[201~", 6))) {
+    bool enable = *rbuffer_get(input->read_buffer, 4) == '0';
     // Advance past the sequence
     rbuffer_consumed(input->read_buffer, 6);
     if (input->paste_enabled == enable) {
@@ -195,11 +194,11 @@ static bool handle_bracketed_paste(TermInput *input)
 
 static bool handle_forced_escape(TermInput *input)
 {
-  char *ptr = rbuffer_read_ptr(input->read_buffer);
-  size_t len = rbuffer_pending(input->read_buffer);
-  if (len > 1 && ptr[0] == ESC && ptr[1] == NUL) {
+  if (rbuffer_size(input->read_buffer) > 1
+      && !rbuffer_cmp(input->read_buffer, "\x1b\x00", 2)) {
     // skip the ESC and NUL and push one <esc> to the input buffer
-    termkey_push_bytes(input->tk, ptr, 1);
+    size_t rcnt;
+    termkey_push_bytes(input->tk, rbuffer_read_ptr(input->read_buffer, &rcnt), 1);
     rbuffer_consumed(input->read_buffer, 2);
     tk_getkeys(input, true);
     return true;
@@ -207,9 +206,9 @@ static bool handle_forced_escape(TermInput *input)
   return false;
 }
 
-static void read_cb(RStream *rstream, void *rstream_data, bool eof)
+static void read_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
-  TermInput *input = rstream_data;
+  TermInput *input = data;
 
   if (eof) {
     if (input->in_fd == 0 && !os_isatty(0) && os_isatty(2)) {
@@ -236,19 +235,33 @@ static void read_cb(RStream *rstream, void *rstream_data, bool eof)
     if (handle_bracketed_paste(input) || handle_forced_escape(input)) {
       continue;
     }
-    char *ptr = rbuffer_read_ptr(input->read_buffer);
-    size_t len = rbuffer_pending(input->read_buffer);
-    // Find the next 'esc' and push everything up to it(excluding)
-    size_t i;
-    for (i = ptr[0] == ESC ? 1 : 0; i < len; i++) {
-      if (ptr[i] == '\x1b') {
+
+    // Find the next 'esc' and push everything up to it(excluding). This is done
+    // so the `handle_bracketed_paste`/`handle_forced_escape` calls above work
+    // as expected.
+    size_t count = 0;
+    RBUFFER_EACH(input->read_buffer, c, i) {
+      count = i + 1;
+      if (c == '\x1b' && count > 1) {
+        break;
+      }
+    }
+
+    RBUFFER_UNTIL_EMPTY(input->read_buffer, ptr, len) {
+      size_t consumed = termkey_push_bytes(input->tk, ptr, MIN(count, len));
+      // termkey_push_bytes can return (size_t)-1, so it is possible that
+      // `consumed > input->read_buffer->size`, but since tk_getkeys is called
+      // soon, it shouldn't happen
+      assert(consumed <= input->read_buffer->size);
+      rbuffer_consumed(input->read_buffer, consumed);
+      // Need to process the keys now since there's no guarantee "count" will
+      // fit into libtermkey's input buffer.
+      tk_getkeys(input, false);
+      if (!(count -= consumed)) {
         break;
       }
     }
-    size_t consumed = termkey_push_bytes(input->tk, ptr, i);
-    rbuffer_consumed(input->read_buffer, consumed);
-    tk_getkeys(input, false);
-  } while (rbuffer_pending(input->read_buffer));
+  } while (rbuffer_size(input->read_buffer));
 }
 
 static TermInput *term_input_new(void)
diff --git a/test/unit/fixtures/rbuffer.c b/test/unit/fixtures/rbuffer.c
new file mode 100644
index 0000000000..d587d6b054
--- /dev/null
+++ b/test/unit/fixtures/rbuffer.c
@@ -0,0 +1,28 @@
+#include "nvim/rbuffer.h"
+#include "rbuffer.h"
+
+
+void ut_rbuffer_each_read_chunk(RBuffer *buf, each_ptr_cb cb)
+{
+  RBUFFER_UNTIL_EMPTY(buf, rptr, rcnt) {
+    cb(rptr, rcnt);
+    rbuffer_consumed(buf, rcnt);
+  }
+}
+
+void ut_rbuffer_each_write_chunk(RBuffer *buf, each_ptr_cb cb)
+{
+  RBUFFER_UNTIL_FULL(buf, wptr, wcnt) {
+    cb(wptr, wcnt);
+    rbuffer_produced(buf, wcnt);
+  }
+}
+void ut_rbuffer_each(RBuffer *buf, each_cb cb)
+{
+  RBUFFER_EACH(buf, c, i) cb(c, i);
+}
+
+void ut_rbuffer_each_reverse(RBuffer *buf, each_cb cb)
+{
+  RBUFFER_EACH_REVERSE(buf, c, i) cb(c, i);
+}
diff --git a/test/unit/fixtures/rbuffer.h b/test/unit/fixtures/rbuffer.h
new file mode 100644
index 0000000000..640092c627
--- /dev/null
+++ b/test/unit/fixtures/rbuffer.h
@@ -0,0 +1,9 @@
+#include "nvim/rbuffer.h"
+
+typedef void(*each_ptr_cb)(char *ptr, size_t cnt);
+typedef void(*each_cb)(char c, size_t i);
+
+void ut_rbuffer_each_read_chunk(RBuffer *buf, each_ptr_cb cb);
+void ut_rbuffer_each_write_chunk(RBuffer *buf, each_ptr_cb cb);
+void ut_rbuffer_each(RBuffer *buf, each_cb cb);
+void ut_rbuffer_each_reverse(RBuffer *buf, each_cb cb);
diff --git a/test/unit/rbuffer_spec.lua b/test/unit/rbuffer_spec.lua
new file mode 100644
index 0000000000..89136410d3
--- /dev/null
+++ b/test/unit/rbuffer_spec.lua
@@ -0,0 +1,350 @@
+local helpers = require("test.unit.helpers")
+
+local ffi     = helpers.ffi
+local eq      = helpers.eq
+local cstr    = helpers.cstr
+local to_cstr = helpers.to_cstr
+
+local rbuffer = helpers.cimport("./test/unit/fixtures/rbuffer.h")
+
+describe('rbuffer functions', function()
+  local capacity = 16
+  local rbuf
+
+  local function inspect()
+    return ffi.string(rbuf.start_ptr, capacity)
+  end
+
+  local function write(str)
+    local buf = to_cstr(str)
+    return rbuffer.rbuffer_write(rbuf, buf, #str)
+  end
+
+  local function read(len)
+    local buf = cstr(len)
+    len = rbuffer.rbuffer_read(rbuf, buf, len)
+    return ffi.string(buf, len)
+  end
+
+  local function get(idx)
+    return ffi.string(rbuffer.rbuffer_get(rbuf, idx), 1)
+  end
+
+  before_each(function()
+    rbuf = ffi.gc(rbuffer.rbuffer_new(capacity), rbuffer.rbuffer_free)
+    -- fill the internal buffer with the character '0' to simplify inspecting
+    ffi.C.memset(rbuf.start_ptr, string.byte('0'), capacity)
+  end)
+
+  describe('RBUFFER_UNTIL_FULL', function()
+    local chunks
+
+    local function collect_write_chunks()
+      rbuffer.ut_rbuffer_each_write_chunk(rbuf, function(wptr, wcnt)
+        table.insert(chunks, ffi.string(wptr, wcnt))
+      end)
+    end
+
+    before_each(function()
+      chunks = {}
+    end)
+
+    describe('with empty buffer in one contiguous chunk', function()
+      it('is called once with the empty chunk', function()
+        collect_write_chunks()
+        eq({'0000000000000000'}, chunks)
+      end)
+    end)
+
+    describe('with partially empty buffer in one contiguous chunk', function()
+      before_each(function()
+        write('string')
+      end)
+
+      it('is called once with the empty chunk', function()
+        collect_write_chunks()
+        eq({'0000000000'}, chunks)
+      end)
+    end)
+
+    describe('with filled buffer in one contiguous chunk', function()
+      before_each(function()
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is not called', function()
+        collect_write_chunks()
+        eq({}, chunks)
+      end)
+    end)
+
+    describe('with buffer partially empty in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(8)
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_write_chunks()
+        eq({'000000', '12345678'}, chunks)
+      end)
+    end)
+
+    describe('with buffer empty in two contiguous chunks', function()
+      before_each(function()
+        write('12345678')
+        read(8)
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_write_chunks()
+        eq({'00000000', '12345678'}, chunks)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('12345678')
+        read(8)
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is not called', function()
+        collect_write_chunks()
+        eq({}, chunks)
+      end)
+    end)
+  end)
+
+  describe('RBUFFER_UNTIL_EMPTY', function()
+    local chunks
+
+    local function collect_read_chunks()
+      rbuffer.ut_rbuffer_each_read_chunk(rbuf, function(rptr, rcnt)
+        table.insert(chunks, ffi.string(rptr, rcnt))
+      end)
+    end
+
+    before_each(function()
+      chunks = {}
+    end)
+
+    describe('with empty buffer', function()
+      it('is not called', function()
+        collect_read_chunks()
+        eq({}, chunks)
+      end)
+    end)
+
+    describe('with partially filled buffer in one contiguous chunk', function()
+      before_each(function()
+        write('string')
+      end)
+
+      it('is called once with the filled chunk', function()
+        collect_read_chunks()
+        eq({'string'}, chunks)
+      end)
+    end)
+
+    describe('with filled buffer in one contiguous chunk', function()
+      before_each(function()
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is called once with the filled chunk', function()
+        collect_read_chunks()
+        eq({'abcdefghijklmnop'}, chunks)
+      end)
+    end)
+
+    describe('with buffer partially filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_read_chunks()
+        eq({'long s', 'tring'}, chunks)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('12345678')
+        read(8)
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_read_chunks()
+        eq({'abcdefgh', 'ijklmnop'}, chunks)
+      end)
+    end)
+  end)
+
+  describe('RBUFFER_EACH', function()
+    local chars
+
+    local function collect_chars()
+      rbuffer.ut_rbuffer_each(rbuf, function(c, i)
+        table.insert(chars, {string.char(c), tonumber(i)})
+      end)
+    end
+    before_each(function()
+      chars = {}
+    end)
+
+    describe('with empty buffer', function()
+      it('is not called', function()
+        collect_chars()
+        eq({}, chars)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('collects each character and index', function()
+        collect_chars()
+        eq({{'l', 0}, {'o', 1}, {'n', 2}, {'g', 3}, {' ', 4}, {'s', 5},
+            {'t', 6}, {'r', 7}, {'i', 8}, {'n', 9}, {'g', 10}}, chars)
+      end)
+    end)
+  end)
+
+  describe('RBUFFER_EACH_REVERSE', function()
+    local chars
+
+    local function collect_chars()
+      rbuffer.ut_rbuffer_each_reverse(rbuf, function(c, i)
+        table.insert(chars, {string.char(c), tonumber(i)})
+      end)
+    end
+    before_each(function()
+      chars = {}
+    end)
+
+    describe('with empty buffer', function()
+      it('is not called', function()
+        collect_chars()
+        eq({}, chars)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('collects each character and index', function()
+        collect_chars()
+        eq({{'g', 10}, {'n', 9}, {'i', 8}, {'r', 7}, {'t', 6}, {'s', 5},
+            {' ', 4}, {'g', 3}, {'n', 2}, {'o', 1}, {'l', 0}}, chars)
+      end)
+    end)
+  end)
+
+  describe('rbuffer_cmp', function()
+    local function cmp(str)
+      local rv = rbuffer.rbuffer_cmp(rbuf, to_cstr(str), #str)
+      if rv == 0 then
+        return 0
+      else
+        return rv / math.abs(rv)
+      end
+    end
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('compares the common longest sequence', function()
+        eq(0, cmp('long string'))
+        eq(0, cmp('long strin'))
+        eq(-1, cmp('long striM'))
+        eq(1, cmp('long strio'))
+        eq(0, cmp('long'))
+        eq(-1, cmp('lonG'))
+        eq(1, cmp('lonh'))
+      end)
+    end)
+
+    describe('with empty buffer', function()
+      it('returns 0 since no characters are compared', function()
+        eq(0, cmp(''))
+      end)
+    end)
+  end)
+
+  describe('rbuffer_write', function()
+    it('fills the internal buffer and returns the write count', function()
+      eq(12, write('short string'))
+      eq('short string0000', inspect())
+    end)
+
+    it('wont write beyond capacity', function()
+      eq(16, write('very very long string'))
+      eq('very very long s', inspect())
+    end)
+  end)
+
+  describe('rbuffer_read', function()
+    it('reads what was previously written', function()
+      write('to read')
+      eq('to read', read(20))
+    end)
+
+    it('reads nothing if the buffer is empty', function()
+      eq('', read(20))
+      write('empty')
+      eq('empty', read(20))
+      eq('', read(20))
+    end)
+  end)
+
+  describe('rbuffer_get', function()
+    it('fetch the pointer at offset, wrapping if required', function()
+      write('1234567890')
+      read(10)
+      write('long string')
+      eq('l', get(0))
+      eq('o', get(1))
+      eq('n', get(2))
+      eq('g', get(3))
+      eq(' ', get(4))
+      eq('s', get(5))
+      eq('t', get(6))
+      eq('r', get(7))
+      eq('i', get(8))
+      eq('n', get(9))
+      eq('g', get(10))
+    end)
+  end)
+
+  describe('wrapping behavior', function()
+    it('writing/reading wraps across the end of the internal buffer', function()
+      write('1234567890')
+      eq('1234', read(4))
+      eq('5678', read(4))
+      write('987654321')
+      eq('3214567890987654', inspect())
+      eq('90987654321', read(20))
+      eq('', read(4))
+      write('abcdefghijklmnopqrs')
+      eq('nopabcdefghijklm', inspect())
+      eq('abcdefghijklmnop', read(20))
+    end)
+  end)
+end)