implement job feed

umdiff/Makefile

@@ -1,5 +1,5 @@
 TARGET = umdiff
-C_OBJS = main.c.o rdiff.c.o file.c.o patch.c.o
+C_OBJS = main.c.o compare.c.o file.c.o patch.c.o
 OBJS   = $(C_OBJS)
 
 CMAKE := cmake

umdiff/compare.c

@@ -1,5 +1,5 @@
 /**
- * @file       rdiff.c
+ * @file       compare.c
  * @author     Karim Vergnes <me@thesola.io>
  * @copyright  GPLv2
  * @brief      rsync-based diff calculator
@@ -8,10 +8,19 @@
  * and convert it in-memory into the UMDiff format.
  */
 
-#include "rdiff.h"
+#include "compare.h"
 
 #include <librsync.h>
 
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+
+#define BUFFERS_SIZE 1048576
+
+#define _impl_umdiff_alignBufferSize$(x) \
+    ( x + BUFFERS_SIZE - (x % BUFFERS_SIZE) )
+
 typedef union {
     int fd;
     void *opaque;
@@ -19,28 +28,61 @@ typedef union {
 
 rs_signature_t *workSignatures;
 
+size_t workSigs_counter = 0;
+size_t workSigs_size = 0;
+
 rs_result
 _impl_umdiff_sigJobSink(rs_job_t *job, rs_buffers_t *buf, void *dest)
 {
+    int res;
+    rs_signature_t *sigs = dest;
 
+    if (buf->avail_out + workSigs_counter > workSigs_size) {
+        sigs = realloc(sigs, workSigs_size + _impl_umdiff_alignBufferSize$(buf->avail_out));
+        workSigs_size += _impl_umdiff_alignBufferSize$(buf->avail_out);
+    }
+    memcpy((char *) sigs + workSigs_counter, buf->next_out, buf->avail_out);
+    workSigs_counter += buf->avail_out;
+
+    buf->avail_out = 0;
+
+    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->avail_in = read(fd.fd, buf->next_in, BUFFERS_SIZE);
+    if (!buf->avail_in)
+        buf->eof_in = 1;
+
+    return RS_DONE;
 }
 
 rs_result
-_impl_umdiff_deltaJobSink(rs_job_t *job, rs_buffers_t *buf, void *unk)
+_impl_umdiff_deltaJobSink(rs_job_t *job, rs_buffers_t *buf, void *file_)
 {
+    umdiff_File *file = file_;
 
+    //TODO: parse each command output
+
+    return RS_DONE;
 }
 
 rs_result
 _impl_umdiff_deltaJobSource(rs_job_t *job, rs_buffers_t *buf, void *fd_)
 {
+    int res;
     _impl_umdiff_OpaqueFd fd = { .opaque = fd_ };
+
+    buf->avail_in = read(fd.fd, buf->next_in, BUFFERS_SIZE);
+    if (!buf->avail_in)
+        buf->eof_in = 1;
+
+    return RS_DONE;
 }
 
 
@@ -51,8 +93,17 @@ umdiff_File_fromCompare(int source_fd, int target_fd)
     _impl_umdiff_OpaqueFd target_fd_ = { .fd = target_fd };
     umdiff_File *resultFile;
 
-    rs_buffers_t buffers;
     rs_job_t *sigJob, *deltaJob;
+    rs_buffers_t buffers = {
+        .eof_in = 0,
+        .avail_in = 0,
+        .avail_out = 0,
+        .next_in = malloc(BUFFERS_SIZE),
+        .next_out = malloc(BUFFERS_SIZE)
+    };
+
+    workSignatures = malloc(BUFFERS_SIZE);
+    workSigs_size = BUFFERS_SIZE;
 
     sigJob = rs_sig_begin(ISO_SECTOR_SIZE,
                           RS_DEFAULT_MIN_STRONG_LEN,

umdiff/compare.h

@@ -1,8 +1,8 @@
-#ifndef __RDIFF_H
-#define __RDIFF_H
+#ifndef __COMPARE_H
+#define __COMPARE_H
 
 /**
- * @file       rdiff.h
+ * @file       compare.h
  * @author     Karim Vergnes <me@thesola.io>
  * @copyright  GPLv2
  * @brief      rsync-based diff calculator functions
@@ -24,6 +24,6 @@
 umdiff_File *
 umdiff_File_fromCompare(int source_fd, int target_fd);
 
-#endif //__RDIFF_H
+#endif //__COMPARE_H
 
 // vim: ft=c.doxygen

umdiff/file.c

@@ -83,3 +83,5 @@ umdiff_File_write(umdiff_File *file, int outfd)
 
     return 0;
 }
+
+// vim: ft=c.doxygen

umdiff/main.c

@@ -10,6 +10,7 @@
  */
 
 #include <fcntl.h>
+#include <stdlib.h>
 
 #include "usage.rl.h"
 #include "rdiff.h"
@@ -18,17 +19,15 @@ int
 umdiff_delta(char *source, char *target, char *output)
 {
     int source_fd, target_fd, output_fd;
-    umdiff_File *result;
+    umdiff_File *result = malloc(sizeof(umdiff_File));
 
     source_fd = open(source, O_RDONLY);
     target_fd = open(target, O_RDONLY);
 
     result = umdiff_File_fromCompare(source_fd, target_fd);
 
-    output_fd = open(output, O_WRONLY|O_CREAT|O_TRUNC);
-    umdiff_File_write(result, output_fd);
-
-    return 1;
+    output_fd = open(output, O_WRONLY|O_CREAT|O_TRUNC, 0644);
+    return umdiff_File_write(result, output_fd);
 }
 
 int