Re #77: Renamed LogBufferAgent and its thread (ReadingLogBufferAndSendingHttp) to LogShipper. Added option to configure time it sleeps between reading from log buffer to not hog CPU. Also now default is 25ms instead of hardcoded 1ms previously.

core/src/main/java/pl/tkowalcz/tjahzi/LogShipperAgent.java

@@ -6,7 +6,7 @@ import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 
 import java.io.IOException;
 
-public class LogBufferAgent implements Agent {
+public class LogShipperAgent implements Agent {
 
     public static final int MAX_MESSAGES_TO_RETRIEVE = 100;
 
@@ -18,7 +18,7 @@ public class LogBufferAgent implements Agent {
     private final LogBufferMessageHandler messageHandler;
     private final TimeCappedBatchingStrategy batchStrategy;
 
-    public LogBufferAgent(
+    public LogShipperAgent(
             TimeCappedBatchingStrategy batchStrategy,
             ManyToOneRingBuffer logBuffer,
             OutputBuffer outputBuffer,
@@ -69,6 +69,6 @@ public class LogBufferAgent implements Agent {
 
     @Override
     public String roleName() {
-        return "ReadingLogBufferAndSendingHttp";
+        return "LogShipper";
     }
 }

core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java

@@ -23,8 +23,10 @@ public class TjahziInitializer {
             long batchSizeBytes,
             long batchWaitMillis,
             int bufferSizeBytes,
+            long logShipperIntervalMillis,
             long shutdownTimeoutMillis,
-            boolean offHeap) {
+            boolean offHeap
+    ) {
         bufferSizeBytes = findNearestPowerOfTwo(bufferSizeBytes);
         ByteBuffer javaBuffer = allocateJavaBuffer(bufferSizeBytes, offHeap);
 
@@ -33,7 +35,7 @@ public class TjahziInitializer {
         );
 
         OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer());
-        LogBufferAgent agent = new LogBufferAgent(
+        LogShipperAgent agent = new LogShipperAgent(
                 new TimeCappedBatchingStrategy(
                         monitoringModule.getClock(),
                         outputBuffer,
@@ -52,7 +54,7 @@ public class TjahziInitializer {
         );
 
         AgentRunner runner = new AgentRunner(
-                new SleepingMillisIdleStrategy(),
+                new SleepingMillisIdleStrategy(logShipperIntervalMillis),
                 monitoringModule::addAgentError,
                 null,
                 agent

core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java

@@ -89,6 +89,7 @@ class HeadersTest {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -141,6 +142,7 @@ class HeadersTest {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -195,6 +197,7 @@ class HeadersTest {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java

@@ -47,7 +47,7 @@ class LogBufferAgentTest {
         NettyHttpClient httpClient = mock(NettyHttpClient.class);
 
         OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer());
-        LogBufferAgent agent = new LogBufferAgent(
+        LogShipperAgent agent = new LogShipperAgent(
                 new TimeCappedBatchingStrategy(
                         clock,
                         outputBuffer,
@@ -98,7 +98,7 @@ class LogBufferAgentTest {
 
         NettyHttpClient httpClient = mock(NettyHttpClient.class);
         OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer());
-        LogBufferAgent agent = new LogBufferAgent(
+        LogShipperAgent agent = new LogShipperAgent(
                 new TimeCappedBatchingStrategy(
                         clock,
                         outputBuffer,
@@ -137,7 +137,7 @@ class LogBufferAgentTest {
 
         NettyHttpClient httpClient = mock(NettyHttpClient.class);
         OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer());
-        LogBufferAgent agent = new LogBufferAgent(
+        LogShipperAgent agent = new LogShipperAgent(
                 new TimeCappedBatchingStrategy(
                         clock,
                         outputBuffer,
@@ -186,7 +186,7 @@ class LogBufferAgentTest {
         SettableClock clock = new SettableClock();
 
         OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer());
-        LogBufferAgent agent = new LogBufferAgent(
+        LogShipperAgent agent = new LogShipperAgent(
                 new TimeCappedBatchingStrategy(
                         clock,
                         outputBuffer,
@@ -242,7 +242,7 @@ class LogBufferAgentTest {
         SettableClock clock = new SettableClock();
 
         OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer());
-        LogBufferAgent agent = new LogBufferAgent(
+        LogShipperAgent agent = new LogShipperAgent(
                 new TimeCappedBatchingStrategy(
                         clock,
                         outputBuffer,
@@ -260,7 +260,7 @@ class LogBufferAgentTest {
                 )
         );
 
-        for (int i = 0; i < LogBufferAgent.MAX_MESSAGES_TO_RETRIEVE * 2 + 1; i++) {
+        for (int i = 0; i < LogShipperAgent.MAX_MESSAGES_TO_RETRIEVE * 2 + 1; i++) {
             logger.log(
                     42L,
                     LabelSerializerCreator.from(Map.of()),

core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java

@@ -67,6 +67,7 @@ class LoggingSystemSanityCheckTest {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java

@@ -71,6 +71,7 @@ class LoggingSystemTest {
                 1024 * 1024,
                 0,
                 0,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java

@@ -84,6 +84,7 @@ public class ReconnectTest {
                 10_000,
                 1000,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -134,6 +135,7 @@ public class ReconnectTest {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java

@@ -70,6 +70,7 @@ class ResourcesCleanupOnCloseTest {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -83,7 +84,7 @@ class ResourcesCleanupOnCloseTest {
             assertThat(threadInfos)
                     .extracting(ThreadInfo::getThreadName)
                     .contains(
-                            "ReadingLogBufferAndSendingHttp",
+                            "LogShipper",
                             "tjahzi-worker"
                     );
         });
@@ -102,7 +103,7 @@ class ResourcesCleanupOnCloseTest {
             assertThat(threadInfos)
                     .extracting(ThreadInfo::getThreadName)
                     .doesNotContain(
-                            "ReadingLogBufferAndSendingHttp",
+                            "LogShipper",
                             "tjahzi-worker"
                     );
         });