Save

praxis3/zmq_distributor.c

@@ -1,15 +1,89 @@
+#include <ctype.h>
 #include <stdio.h>
 #include <stdlib.h>
-#include "zmq.h"
-#include "zmq_worker.h"
+#include <pthread.h>
+#include <stdbool.h>
+#include <string.h>
+#include <regex.h>
+#include <zmq.h>
+#include "transfer_defines.h"
 
+char *get_file_content(FILE *fp)
+{
+    char *buffer = (char*)malloc(sizeof(char) * MESSAGE_LENGTH);
+    if (buffer == NULL)
+    {
+        perror("Failed to allocate memory for buffer");
+        return NULL;
+    }
 
-int main(int argc, char **argv) {
-    if(argc < 2){
+    const size_t bytes_read = fread(buffer, sizeof(char), MESSAGE_LENGTH - 1, fp);
+    if (bytes_read == 0)
+    {
+        perror("File is empty or nothing is left to read");
+        free(buffer);
+        return NULL;
+    }
+
+    buffer[bytes_read] = '\0';
+
+    int delimiter_pos = (int)bytes_read - 1;
+    while (delimiter_pos >= 0 && !isalpha(buffer[delimiter_pos])) { delimiter_pos--; }
+
+    if (delimiter_pos < 0)
+    {
+        perror("No delimiting character was found -> taking whole buffer");
+        delimiter_pos = (int)bytes_read;
+    }
+
+    buffer[delimiter_pos] = '\0';
+
+    if (fseek(fp, delimiter_pos - (int) bytes_read + 1, SEEK_CUR) != 0)
+    {
+        perror("Failed to seek in file");
+        free(buffer);
+        return NULL;
+    }
+
+    return buffer;
+}
+
+pthread_mutex_t words_mutex;
+
+void *receiver()
+{
+
+}
+
+int main(const int argc, char **argv) {
+    if(argc < 3){
         fprintf(stderr, "Usage: %s <file.txt> <worker port 1> <worker port 2> ... <worker port n>", argv[0]);
         exit(1);
     }
 
+    void *context = zmq_ctx_new();
+    void *responder = zmq_socket(context, ZMQ_PULL);
+    int rc = zmq_bind(responder, "tcp://localhost:5555");
+
+    if (rc != 0)
+    {
+        perror("Failed to bind");
+        exit(1);
+    }
+
+    pthread_mutex_init(&words_mutex, NULL);
+
     FILE *fp = fopen(argv[1], "r");
+    char *file_content_chunks;
+    int worker_index = 1;
+    while ((file_content_chunks = get_file_content(fp)) != NULL)
+    {
+        // TODO: send to worker
+
+        free(file_content_chunks);
+    }
+    fclose(fp);
+
+    return 0;
 }