First approach of nexusTransferAPI in C

APIs/nexusTransferAPI.c

+// API to ingest data into a data transfer unit
+
+//#ifndef DATAINGEST_H
+#define DATAINGEST_H
+
+#define version "0.0.1"
+
+#include <zmq.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <string.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <errno.h>
+
+
+typedef int bool;
+#define true 1
+#define false 0
+
+// helper functions for sending and receiving messages
+// source: Pieter Hintjens: ZeroMQ; O'Reilly
+static char* s_recv (void *socket)
+{
+    zmq_msg_t message;
+    zmq_msg_init (&message);
+    int size = zmq_msg_recv (&message, socket, 0);
+    if (size == -1)
+        return NULL;
+    char *string = malloc (size + 1);
+    memcpy (string, zmq_msg_data (&message), size);
+    printf("recieved string: %s\n", string);
+    zmq_msg_close (&message);
+    string [size] = 0;
+    return string;
+};
+
+
+static int s_send (void * socket, char* string)
+{
+    printf("Sending: %s\n", string);
+    int size = zmq_send (socket, string, strlen (string), 0);
+    if (size == -1) {
+        printf("Sending failed\n");
+//        fprintf (stderr, "ERROR: Sending failed\n");
+//        perror("");
+    };
+    return size;
+
+};
+
+
+typedef struct {
+    char *extHost;
+    char *localhost;
+
+    char *signalPort;
+    char *dataPort;
+
+    //  Prepare our context and socket
+    void *context;
+
+    void *signalSocket;
+    void *dataSocket;
+
+    //  Initialize poll set
+//    zmq::pollitem_t items [];
+
+
+    int numberOfStreams;
+//    self.recvdCloseFrom  = []
+    char *replyToSignal;
+    bool allCloseRecvd;
+
+} nexusTransfer;
+
+
+int nexusTransfer_init (nexusTransfer *nT)
+{
+    nT->extHost    = "0.0.0.0";
+    nT->localhost  = "localhost";
+
+    nT->signalPort = "50050";
+    nT->dataPort   = "50100";
+
+    if ( (nT->context = zmq_ctx_new ()) == NULL )
+    {
+		perror("ERROR: Cannot create 0MQ context: ");
+		exit(9);
+	}
+
+	if ( (nT->signalSocket = zmq_socket (nT->context, ZMQ_REP)) == NULL )
+    {
+		perror("ERROR: Could not create 0MQ signalSocket: ");
+		exit(9);
+	}
+
+    if ( (nT->dataSocket = zmq_socket (nT->context, ZMQ_PULL)) == NULL )
+    {
+		perror("ERROR: Could not create 0MQ dataSocket: ");
+		exit(9);
+	}
+
+    nT->replyToSignal = ""
+    nT->allCloseRecvd = false
+
+    char connectionStr[128];
+    int rc;
+
+    // Create sockets
+    snprintf(connectionStr, sizeof(connectionStr), "tcp://%s:%s", nT->extlHost, nT->signalPort);
+//    rc = zmq_connect(nT->signalSocket, connectionStr);
+//    assert (rc == 0);
+    if ( zmq_bind(nT->signalSocket, connectionStr) )
+    {
+        fprintf (stderr, "ERROR: Could not start signalSocket (bind) for '%s'\n", connectionStr);
+        perror("");
+    } else {
+        printf("signalSocket started (bind) for '%s'\n", connectionStr);
+    }
+
+    snprintf(connectionStr, sizeof(connectionStr), "tcp://%s:%s", nT->localhost, nT->dataPort);
+//    rc = zmq_connect(nT->dataSocket, connectionStr);
+//    assert (rc == 0);
+    if ( zmq_bind(nT->dataSocket, connectionStr) )
+    {
+        fprintf (stderr, "ERROR: Could not start dataSocket (bind) for '%s'\n", connectionStr);
+        perror("");
+    } else {
+        printf("dataSocket started (bind) for '%s'\n", connectionStr);
+    }
+
+
+    return 0;
+
+}
+
+
+int nexusTransfer_read (nexusTransfer *nT, char *data, int size)
+{
+
+    zmq_pollitem_t items [] = {
+        { nT->signalSocket,   0, ZMQ_POLLIN, 0 },
+        { nT->dataSocket, 0, ZMQ_POLLIN, 0 }
+    };
+
+    while (1)
+    {
+        printf ("polling");
+        zmq_poll (items, 2, -1);
+
+        if (items [0].revents & ZMQ_POLLIN)
+        {
+            printf ("signalSocket is polling\n");
+
+            data = s_recv (dI->signalSocket);
+            printf ("signalSocket recv: '%s'\n", message);
+
+            if (data == b"CLOSE_FILE" && nT->allCloseRecvd == false):
+                nT->replyToSignal = data;
+            else:
+                rc = zmq_send (nT-signalSocket, data, strlen(data), 0);
+                printf ("signalSocket send: '%s'\n", data);
+
+                return 0;
+        }
+
+        if (items [1].revents & ZMQ_POLLIN) {
+
+            printf ("dataSocket is polling\n");
+            char *multipartMessage[2];
+            int j;
+
+            rc = getMultipartMessage(dI->dataSocket, multipartMessage)
+
+            for (j = 0; j < 2; j++)
+            {
+                free(multipartMessage[j]);
+            };
+        }
+
+};
+
+
+int recv_multipartMessage (void *socket, char **multipartMessage, int *len)
+{
+    int i = 0;
+    int size;
+    int more;
+    size_t more_size = sizeof(more);
+    while (1)
+    {
+
+        //  Wait for next request from client
+        zmq_msg_t message;
+        zmq_msg_init (&message);
+        m_size = zmq_msg_recv (&message, socket, 0);
+
+        //Process message
+        multipartMessage[i] = malloc (m_size + 1);
+        memcpy (multipartMessage[i], zmq_msg_data (&message), m_size);
+        printf("recieved string: %s\n", multipartMessage[i]);
+        multipartMessage[i] [m_size] = 0;
+
+        i++;
+
+        zmq_msg_close(&message);
+        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
+        if (more == 0)
+        {
+            printf("last received\n");
+            *len = i;
+            break;
+        };
+
+    }
+
+    return 0;
+
+
+};
+
+
+
+int getMultipartMessage (void *socket, char *data)
+{
+    char* mulitpartMessage[2];
+    int len;
+    int rc;
+
+    //TODO multipart
+    rc = recv_multipartMessage (dI-dataSocket, multipartMessage, len);
+    printf ("multipartMessage[0]=%s\nmultipartMessage[1]=%s\n", multipartMessage[0], multipartMessage[1]);
+    //self.log.error("Could not receive data due to unknown error.", exc_info=True)
+
+    if (len < 2)
+    {
+        printf ("Received mutipart-message is too short. Either config or file content is missing.");
+//            self.log.debug("multipartMessage=" + str(mutipartMessage))
+        //TODO coorect errorcode
+        return -1;
+
+
+    }
+
+    if (multipartMessage[0] == "CLOSE_FILE")
+    {
+/*
+            id = multipartMessage[1]
+            self.recvdCloseFrom.append(id)
+            self.log.debug("Received close-file-signal from DataDispatcher-" + id)
+
+            # get number of signals to wait for
+            if not self.numberOfStreams:
+                self.numberOfStreams = int(id.split("/")[1])
+
+            # have all signals arrived?
+            if len(self.recvdCloseFrom) == self.numberOfStreams:
+                self.log.info("All close-file-signals arrived")
+                self.allCloseRecvd = True
+                if self.replyToSignal:
+                    self.signalSocket.send(self.replyToSignal)
+                    self.log.debug("signalSocket send: " + self.replyToSignal)
+                    self.replyToSignal = False
+                else:
+                    pass
+
+*/
+        data = "CLOSE_FILE"
+        return 0;
+
+    } else {
+        data = multipartMessage[1];
+        return 0;
+    }
+};
+
+
+int nexusTransfer_stop (dataIngest *nT)
+{
+
+    printf ("closing signalSocket...\n");
+    zmq_close(nT->signalSocket);
+    printf ("closing dataSocket...\n");
+    zmq_close(nT->dataSocket);
+//            self.log.error("closing ZMQ Sockets...failed.", exc_info=True)
+
+    printf ("Closing ZMQ context...\n");
+    zmq_ctx_destroy(nT->context);
+//                self.log.error("Closing ZMQ context...failed.", exc_info=True)
+
+//    free (nT);
+    printf ("Cleanup finished.\n");
+
+    return 0;
+};
+
+//#endif
+
+
+int main ()
+{
+    nexusTransfer *obj;
+    obj = malloc(sizeof(nexusTransfer));
+
+    char *data;
+    int i;
+    int size;
+    int rc;
+
+    rc = nexusTransfer_init (obj);
+
+    rc = nexusTransfer_read (obj, data, size);
+    printf("Read data: %s, size: %d\n", data, size);
+
+    for (i=0; i < 5; i++)
+    {
+        rc = nexusTransfer_read (obj, data, size);
+        printf("Read data: %s, size: %d\n", data, size);
+    };
+
+    rc = nexusTransfer_read (obj, data, size);
+    printf("Read data: %s, size: %d\n", data, size);
+
+    printf ("Stopping\n");
+    rc = nexusTransfer_stop(obj);
+
+    free(obj);
+
+    return 0;
+};
+
+