Added c downstream api

lib/downstream_api.c

+#include "nrm.h"
+#include<zmq.h>
+#include<stdio.h>
+#include<unistd.h>
+#include<string.h>
+#include<stdlib.h>
+#include<assert.h>
+
+int nrm_init(struct nrm_context *ctxt, const char *uuid)
+{
+    assert(ctxt != NULL);
+    assert(uuid != NULL);
+    char *uri = getenv(NRM_ENV_URI);
+    if(uri == NULL)
+        uri = NRM_DEFAULT_URI;
+    ctxt->container_uuid = getenv("ARGO_CONTAINER_UUID");
+    assert(ctxt->container_uuid != NULL);
+    ctxt->app_uuid = (char *)uuid;
+    ctxt->context = zmq_ctx_new();
+    ctxt->socket = zmq_socket(ctxt->context, ZMQ_PUB);
+    int err = zmq_connect(ctxt->socket, uri);
+    assert(err == 0);
+    char buf[512];
+    snprintf(buf, 512, NRM_START_FORMAT, ctxt->container_uuid, ctxt->app_uuid);
+    sleep(1);
+    err = zmq_send(ctxt->socket, buf, strnlen(buf, 512), 0);
+    assert(err > 0);
+    assert(!clock_gettime(CLOCK_REALTIME, &ctxt->time));
+    ctxt->acc = 0;
+    return 0;
+}
+
+int nrm_fini(struct nrm_context *ctxt)
+{
+    assert(ctxt != NULL);
+    char buf[512];
+    snprintf(buf, 512, NRM_EXIT_FORMAT, ctxt->app_uuid);
+    int err = zmq_send(ctxt->socket, buf, strnlen(buf, 512), 0);
+    assert(err > 0);
+    zmq_close(ctxt->socket);
+    zmq_ctx_destroy(ctxt->context);
+    return 0;
+}
+
+int nrm_send_progress(struct nrm_context *ctxt, unsigned long progress)
+{
+    char buf[512];
+    struct timespec now;
+    clock_gettime(CLOCK_REALTIME, &now);
+    long long int timediff = (now.tv_nsec - ctxt->time.tv_nsec) +
+                 1e9* (now.tv_sec - ctxt->time.tv_sec);
+    ctxt->acc += progress;
+    if(timediff > NRM_RATELIMIT_THRESHOLD) {
+        snprintf(buf, 512, NRM_PROGRESS_FORMAT, ctxt->acc, ctxt->app_uuid);
+        int err = zmq_send(ctxt->socket, buf, strnlen(buf, 512), 0);
+        assert(err > 0);
+        ctxt->acc = 0;
+    }
+    ctxt->time = now;
+    return 0;
+}

lib/downstream_api.h

+#ifndef NRM_H
+#define NRM_H 1
+
+#include<time.h>
+
+/* min time in nsec between messages: necessary for rate-limiting progress
+ * report. For now, 10ms is the threashold. */
+#define NRM_RATELIMIT_THRESHOLD (10000000LL)
+
+struct nrm_context {
+    void *context;
+    void *socket;
+    char *container_uuid;
+    char *app_uuid;
+    struct timespec time;
+    unsigned long acc;
+};
+
+#define NRM_DEFAULT_URI "ipc:///tmp/nrm-downstream-in"
+#define NRM_ENV_URI "ARGO_NRM_DOWNSTREAM_IN_URI"
+
+#define NRM_START_FORMAT "{\"type\":\"application\", \"event\":\"start\", \"container\": \"%s\", \"uuid\": \"%s\", \"progress\": true, \"threads\": null}"
+#define NRM_PROGRESS_FORMAT "{\"type\":\"application\", \"event\":\"progress\", \"payload\": \"%lu\", \"uuid\": \"%s\"}"
+#define NRM_EXIT_FORMAT "{\"type\":\"application\", \"event\":\"exit\", \"uuid\": \"%s\"}"
+
+int nrm_init(struct nrm_context *, const char *);
+int nrm_fini(struct nrm_context *);
+
+int nrm_send_progress(struct nrm_context *, unsigned long);
+
+#endif