Implementing rsync data sinks

umdiff/compare.c

@@ -39,28 +39,20 @@ _impl_umdiff_sigJobSink(rs_job_t *job, rs_buffers_t *buf, void *dest)
 {
     int res;
 
+    if (!buf->next_out) {
+        buf->next_out = (char *) workSignatures;
+        buf->avail_out = workSigs_size;
+        return RS_DONE;
+    }
+
     if (BUFFERS_SIZE + workSigs_counter > workSigs_size) {
         workSignatures = realloc(workSignatures, workSigs_size + _impl_umdiff_alignBufferSize$(buf->avail_out));
         workSigs_size += _impl_umdiff_alignBufferSize$(buf->avail_out);
     }
     workSigs_counter += BUFFERS_SIZE;
 
-    buf->avail_out = BUFFERS_SIZE;
-    buf->next_out = dest + workSigs_counter;
-
-    return RS_DONE;
-}
-
-rs_result
-_impl_umdiff_sigJobSource(rs_job_t *job, rs_buffers_t *buf, void *fd_)
-{
-    int res;
-    _impl_umdiff_OpaqueFd fd = { .opaque = fd_ };
-
-    buf->next_in = buf_next_in;
-    buf->avail_in = read(fd.fd, buf->next_in, BUFFERS_SIZE);
-    if (!buf->avail_in)
-        buf->eof_in = 1;
+    buf->avail_out = workSigs_size - workSigs_counter;
+    //buf->next_out = dest + workSigs_counter;
 
     return RS_DONE;
 }
@@ -69,8 +61,10 @@ rs_result
 _impl_umdiff_deltaJobSink(rs_job_t *job, rs_buffers_t *buf, void *file_)
 {
     umdiff_File *file = file_;
+    size_t output_size = buf->next_out - buf_next_out;
 
-    //TODO: parse each command output
+    if (buf->next_out)
+        umdiff_File_feedCommands(file, buf->next_out, output_size);
 
     buf->next_out = buf_next_out;
     buf->avail_out = BUFFERS_SIZE;
@@ -79,15 +73,17 @@ _impl_umdiff_deltaJobSink(rs_job_t *job, rs_buffers_t *buf, void *file_)
 }
 
 rs_result
-_impl_umdiff_deltaJobSource(rs_job_t *job, rs_buffers_t *buf, void *fd_)
+_impl_umdiff_fileJobSource(rs_job_t *job, rs_buffers_t *buf, void *fd_)
 {
     int res;
     _impl_umdiff_OpaqueFd fd = { .opaque = fd_ };
 
     buf->next_in = buf_next_in;
     buf->avail_in = read(fd.fd, buf->next_in, BUFFERS_SIZE);
-    if (!buf->avail_in)
+    if (!buf->avail_in) {
         buf->eof_in = 1;
+        return RS_INPUT_ENDED;
+    }
 
     return RS_DONE;
 }
@@ -104,8 +100,8 @@ umdiff_File_fromCompare(umdiff_File *file, int source_fd, int target_fd)
         .eof_in = 0,
         .avail_in = 0,
         .avail_out = 0,
-        .next_in = buf_next_in,
-        .next_out = buf_next_out
+        .next_in = NULL,
+        .next_out = NULL
     };
 
     workSignatures = malloc(BUFFERS_SIZE);
@@ -117,12 +113,21 @@ umdiff_File_fromCompare(umdiff_File *file, int source_fd, int target_fd)
     deltaJob = rs_delta_begin(workSignatures);
 
     rs_job_drive(sigJob, &buffers,
-                _impl_umdiff_sigJobSource,  source_fd_.opaque,
+                _impl_umdiff_fileJobSource, source_fd_.opaque,
                 _impl_umdiff_sigJobSink,    workSignatures);
+    buffers = (rs_buffers_t) {
+        .eof_in = 0,
+        .avail_in = 0,
+        .avail_out = 0,
+        .next_in = NULL,
+        .next_out = NULL
+    };
     rs_job_drive(deltaJob, &buffers,
-                _impl_umdiff_deltaJobSource, target_fd_.opaque,
+                _impl_umdiff_fileJobSource, target_fd_.opaque,
                 _impl_umdiff_deltaJobSink,  file);
 
+    free(workSignatures);
+
     return 0;
 }
 

umdiff/compare.h

@@ -13,6 +13,19 @@
 
 #include "umdiff.h"
 
+#include <stddef.h>
+
+/**
+ * @brief      Add commands from a rdiff stream.
+ *
+ * This function takes a buffer in the rdiff delta file format, and adds UMDiff
+ * commands matching the parsed rdiff commands to the specified file.
+ *
+ * As it is intended to be called repeatedly, it will statefully alter the
+ * provided @ref umdiff_File object as it goes.
+ */
+int
+umdiff_File_feedCommands(umdiff_File *file, char *buf, size_t len);
 
 /**
  * @brief      Generate a UMDiff file from a source and target file contents.

umdiff/rdiff.c

@@ -8,8 +8,31 @@
  * and manages the corresponding data bucket.
  */
 
+#include "compare.h"
+
+#include <librsync.h>
 #include <prototab.h>
+#include <stddef.h>
 
+int
+umdiff_File_feedCommands(umdiff_File *file, char *buf, size_t len)
+{
+    umdiff_Command *lastCommand;
 
+    //TODO:
+    // 1. skip magic number
+    if (file->hdr.cmd_count == 0 && file->data_len == 0
+            && (int) *buf == RS_DELTA_MAGIC) {
+        buf += sizeof RS_DELTA_MAGIC;
+        len -= sizeof RS_DELTA_MAGIC;
+    }
+
+    if (file->hdr.cmd_count > 0)
+        lastCommand = &file->commands[file->hdr.cmd_count - 1];
+
+    // 2. Update cmd_len on new command
+    // 3. Update data_len otherwise
+    // 4. If data_len < last command's patch area: complete data
+}
 
 // vim: ft=c.doxygen