From 8b5f19f1d43f3d4e8e5df8c9ef2c0ca622bcb3a3 Mon Sep 17 00:00:00 2001
From: Snaipe <franklinmathieu@gmail.com>
Date: Thu, 10 Dec 2015 12:40:49 +0100
Subject: [PATCH] Added nanopb/nanomsg layer

---
 CMakeLists.txt          |  1 +
 src/protocol/protocol.c | 38 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+)
 create mode 100644 src/protocol/protocol.c

diff --git a/CMakeLists.txt b/CMakeLists.txt
index b93c500..b9cfd81 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -209,6 +209,7 @@ set(SOURCE_FILES
   src/entry/entry.c
   src/protocol/criterion.pb.c
   src/protocol/criterion.pb.h
+  src/protocol/protocol.c
   src/common.h
   src/config.h
 
diff --git a/src/protocol/protocol.c b/src/protocol/protocol.c
new file mode 100644
index 0000000..bb7c0ac
--- /dev/null
+++ b/src/protocol/protocol.c
@@ -0,0 +1,38 @@
+#include <pb_encode.h>
+#include <pb_decode.h>
+#include <nanomsg/nn.h>
+#include <nanomsg/pipeline.h>
+
+static bool write_callback(pb_ostream_t *stream, const uint8_t *buf, size_t count) {
+    int sock = (intptr_t) stream->state;
+    int result = nn_send(sock, buf, count, 0);
+
+    if (result < 0)
+        return false;
+    return (size_t) result == count;
+}
+
+static bool read_callback(pb_istream_t *stream, uint8_t *buf, size_t count) {
+    int sock = (intptr_t)stream->state;
+    int result;
+
+    result = nn_recv(sock, buf, count, 0);
+
+    if (result == 0)
+        stream->bytes_left = 0; /* EOF */
+
+    if (result < 0)
+        return false;
+
+    return (size_t) result == count;
+}
+
+pb_ostream_t pb_ostream_from_nn_socket(int sock) {
+    pb_ostream_t stream = {&write_callback, (void*)(intptr_t) sock, SIZE_MAX, 0, ""};
+    return stream;
+}
+
+pb_istream_t pb_istream_from_nn_socket(int sock) {
+    pb_istream_t stream = {&read_callback, (void*)(intptr_t) sock, SIZE_MAX, ""};
+    return stream;
+}