Attempts fixing logic for retryAllFailed() (#46)

* Attempts fixing logic for retryAllFailed() * workQueue to TreeSet * Revert some changes * Readd retryAllFailed but better
teragrep · Aug 22, 2023 · 35b8b37 · 35b8b37
1 parent a050231
commit 35b8b37
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 8 deletions.
diff --git a/src/main/java/com/teragrep/rlp_01/RelpBatch.java b/src/main/java/com/teragrep/rlp_01/RelpBatch.java
@@ -17,10 +17,7 @@
 
 package com.teragrep.rlp_01;
 
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.Set;
-import java.util.TreeMap;
+import java.util.*;
 
 /**
  * A class that is used to send RELP messages to the server.
@@ -37,13 +34,13 @@ public class RelpBatch {
     private TreeMap<Long, RelpFrameRX> responses;
 
     // Not processed queue, for asynchronous use.
-    private LinkedList<Long> workQueue;
+    private TreeSet<Long> workQueue;
 
     public RelpBatch() {
         this.reqID = new requestID();
         this.requests = new TreeMap<Long, RelpFrameTX>();
         this.responses = new TreeMap<Long, RelpFrameRX>();
-        this.workQueue = new LinkedList<Long>();
+        this.workQueue = new TreeSet<>();
     }
 
     /**
@@ -153,7 +150,7 @@ public int getWorkQueueLength() {
     }
 
     public Long popWorkQueue() {
-        return this.workQueue.pop();
+        return this.workQueue.pollFirst();
     }
 }
 
diff --git a/src/test/java/com/teragrep/rlp_01/RelpBatchTest.java b/src/test/java/com/teragrep/rlp_01/RelpBatchTest.java
@@ -126,16 +126,17 @@ public void testVerifyTransactionAll() {
         Assertions.assertTrue(batch.verifyTransactionAll(), "Did not verify all transactions");
     }
 
-    // FIXME: https://github.com/teragrep/rlp_01/issues/12
     @Test
     public void testRetryAllFailed() {
         RelpBatch batch = new RelpBatch();
         int messages = 5;
         Long[] ids = new Long[messages];
+        // Fill 5 messages
         for(int i=0; i<messages; i++) {
             ids[i] = batch.insert(message.getBytes(StandardCharsets.UTF_8));
         }
         Assertions.assertEquals(messages, batch.getWorkQueueLength(), "Worker queue did not match");
+        // Resolve 3 of them
         int resolve = 3;
         for(int i=0; i<resolve; i++) {
             String response = "200 OK";
@@ -144,8 +145,30 @@ public void testRetryAllFailed() {
             buffer.flip();
             batch.putResponse(ids[i], new RelpFrameRX(ids[i].intValue(), RelpCommand.SYSLOG, response.length(), buffer));
         }
+        // Clean work queue to pretend we have sent all
+        int len = batch.getWorkQueueLength();
+        for(int i=0; i<len; i++) {
+            batch.popWorkQueue();
+        }
+        // Refill the work queue as some wasn't sent
         batch.retryAllFailed();
         Assertions.assertEquals(messages-resolve, batch.getWorkQueueLength(), "Worker queue count did not match");
+        // Resolve last messages
+        for(int i=resolve; i<messages; i++) {
+            String response = "200 OK";
+            ByteBuffer buffer = ByteBuffer.allocateDirect(response.length());
+            buffer.put(response.getBytes(StandardCharsets.UTF_8));
+            buffer.flip();
+            batch.putResponse(ids[i], new RelpFrameRX(ids[i].intValue(), RelpCommand.SYSLOG, response.length(), buffer));
+        }
+        // Resolve work queue again
+        len = batch.getWorkQueueLength();
+        for(int i=0; i<len; i++) {
+            batch.popWorkQueue();
+        }
+        // Refill the work queue
+        batch.retryAllFailed();
+        Assertions.assertEquals(0, batch.getWorkQueueLength(), "Worker queue count did not match");
     }
 
     @Test