70p - with coding style

src/consumer.c

@@ -16,7 +16,7 @@ void consumer_thread(void *args)
 	so_packet_t packet;
 	so_consumer_ctx_t *ctx = (so_consumer_ctx_t *)args;
 
-	while (1) {
+	while (true) {
 		pthread_mutex_lock(&ctx->producer_rb->mutex);
 
 		while (ctx->producer_rb->len == 0 && ctx->producer_rb->stop == 0)
@@ -34,7 +34,7 @@ void consumer_thread(void *args)
 			continue;
 
 		pthread_mutex_lock(&ctx->timestamp);
-		if (packet.hdr.timestamp >= (unsigned int)ctx->lasttimestamp) {
+		if (packet.hdr.timestamp >= ctx->lasttimestamp) {
 			ctx->lasttimestamp = packet.hdr.timestamp;
 
 			pthread_mutex_lock(&ctx->logs);
@@ -58,7 +58,7 @@ int create_consumers(pthread_t *tids,
 	FILE *f = fopen(out_filename, "a");
 
 	if (!f) {
-		perror("Failed to open output file");
+		printf("Failed to open output file\n");
 		return -1;
 	}
 
@@ -89,6 +89,7 @@ int create_consumers(pthread_t *tids,
 			for (int j = 0; j < i; j++)
 				pthread_join(tids[j], NULL);
 			pthread_mutex_destroy(&ctx->logs);
+			pthread_mutex_destroy(&ctx->timestamp);
 			free(ctx);
 			fclose(f);
 			return -1;

src/consumer.h

@@ -15,7 +15,7 @@ typedef struct so_consumer_ctx_t {
 	FILE *f;
 	pthread_mutex_t logs;
 	pthread_mutex_t timestamp;
-	int lasttimestamp;
+	unsigned long lasttimestamp;
 } so_consumer_ctx_t;
 
 int create_consumers(pthread_t *tids,

src/ring_buffer.c

@@ -50,6 +50,7 @@ ssize_t ring_buffer_enqueue(so_ring_buffer_t *ring, void *data, size_t size)
 		memcpy(ring->data + ring->write_pos, data, sizeleft);
 		memcpy(ring->data, (char *)data + sizeleft, size - sizeleft);
 	}
+
 	ring->write_pos = (ring->write_pos + size) % ring->cap;
 	ring->len = ring->len + size;