Fixed double words (for first word the pointer was not moved) - Fixed last char overwritten (\0 was inserted one to early

praxis3/zmq_worker.c

@@ -40,7 +40,7 @@ int main(int argc, char **argv){
         printf("Initialized Thread: %d\n", thread_count);
     }
 
-    printf("Done initialization");
+    printf("Init DONE\n");
 
     // wait for workers to finish
     for(int i = 0; i < thread_count; i++){
@@ -96,6 +96,16 @@ void* workerFunc(void *args){
 
         char request[MESSAGE_LENGTH + 1]; // CHECK: off by one error
         int bytes_recv = zmq_recv(responder, request, MESSAGE_LENGTH, 0);
+        if(bytes_recv <= 0){
+            perror("Failed to receive request");
+            continue;
+        }
+
+        char endpoint[256];
+        size_t endpoint_size = sizeof(endpoint);
+        if (zmq_getsockopt(responder, ZMQ_LAST_ENDPOINT, endpoint, &endpoint_size) == 0) {
+            printf("Got request from: %s\n", endpoint);
+        }
 
         const TYPE type = get_request_type(request);
         char *content = request + TYPE_LENGTH;
@@ -104,15 +114,15 @@ void* workerFunc(void *args){
 
         switch(type){
             case MAP:
-                perror("Handling MAP\n");
+                printf("Handling MAP\n");
                 ret = map(content, &head, &tail);
                 break;
             case RED:
-                perror("Handling RED\n");
+                printf("Handling RED\n");
                 ret = reduce(content);
                 break;
             case RIP:
-                perror("Handling RIP\n");
+                printf("Handling RIP\n");
             // send errors are redundant to handle because exit
                 zmq_send(responder, "rip", TYPE_LENGTH, 0);
                 zmq_close(responder);
@@ -125,8 +135,8 @@ void* workerFunc(void *args){
                 exit(1);
                 break;
         }
-
-        if(zmq_send(responder, ret, MESSAGE_LENGTH, 0) != 0){
+        printf("Sending content: %s\n", ret);
+        if(zmq_send(responder, ret, strlen(ret), 0) != 0){
             perror("Failed to send response");
             zmq_close(responder);
             zmq_ctx_destroy(context);
@@ -158,22 +168,18 @@ char* get_word(const char *request){
     if(request == NULL || *request == '\0')
         return strdup("");
 
-    int capacity = 16;
     int length = 0;
-    char *word = (char*)malloc(capacity);
+    char *word = (char*)malloc(sizeof(char));
     if(!word)
         return NULL;
 
     while(*request && isalpha(*request)){
-        if(length + 1 >= capacity){
-            capacity *= 2;
-            char *new_word = (char*)realloc(word, capacity);
-            if(!new_word){
-                free(word);
-                return NULL;
-            }
-            word = new_word;
+        char *new_word = (char*)realloc(word, strlen(word) + 2);    // realloc word size + \0 and extra char space
+        if(!new_word){
+            free(word);
+            return NULL;
         }
+        word = new_word;
         word[length++] = *request;
         request++;
     }
@@ -193,7 +199,7 @@ char* map(const char *content, key_value_pair **head, key_value_pair **tail){
             if(strcmp(word, "") != 0){
                 add_word(word, head, tail);
             }
-            content_iterator += strlen(word) + 1;
+            content_iterator += strlen(word) + 1;   // skip current read word + one delimiter char
             free(word);
         }
     }
@@ -210,12 +216,13 @@ char* map(const char *content, key_value_pair **head, key_value_pair **tail){
         }
         if(!ret){
             ret = strdup(kvp2str);
+            kvp_it = kvp_it->next;
         }
         else{
             char *tmp = realloc(ret, strlen(ret) + strlen(kvp2str) + 1);
             if(!tmp){
                 free(ret);
-                perror("Failed to realloc memory for ret");
+                perror("Failed to reallocate memory for ret");
                 free(kvp2str);
                 return NULL;
             }