Re #7: Added more tests and some small changes of monitoring and Netty code. Added closing connection if keepalive is not in force.

loki-log4j2-appender/pom.xml

@@ -55,6 +55,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>4.1.17</version>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- Testcontainers -->
         <dependency>
             <groupId>org.junit.jupiter</groupId>

loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java

@@ -68,6 +68,10 @@ public class LokiAppender extends AbstractAppender {
         return loggingSystem;
     }
 
+    /**
+     * This is an entry point to set monitoring (statistics) hooks for this appender. This
+     * API is in beta and is subject to change (and probably will).
+     */
     public void setMonitoringModule(MonitoringModule monitoringModule) {
         monitoringModuleWrapper.setMonitoringModule(monitoringModule);
     }

loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java

@@ -0,0 +1,58 @@
+package pl.tkowalcz.tjahzi.log4j2;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.MetricRegistry;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import pl.tkowalcz.tjahzi.stats.DropwizardMonitoringModule;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@Testcontainers
+class LokiAppenderMonitoringTest {
+
+    @Test
+    void shouldInjectMonitoringAndUseIt() throws URISyntaxException {
+        // Given
+        URI uri = getClass()
+                .getClassLoader()
+                .getResource("basic-appender-test-log4j2-configuration.xml")
+                .toURI();
+
+        LoggerContext context = (LoggerContext) LogManager.getContext(false);
+        context.setConfigLocation(uri);
+
+        LokiAppender loki = context.getConfiguration().getAppender("Loki");
+        MetricRegistry metricRegistry = new MetricRegistry();
+        loki.setMonitoringModule(
+                new DropwizardMonitoringModule(
+                        metricRegistry,
+                        "appender.loki"
+                )
+        );
+
+        // When
+        Logger logger = LogManager.getLogger(LokiAppenderMonitoringTest.class);
+        logger.info("Test test test");
+
+        // Then
+        assertThat(metricRegistry.getMetrics()).isNotEmpty();
+
+        Counter connectAttemptsCounter = metricRegistry
+                .getCounters()
+                .get("appender.loki.httpConnectAttempts");
+        assertThat(connectAttemptsCounter).isNotNull();
+
+        await().untilAsserted(() -> {
+                    assertThat(connectAttemptsCounter.getCount()).isEqualTo(1);
+                }
+        );
+    }
+}

tjahzi-core/pom.xml

@@ -76,9 +76,9 @@
         </dependency>
 
         <dependency>
-            <groupId>io.dropwizard</groupId>
-            <artifactId>dropwizard-core</artifactId>
-            <version>2.0.18</version>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>4.1.17</version>
             <scope>provided</scope>
         </dependency>
 

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java

@@ -1,21 +1,21 @@
 package pl.tkowalcz.tjahzi.http;
 
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
-import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpClientCodec;
 import io.netty.handler.codec.http.HttpContentDecompressor;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.timeout.IdleStateHandler;
 import pl.tkowalcz.tjahzi.stats.MonitoringModule;
 
-class HttpClientInitializer extends ChannelInitializer<SocketChannel> {
+class HttpClientInitializer extends ChannelInitializer<Channel> {
 
     private static final int BYTES_IN_MEGABYTE = 1024 * 1024;
     private static final int MAX_CONTENT_LENGTH = 10 * BYTES_IN_MEGABYTE;
 
     private final MonitoringModule monitoringModule;
-    private final ResponseHandler responseHandler;
+    private final RequestAndResponseHandler responseHandler;
 
     private final int requestTimeoutMillis;
     private final int maxRequestsInFlight;
@@ -26,14 +26,14 @@ class HttpClientInitializer extends ChannelInitializer<SocketChannel> {
             int maxRequestsInFlight
     ) {
         this.monitoringModule = monitoringModule;
-        this.responseHandler = new ResponseHandler(monitoringModule);
+        this.responseHandler = new RequestAndResponseHandler(monitoringModule);
 
         this.requestTimeoutMillis = requestTimeoutMillis;
         this.maxRequestsInFlight = maxRequestsInFlight;
     }
 
     @Override
-    protected void initChannel(SocketChannel ch) {
+    protected void initChannel(Channel ch) {
         ChannelPipeline p = ch.pipeline();
         p.addLast(new IdleStateHandler(
                         requestTimeoutMillis,
@@ -50,7 +50,6 @@ class HttpClientInitializer extends ChannelInitializer<SocketChannel> {
                         maxRequestsInFlight
                 )
         );
-        p.addLast(new InFlightRequestsTracker());
         p.addLast(responseHandler);
     }
 }

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java

@@ -68,7 +68,6 @@ public class HttpConnection implements Closeable {
         stableReference.awaitUninterruptibly();
         if (stableReference.isSuccess() && stableReference.channel().isActive()) {
             stableReference.channel().writeAndFlush(request);
-            monitoringModule.incrementSentHttpRequests(request.content().readableBytes());
         } else {
             retry.retry();
         }

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java

@@ -1,24 +0,0 @@
-package pl.tkowalcz.tjahzi.http;
-
-import io.netty.channel.ChannelDuplexHandler;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelPromise;
-
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class InFlightRequestsTracker extends ChannelDuplexHandler {
-
-    private final AtomicInteger requestsInFlight = new AtomicInteger();
-
-    @Override
-    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
-        super.write(ctx, msg, promise);
-        requestsInFlight.incrementAndGet();
-    }
-
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        super.channelRead(ctx, msg);
-        requestsInFlight.decrementAndGet();
-    }
-}

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java

@@ -31,3 +31,4 @@ public class PipelinedHttpRequestTimer extends ChannelDuplexHandler {
         monitoringModule.recordResponseTime(timer.measure());
     }
 }
+ 

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/RequestAndResponseHandler.java

@@ -1,28 +1,46 @@
 package pl.tkowalcz.tjahzi.http;
 
+import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.FullHttpResponse;
 import io.netty.handler.codec.http.HttpStatusClass;
+import io.netty.handler.codec.http.HttpUtil;
 import pl.tkowalcz.tjahzi.stats.MonitoringModule;
 
 @ChannelHandler.Sharable
-class ResponseHandler extends SimpleChannelInboundHandler<FullHttpResponse> {
+class RequestAndResponseHandler extends ChannelDuplexHandler {
 
     private final MonitoringModule monitoringModule;
 
-    ResponseHandler(MonitoringModule monitoringModule) {
-        super(false);
+    RequestAndResponseHandler(MonitoringModule monitoringModule) {
         this.monitoringModule = monitoringModule;
     }
 
     @Override
-    protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) {
+    public void write(ChannelHandlerContext ctx, Object object, ChannelPromise promise) throws Exception {
+        if (object instanceof FullHttpRequest) {
+            int payloadSize = ((FullHttpRequest) object).content().readableBytes();
+            monitoringModule.incrementSentHttpRequests(payloadSize);
+        }
+
+        super.write(ctx, object, promise);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object object) {
+        FullHttpResponse msg = (FullHttpResponse) object;
+
         monitoringModule.incrementHttpResponses();
         if (msg.status().codeClass() != HttpStatusClass.SUCCESS) {
             monitoringModule.incrementHttpErrors(msg.status(), msg.content());
         }
+
+        if (!HttpUtil.isKeepAlive(msg)) {
+            ctx.close();
+        }
     }
 
     @Override

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java

@@ -58,16 +58,32 @@ public class StandardMonitoringModule implements MonitoringModule {
         sentBytes.addAndGet(sizeBytes);
     }
 
+    public long getSentHttpRequests() {
+        return sentHttpRequests.get();
+    }
+
+    public long getSentBytes() {
+        return sentBytes.get();
+    }
+
     @Override
     public void incrementFailedHttpRequests() {
         failedHttpRequests.incrementAndGet();
     }
 
+    public long getFailedHttpRequests() {
+        return failedHttpRequests.get();
+    }
+
     @Override
     public void incrementRetriedHttpRequests() {
         retriedHttpRequests.incrementAndGet();
     }
 
+    public long getRetriedHttpRequests() {
+        return retriedHttpRequests.get();
+    }
+
     @Override
     public void addAgentError(Throwable throwable) {
         agentErrors.incrementAndGet();
@@ -79,6 +95,10 @@ public class StandardMonitoringModule implements MonitoringModule {
         httpConnectAttempts.incrementAndGet();
     }
 
+    public long getHttpConnectAttempts() {
+        return httpConnectAttempts.get();
+    }
+
     @Override
     public void incrementChannelInactive() {
         channelInactive.incrementAndGet();
@@ -89,6 +109,10 @@ public class StandardMonitoringModule implements MonitoringModule {
         httpResponses.incrementAndGet();
     }
 
+    public long getHttpResponses() {
+        return httpResponses.get();
+    }
+
     @Override
     public void addPipelineError(Throwable cause) {
         distinctErrorLog.record(cause);
@@ -96,6 +120,7 @@ public class StandardMonitoringModule implements MonitoringModule {
 
     @Override
     public void incrementHttpErrors(HttpResponseStatus status, ByteBuf content) {
+        failedHttpRequests.incrementAndGet();
         distinctErrorLog.record(
                 new RuntimeException(
                         content.toString(Charset.defaultCharset())

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java

@@ -6,16 +6,14 @@ import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
 
 class BlockingRetryTest {
 
-    private StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class);
-
     @Test
     void shouldRetryAndInvokeOperation() {
         // Given
         AtomicInteger retryCounter = new AtomicInteger();
+        StandardMonitoringModule monitoringModule = new StandardMonitoringModule();
 
         BlockingRetry retry = new BlockingRetry(
                 __ -> {
@@ -32,12 +30,15 @@ class BlockingRetryTest {
 
         // Then
         assertThat(retryCounter).hasValue(5);
+        assertThat(monitoringModule.getRetriedHttpRequests()).isEqualTo(5);
+        assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(1);
     }
 
     @Test
     void shouldNotRetryOnItsOwn() {
         // Given
         AtomicInteger retryCounter = new AtomicInteger();
+        StandardMonitoringModule monitoringModule = new StandardMonitoringModule();
 
         BlockingRetry retry = new BlockingRetry(
                 __ -> {
@@ -53,5 +54,7 @@ class BlockingRetryTest {
 
         // Then
         assertThat(retryCounter).hasValue(1);
+        assertThat(monitoringModule.getRetriedHttpRequests()).isEqualTo(1);
+        assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(0);
     }
 }

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java

@@ -12,17 +12,16 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
-import static org.mockito.Mockito.mock;
 
 class EventLoopGroupRetryTest {
 
-    private final StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class);
-
     private NioEventLoopGroup eventLoopGroup;
+    private StandardMonitoringModule monitoringModule;
 
     @BeforeEach
     void setUp() {
         eventLoopGroup = new NioEventLoopGroup(Executors.defaultThreadFactory());
+        monitoringModule = new StandardMonitoringModule();
     }
 
     @AfterEach
@@ -34,6 +33,7 @@ class EventLoopGroupRetryTest {
     void shouldRetryAndInvokeOperation() {
         // Given
         AtomicInteger retryCounter = new AtomicInteger();
+        StandardMonitoringModule monitoringModule = new StandardMonitoringModule();
 
         EventLoopGroupRetry retry = new EventLoopGroupRetry(
                 eventLoopGroup,
@@ -52,6 +52,8 @@ class EventLoopGroupRetryTest {
         await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> {
             assertThat(retryCounter).hasValueGreaterThan(3);
         });
+
+        assertThat(monitoringModule.getHttpConnectAttempts()).isGreaterThanOrEqualTo(3);
     }
 
     @Test
@@ -75,5 +77,7 @@ class EventLoopGroupRetryTest {
         await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> {
             assertThat(retryCounter).hasValue(1);
         });
+
+        assertThat(monitoringModule.getHttpConnectAttempts()).isEqualTo(1);
     }
 }

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimerTest.java

@@ -0,0 +1,228 @@
+package pl.tkowalcz.tjahzi.http;
+
+import io.netty.buffer.Unpooled;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http.*;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.shaded.org.apache.commons.lang.mutable.MutableLong;
+import pl.tkowalcz.tjahzi.stats.SettableClock;
+import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule;
+
+import java.time.Clock;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class PipelinedHttpRequestTimerTest {
+
+    @Test
+    void shouldCloseChannelIfNoKeepalive() {
+        // Given
+        SettableClock clock = new SettableClock();
+
+        StandardMonitoringModule monitoringModule = new StandardMonitoringModule() {
+            @Override
+            public Clock getClock() {
+                return clock;
+            }
+        };
+
+        HttpClientInitializer initializer = new HttpClientInitializer(
+                monitoringModule,
+                10_000,
+                42
+        );
+
+        FullHttpRequest request = new DefaultFullHttpRequest(
+                HttpVersion.HTTP_1_1,
+                HttpMethod.POST,
+                ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT,
+                Unpooled.wrappedBuffer("Boo hoo".getBytes())
+        );
+
+        HttpHeaders headers = new DefaultHttpHeaders();
+        headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
+
+        DefaultFullHttpResponse response = new DefaultFullHttpResponse(
+                HttpVersion.HTTP_1_1,
+                HttpResponseStatus.NO_CONTENT,
+                Unpooled.wrappedBuffer("OK".getBytes()),
+                headers,
+                EmptyHttpHeaders.INSTANCE
+        );
+
+        // When
+        EmbeddedChannel channel = new EmbeddedChannel(initializer);
+        channel.writeOneOutbound(request);
+        channel.flush();
+
+        channel.writeOneInbound(response);
+        channel.flush();
+
+        // Then
+        assertThat(channel.isOpen()).isFalse();
+    }
+
+    @Test
+    void shouldNotCloseChannelOnKeepalive() {
+        // Given
+        SettableClock clock = new SettableClock();
+
+        StandardMonitoringModule monitoringModule = new StandardMonitoringModule() {
+            @Override
+            public Clock getClock() {
+                return clock;
+            }
+        };
+
+        HttpClientInitializer initializer = new HttpClientInitializer(
+                monitoringModule,
+                10_000,
+                42
+        );
+
+        FullHttpRequest request = new DefaultFullHttpRequest(
+                HttpVersion.HTTP_1_1,
+                HttpMethod.POST,
+                ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT,
+                Unpooled.wrappedBuffer("Boo hoo".getBytes())
+        );
+
+        HttpHeaders headers = new DefaultHttpHeaders();
+        headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+
+        DefaultFullHttpResponse response = new DefaultFullHttpResponse(
+                HttpVersion.HTTP_1_1,
+                HttpResponseStatus.NO_CONTENT,
+                Unpooled.wrappedBuffer("OK".getBytes()),
+                headers,
+                EmptyHttpHeaders.INSTANCE
+        );
+
+        // When
+        EmbeddedChannel channel = new EmbeddedChannel(initializer);
+        channel.writeOneOutbound(request);
+        channel.flush();
+
+        channel.writeOneInbound(response);
+        channel.flush();
+
+        // Then
+        assertThat(channel.isOpen()).isTrue();
+    }
+
+    @Test
+    void shouldReportRequestMetrics() {
+        // Given
+        SettableClock clock = new SettableClock();
+        MutableLong requestRTT = new MutableLong();
+
+        StandardMonitoringModule monitoringModule = new StandardMonitoringModule() {
+            @Override
+            public Clock getClock() {
+                return clock;
+            }
+
+            @Override
+            public void recordResponseTime(long time) {
+                requestRTT.setValue(time);
+            }
+        };
+
+        HttpClientInitializer initializer = new HttpClientInitializer(
+                monitoringModule,
+                10_000,
+                42
+        );
+
+        FullHttpRequest request = new DefaultFullHttpRequest(
+                HttpVersion.HTTP_1_1,
+                HttpMethod.POST,
+                ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT,
+                Unpooled.wrappedBuffer("Boo hoo".getBytes())
+        );
+
+        HttpHeaders headers = new DefaultHttpHeaders();
+        headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+
+        DefaultFullHttpResponse response = new DefaultFullHttpResponse(
+                HttpVersion.HTTP_1_1,
+                HttpResponseStatus.NO_CONTENT
+        );
+
+        // When
+        EmbeddedChannel channel = new EmbeddedChannel(initializer);
+        channel.writeOneOutbound(request);
+        channel.flush();
+
+        clock.tick();
+
+        channel.writeOneInbound(response);
+        channel.flush();
+
+        // Then
+        assertThat(monitoringModule.getFailedHttpRequests()).isZero();
+        assertThat(monitoringModule.getSentHttpRequests()).isEqualTo(1);
+        assertThat(monitoringModule.getHttpResponses()).isEqualTo(1);
+        assertThat(monitoringModule.getSentBytes()).isEqualTo(7);
+
+        assertThat(requestRTT.intValue()).isEqualTo(SettableClock.TICK_AMOUNT);
+    }
+
+    @Test
+    void shouldReportRequestMetricsOnError() {
+        // Given
+        SettableClock clock = new SettableClock();
+        MutableLong requestRTT = new MutableLong();
+
+        StandardMonitoringModule monitoringModule = new StandardMonitoringModule() {
+            @Override
+            public Clock getClock() {
+                return clock;
+            }
+
+            @Override
+            public void recordResponseTime(long time) {
+                requestRTT.setValue(time);
+            }
+        };
+
+        HttpClientInitializer initializer = new HttpClientInitializer(
+                monitoringModule,
+                10_000,
+                42
+        );
+
+        FullHttpRequest request = new DefaultFullHttpRequest(
+                HttpVersion.HTTP_1_1,
+                HttpMethod.POST,
+                ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT,
+                Unpooled.wrappedBuffer("Boo hoo".getBytes())
+        );
+
+        HttpHeaders headers = new DefaultHttpHeaders();
+        headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+
+        DefaultFullHttpResponse response = new DefaultFullHttpResponse(
+                HttpVersion.HTTP_1_1,
+                HttpResponseStatus.INTERNAL_SERVER_ERROR
+        );
+
+        // When
+        EmbeddedChannel channel = new EmbeddedChannel(initializer);
+        channel.writeOneOutbound(request);
+        channel.flush();
+
+        clock.tick();
+
+        channel.writeOneInbound(response);
+        channel.flush();
+
+        // Then
+        assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(1);
+        assertThat(monitoringModule.getSentHttpRequests()).isEqualTo(1);
+        assertThat(monitoringModule.getHttpResponses()).isEqualTo(1);
+        assertThat(monitoringModule.getSentBytes()).isEqualTo(7);
+
+        assertThat(requestRTT.intValue()).isEqualTo(SettableClock.TICK_AMOUNT);
+    }
+}

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/SettableClock.java

@@ -0,0 +1,44 @@
+package pl.tkowalcz.tjahzi.stats;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
+
+public class SettableClock extends Clock {
+
+    public static final long TICK_AMOUNT = 11L;
+
+    private long millis;
+
+    public void setMillis(long millis) {
+        this.millis = millis;
+    }
+
+    public void tick() {
+        millis += TICK_AMOUNT;
+    }
+
+    public void tick(long amount) {
+        millis += amount;
+    }
+
+    @Override
+    public ZoneId getZone() {
+        return ZoneId.systemDefault();
+    }
+
+    @Override
+    public Clock withZone(ZoneId zone) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Instant instant() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public long millis() {
+        return millis;
+    }
+}

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/TimingRingBufferTest.java

@@ -0,0 +1,120 @@
+package pl.tkowalcz.tjahzi.stats;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Clock;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TimingRingBufferTest {
+
+    @Test
+    void shouldActLikeARingBufferOfTimings() {
+        // Given
+        Clock clock = mock(Clock.class);
+
+        long firstRecording = 10L;
+        long secondRecording = 20L;
+        long thirdRecording = 33L;
+        long firstResponse = 38L;
+        long secondResponse = 41L;
+        long thirdResponse = 56L;
+
+        when(clock.millis())
+                .thenReturn(firstRecording)
+                .thenReturn(secondRecording)
+                .thenReturn(thirdRecording)
+                .thenReturn(firstResponse)
+                .thenReturn(secondResponse)
+                .thenReturn(thirdResponse);
+
+        TimingRingBuffer buffer = new TimingRingBuffer(
+                clock,
+                10
+        );
+
+        // When
+        buffer.record();
+        buffer.record();
+        buffer.record();
+
+        long firstRTT = buffer.measure();
+        long secondRTT = buffer.measure();
+        long thirdRTT = buffer.measure();
+
+        // Then
+        assertThat(firstRTT).isEqualTo(firstResponse - firstRecording);
+        assertThat(secondRTT).isEqualTo(secondResponse - secondRecording);
+        assertThat(thirdRTT).isEqualTo(thirdResponse - thirdRecording);
+    }
+
+    @Test
+    void shouldHandleIntermixedResponsesAndNewRexuests() {
+        // Given
+        Clock clock = mock(Clock.class);
+
+        long firstRecording = 10L;
+        long secondRecording = 20L;
+        long firstResponse = 22L;
+        long thirdRecording = 33L;
+        long secondResponse = 41L;
+        long fourthRecording = 48L;
+        long thirdResponse = 56L;
+        long fourthResponse = 58L;
+
+        when(clock.millis())
+                .thenReturn(firstRecording)
+                .thenReturn(secondRecording)
+                .thenReturn(firstResponse)
+                .thenReturn(thirdRecording)
+                .thenReturn(secondResponse)
+                .thenReturn(fourthRecording)
+                .thenReturn(thirdResponse)
+                .thenReturn(fourthResponse);
+
+        TimingRingBuffer buffer = new TimingRingBuffer(
+                clock,
+                10
+        );
+
+        // When
+        buffer.record();
+        buffer.record();
+        long firstRTT = buffer.measure();
+        buffer.record();
+        long secondRTT = buffer.measure();
+        buffer.record();
+        long thirdRTT = buffer.measure();
+        long fourthRTT = buffer.measure();
+
+        // Then
+        assertThat(firstRTT).isEqualTo(firstResponse - firstRecording);
+        assertThat(secondRTT).isEqualTo(secondResponse - secondRecording);
+        assertThat(thirdRTT).isEqualTo(thirdResponse - thirdRecording);
+        assertThat(fourthRTT).isEqualTo(fourthResponse - fourthRecording);
+    }
+
+    @Test
+    void shouldHandleWrappingAround() {
+        // Given
+        SettableClock clock = new SettableClock();
+
+        TimingRingBuffer buffer = new TimingRingBuffer(
+                clock,
+                10
+        );
+
+        // When & Then
+        for (int i = 0; i < 23; i++) {
+            long tick = 40L;
+
+            buffer.record();
+            clock.tick(tick);
+            long actual = buffer.measure();
+
+            assertThat(actual).isEqualTo(tick);
+        }
+    }
+}