F/delayed handler (StubbornJava#100)

* Add a non blocking delayed handler

Author: billoneil

stubbornjava-common/src/main/java/com/stubbornjava/common/Http.java

@@ -0,0 +1,42 @@
+package com.stubbornjava.common;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.jooq.lambda.Unchecked;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.util.concurrent.MoreExecutors;
+
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+
+public class Http {
+    private static final Logger log = LoggerFactory.getLogger(Http.class);
+
+    // {{start:get}}
+    public static Response get(OkHttpClient client, String url) {
+        Request request = new Request.Builder()
+                .url(url)
+                .get()
+                .build();
+        return Unchecked.supplier(() -> {
+            Response response = client.newCall(request).execute();
+            return response;
+        }).get();
+    }
+    // {{end:get}}
+
+    // {{start:getInParallel}}
+    public static void getInParallel(OkHttpClient client, String url, int count) {
+        ExecutorService exec = Executors.newFixedThreadPool(count);
+        for (int i = 0; i < count; i++) {
+            exec.submit(() -> Http.get(client, url));
+        }
+        MoreExecutors.shutdownAndAwaitTermination(exec, 30, TimeUnit.SECONDS);
+    }
+    // {{end:getInParallel}}
+}

stubbornjava-common/src/main/java/com/stubbornjava/common/HttpClient.java

@@ -40,7 +40,11 @@ private HttpClient() {
             log.debug(msg);
         });
     static {
-        loggingInterceptor.setLevel(Level.BODY);
+        if (log.isDebugEnabled()) {
+            loggingInterceptor.setLevel(Level.BASIC);
+        } else if (log.isTraceEnabled()) {
+            loggingInterceptor.setLevel(Level.BODY);
+        }
     }
 
     public static HttpLoggingInterceptor getLoggingInterceptor() {

stubbornjava-common/src/main/java/com/stubbornjava/common/Timers.java

@@ -0,0 +1,29 @@
+package com.stubbornjava.common;
+
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Stopwatch;
+
+public class Timers {
+    private static final Logger logger = LoggerFactory.getLogger(Timers.class);
+
+    private Timers() {}
+
+    public static void time(String message, Runnable runnable) {
+        Stopwatch sw = Stopwatch.createStarted();
+        try {
+            logger.info("{}", message);
+            runnable.run();
+        } catch (Exception ex) {
+            logger.warn("Exception in runnable", ex);
+            throw ex;
+        } finally {
+            logger.info("{} took {}ms", message, sw.elapsed(TimeUnit.MILLISECONDS));
+        }
+    }
+
+}

stubbornjava-common/src/main/java/com/stubbornjava/common/undertow/SimpleServer.java

@@ -47,6 +47,8 @@ public static SimpleServer simpleServer(HttpHandler handler) {
              * If you base64 encode any cookie values you probably want it on.
              */
             .setServerOption(UndertowOptions.ALLOW_EQUALS_IN_COOKIE_VALUE, true)
+            // Needed to set request time in access logs
+            .setServerOption(UndertowOptions.RECORD_REQUEST_START_TIME, true)
             .addHttpListener(DEFAULT_PORT, DEFAULT_HOST, handler)
         ;
         return new SimpleServer(undertow);

stubbornjava-common/src/main/java/com/stubbornjava/common/undertow/UndertowUtil.java

@@ -0,0 +1,47 @@
+package com.stubbornjava.common.undertow;
+
+import java.net.InetSocketAddress;
+import java.util.function.Consumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.undertow.Undertow;
+import io.undertow.Undertow.ListenerInfo;
+import io.undertow.server.HttpHandler;
+
+public class UndertowUtil {
+    private static final Logger logger = LoggerFactory.getLogger(UndertowUtil.class);
+
+    /**
+     * This is currently intended to be used in unit tests but may
+     * be appropriate in other situations as well. It's not worth building
+     * out a test module at this time so it lives here.
+     *
+     * This helper will spin up the http handler on a random available port.
+     * The full host and port will be passed to the hostConsumer and the server
+     * will be shut down after the consumer completes.
+     *
+     * @param builder
+     * @param handler
+     * @param hostConusmer
+     */
+    public static void useLocalServer(Undertow.Builder builder,
+                                      HttpHandler handler,
+                                      Consumer<String> hostConusmer) {
+        Undertow undertow = null;
+        try  {
+            // Starts server on a random open port
+            undertow = builder.addHttpListener(0, "127.0.0.1", handler).build();
+            undertow.start();
+            ListenerInfo listenerInfo = undertow.getListenerInfo().get(0);
+            InetSocketAddress addr = (InetSocketAddress) listenerInfo.getAddress();
+            String host = "http://localhost:" + addr.getPort();
+            hostConusmer.accept(host);
+        } finally {
+            if (undertow != null) {
+                undertow.stop();
+            }
+        }
+    }
+}

stubbornjava-common/src/main/java/com/stubbornjava/common/undertow/handlers/CustomHandlers.java

@@ -46,7 +46,9 @@ public class CustomHandlers {
     private static final Logger log = LoggerFactory.getLogger(CustomHandlers.class);
 
     public static AccessLogHandler accessLog(HttpHandler next, Logger logger) {
-        return new AccessLogHandler(next, new Slf4jAccessLogReceiver(logger), "combined", CustomHandlers.class.getClassLoader());
+        // see http://undertow.io/javadoc/2.0.x/io/undertow/server/handlers/accesslog/AccessLogHandler.html
+        String format = "%H %h %u \"%r\" %s %Dms %b bytes \"%{i,Referer}\" \"%{i,User-Agent}\"";
+        return new AccessLogHandler(next, new Slf4jAccessLogReceiver(logger), format, CustomHandlers.class.getClassLoader());
     }
 
     public static AccessLogHandler accessLog(HttpHandler next) {

stubbornjava-common/src/main/java/com/stubbornjava/common/undertow/handlers/diagnostic/DelayedExecutionHandler.java

@@ -0,0 +1,52 @@
+package com.stubbornjava.common.undertow.handlers.diagnostic;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import io.undertow.server.Connectors;
+import io.undertow.server.HttpHandler;
+import io.undertow.server.HttpServerExchange;
+import io.undertow.server.handlers.BlockingHandler;
+
+// {{start:delayedHandler}}
+/**
+ * A non blocking handler to add a time delay before the next handler
+ * is executed. If the exchange has already been dispatched this will
+ * un-dispatch the exchange and re-dispatch it before next is called.
+ */
+public class DelayedExecutionHandler implements HttpHandler {
+
+    private final HttpHandler next;
+    private final Function<HttpServerExchange, Duration> durationFunc;
+
+    DelayedExecutionHandler(HttpHandler next,
+                            Function<HttpServerExchange, Duration> durationFunc) {
+        this.next = next;
+        this.durationFunc = durationFunc;
+    }
+
+    @Override
+    public void handleRequest(HttpServerExchange exchange) throws Exception {
+        Duration duration = durationFunc.apply(exchange);
+
+        final HttpHandler delegate;
+        if (exchange.isBlocking()) {
+            // We want to undispatch here so that we are not blocking
+            // a worker thread. We will spin on the IO thread using the
+            // built in executeAfter.
+            exchange.unDispatch();
+            delegate = new BlockingHandler(next);
+        } else {
+            delegate = next;
+        }
+
+        exchange.dispatch(exchange.getIoThread(), () -> {
+            exchange.getIoThread().executeAfter(() ->
+                Connectors.executeRootHandler(delegate, exchange),
+                duration.toMillis(),
+                TimeUnit.MILLISECONDS);
+        });
+    }
+}
+// {{end:delayedHandler}}

stubbornjava-common/src/main/java/com/stubbornjava/common/undertow/handlers/diagnostic/DiagnosticHandlers.java

@@ -0,0 +1,49 @@
+package com.stubbornjava.common.undertow.handlers.diagnostic;
+
+import java.time.Duration;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+import io.undertow.server.HttpHandler;
+
+public class DiagnosticHandlers {
+
+    // {{start:delayedHandler}}
+    /**
+     * Add a fixed delay before execution of the next handler
+     * @param next
+     * @param duration
+     * @param unit
+     * @return
+     */
+    public static DelayedExecutionHandler fixedDelay(HttpHandler next,
+                                                     long duration,
+                                                     TimeUnit unit) {
+        return new DelayedExecutionHandler(
+            next, (exchange) -> Duration.ofMillis(unit.toMillis(duration)));
+    }
+
+    /**
+     * Add a random delay between minDuration (inclusive) and
+     * maxDuration (exclusive) before execution of the next handler.
+     * This can be used to add artificial latency for requests.
+     *
+     * @param next
+     * @param minDuration inclusive
+     * @param maxDuration exclusive
+     * @param unit
+     * @return
+     */
+    public static DelayedExecutionHandler randomDelay(HttpHandler next,
+                                                     long minDuration,
+                                                     long maxDuration,
+                                                     TimeUnit unit) {
+    return new DelayedExecutionHandler(
+        next, (exchange) ->  {
+            long duration = ThreadLocalRandom.current()
+                                             .nextLong(minDuration, maxDuration);
+            return Duration.ofMillis(unit.toMillis(duration));
+        });
+    }
+    // {{end:delayedHandler}}
+}

stubbornjava-common/src/test/java/com/stubbornjava/common/undertow/handlers/diagnostic/DelayedExecutionHandlerTest.java

@@ -0,0 +1,97 @@
+package com.stubbornjava.common.undertow.handlers.diagnostic;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.jooq.lambda.Seq;
+import org.jooq.lambda.Unchecked;
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.stubbornjava.common.Http;
+import com.stubbornjava.common.HttpClient;
+import com.stubbornjava.common.undertow.Exchange;
+import com.stubbornjava.common.undertow.UndertowUtil;
+import com.stubbornjava.common.undertow.handlers.CustomHandlers;
+import com.stubbornjava.undertow.handlers.MiddlewareBuilder;
+
+import io.undertow.Undertow;
+import io.undertow.server.HttpHandler;
+import io.undertow.server.handlers.BlockingHandler;
+import okhttp3.OkHttpClient;
+import okhttp3.Response;
+
+public class DelayedExecutionHandlerTest {
+
+    // Delay for 500ms then return "ok"
+    private static final DelayedExecutionHandler delayedHandler =
+        DiagnosticHandlers.fixedDelay((exchange) -> {
+                Exchange.body().sendText(exchange, "ok");
+            },
+            500, TimeUnit.MILLISECONDS);
+
+    @Test
+    public void testOnXIoThread() throws InterruptedException {
+        int numThreads = 10;
+        run(delayedHandler, numThreads);
+    }
+
+    @Test
+    public void testOnWorkerThread() throws InterruptedException {
+        int numThreads = 10;
+        run(new BlockingHandler(delayedHandler), numThreads);
+    }
+
+    /**
+     * Spin up a new server with a single IO thread and worker thread.
+     * Run N GET requests against it concurrently and make sure they
+     * do not take N * 500ms total. This is not the best test but it
+     * should show that we are delaying N requests at once using a single
+     * thread.
+     *
+     * @param handler
+     * @param numThreads
+     * @throws InterruptedException
+     */
+    private void run(HttpHandler handler, int numThreads) throws InterruptedException {
+        HttpHandler route = MiddlewareBuilder.begin(CustomHandlers::accessLog)
+                                                  .complete(handler);
+        Undertow.Builder builder = Undertow.builder()
+            .setWorkerThreads(1)
+            .setIoThreads(1);
+        UndertowUtil.useLocalServer(builder, route, host -> {
+            ExecutorService exec = Executors.newFixedThreadPool(numThreads);
+            OkHttpClient client = new OkHttpClient().newBuilder()
+                .addInterceptor(HttpClient.getLoggingInterceptor())
+                .build();
+
+            // Using time in tests isn't the best approach but this one seems
+            // A little difficult to test another way.
+            Stopwatch sw = Stopwatch.createStarted();
+            List<Callable<Response>> callables = IntStream.range(0, numThreads)
+                 .mapToObj(i -> (Callable<Response>) () -> Http.get(client, host))
+                 .collect(Collectors.toList());
+            sw.stop();
+            Seq.seq(Unchecked.supplier(() -> exec.invokeAll(callables)).get())
+               .map(Unchecked.function(Future::get))
+               .forEach(DelayedExecutionHandlerTest::assertSuccess);
+            assertTrue("Responses took too long", sw.elapsed().toMillis() < 1_000);
+            MoreExecutors.shutdownAndAwaitTermination(exec, 10, TimeUnit.SECONDS);
+        });
+    }
+
+    private static void assertSuccess(Response response) {
+       Assert.assertTrue("Response should be a 200", response.isSuccessful());
+    }
+
+}