Forem: Luc Engelen

Early termination of transducers and reducing functions

Luc Engelen — Sat, 23 Nov 2024 23:59:59 +0000

In the previous post about transducers, I did not discuss early termination of reducing functions and transducers. For the examples given in that post, early termination was irrelevant. It is, however, an important and tricky aspect of reducing functions and transducers.

Early termination of reducing functions

The following reducing function is a variant on the function first and the transducer (take 1), created specially for people with nostalgia for the 80s.

(defn highlander-rf
  "There can be only one"
  [_ input] (reduced input))

Although this function is largely useless in practice, it provides the most minimalist example of early termination. It takes an intermediate result and an input value, and returns this input value wrapped with a special object. Functions that use this reducing function will recognize this special object and know that they shouldn't provide any more input to this function.

The function reduce is one of those functions that recognize the wrapper object:

(reduce highlander-rf nil [1 2 3 4]) ;; Evaluates to 1

The function reduce will call highlander-rf with nil and 1 as arguments, and then won't call it any more.

This mechanism makes it possible to efficiently reduce large or even infinite collections. Once all relevant input has been processed, all irrelevant input that follows can be ignored.

Using ensure-reduced inside transducers

The following transducer has the same behavior as the reducing function above:

(defn highlander-tr
  "There can be only one"
  [rf]
  (fn
    ([] (rf))
    ([result] (rf result))
    ([result input] (ensure-reduced (rf result input)))))

The two-arity variant of this transducer takes a value as input and returns this value wrapped in the same special object again. It uses ensure-reduced to do this, instead of reduced. When implementing highlander-tr, it's not known what reducing function rf it will be used with. It could be that some of these functions will return a result that is already wrapped. The function ensure-reduced will ensure that an already wrapped result is not wrapped again. If the return value of (rf result input) is already wrapped using reduced, the result will be returned as-is. If it is not wrapped, ensure-reduced will wrap it.

The following example demonstrates the behavior of highlander-tr:

(into []
      highlander-tr
      [1 2 3 4]) ;; Evaluates to [1]

Using reduced? inside transducers

The following reducer does a little more than the one we just saw:

(defn broken-stutter [rf]
  (fn
    ([] (rf))
    ([result] (rf result))
    ([result input]
     (rf (rf result input) input))))

This transducer will output each value it receives as input twice. As its name suggests, it is broken. Unfortunately, that is not that easy to notice.

Evaluating the following example leads to the result you'd expect:

(transduce
 broken-stutter
 conj
 []
 [1 2 3 4 5 6]) ;; Evaluates to [1 1 2 2 3 3 4 4 5 5 6 6]

When used together with the following reducing function that terminates early, however, the transducer will not always behave as expected:

(defn limited-conj [n]
  (fn 
    ([result] result)
    ([result input]
     (if (>= (count result) n)
       (reduced result)
       (conj result input)))))

The reducing function above behaves the same as conj, until the collection passed as intermediate result contains at least n values. Once that threshold has been reached, the intermediate result is returned as final result.

Combining the reducing function limited-conj with the transducer broken-stutter can bring the issue with broken-stutter to the surface. The following expression cannot be evaluated, for example:

(transduce
 broken-stutter
 (limited-conj 6)
 []
 [1 2 3 4 5 6])

The following expression can be evaluated, however, which highlights why dealing with early termination can be tricky:

(transduce
 broken-stutter
 (limited-conj 5)
 []
 [1 2 3 4 5 6]) ;; Evaluates to [1 1 2 2 3]

The problem with broken-stutter is that it may apply the reducing function rf to a result that is already wrapped using reduced. The following variant fixes that issue:

(defn stutter [rf]
  (fn
    ([] (rf))
    ([result] (rf result))
    ([result input]
     (let [intermediate (rf result input)]
       (if (reduced? intermediate)
         intermediate
         (rf intermediate input))))))

This variant checks whether the intermediate result is already reduced and does not apply rf if that is the case.

(transduce
 stutter
 (limited-conj 6)
 []
 [1 2 3 4 5 6]) ;; Evaluates to [1 1 2 2 3 3]

Using unreduced inside transducers

There's one more scenario regarding early termination that I'd like to describe. Because this scenario is also tricky, we need another reducing function that terminates early to demonstrate it. The following function is taken from ClojureDocs:

(defn conj-till-odd
  ([coll] coll)
  ([coll x] (cond-> (conj coll x)
              (odd? x) reduced)))

It behaves as conj until an odd value is received as input.

The following stateful transducer outputs each input value it receives and stores the last one. Once all input has been processed, the last input value is added to the output again.

(defn repeat-last [rf]
  (let [pv (volatile! nil)]
    (fn
      ([] (rf))
      ([result]
       (if-let [p @pv]
         (unreduced (rf result p))
         (rf result)))
      ([result input]
       (let [result (rf result input)]
         (vreset! pv input)
         result)))))

This transducer uses unreduced to ensure that the final result returned by the single-arity variant of this function is not wrapped with reduced. It could be that the result of (rf result p) is wrapped with reduced. If that is the case, this wrapper should be removed because it has no place in the output of the transducer.

If a reducing function or the two-arity variant of a transducer returns a value wrapped with reduced, this wrapper is only intended to signal that the reducing function or transducer will not process any more input values. Whoever is using the reducing function or transducer should stop feeding it more input values and remove the wrapper from the final result. The transducer stutter demonstrates the first action, and the transducer repeat-last demonstrates the second action.

The following example illustrates the use of repeat-last in combination with conj-till-odd:

(transduce
 repeat-last
 conj-till-odd
 []
 [2 4 3 2]) ;; Evaluates to [2 4 3 3]

Conclusion

See https://github.com/ljpengelen/transduce-shakespeare/tree/main/transduce-clj for a Clojure project containing some expressions that illustrate the concepts described in this post.

Error handlers and failure handlers in Vert.x

Luc Engelen — Fri, 22 Nov 2024 23:59:59 +0000

Vert.x is a toolkit for developing reactive applications on the JVM. I wrote a short introductory post about it earlier, when I used it for a commercial project. I had to revisit a Vert.x-based hobby project a few weeks ago, and I learned that there were some gaps in my knowledge about how Vert.x handles failures and errors. To fill those gaps, I did some experiments, wrote a few tests, and then wrote this blog post.

The heart of most Vert.x-based web applications is a router. The router routes requests to zero or more requests handlers, based on the path of the requests. If all goes well, the handler that is handling a given request will issue a response. When something does go wrong, Vert.x offers failure handlers and error handlers to handle the situation.

How to signal that something went wrong in a request handler?

Errors in request handlers come in two flavors: either an exception is thrown (intentionally or unintentionally) or an error is signalled explicitly by calling the fail method on the routing context. If you want to signal something went wrong by calling this method, you have three options:

you can supply a status code,
you can supply a status code and an exception, or
you can supply an exception.

Throwing an exception has the same effect as calling the fail method with an exception as argument. If no status code is provided when calling fail, status code 500 is used. If an exception is provided when calling fail, this exception will be available to all failure and error handlers.

Without any error or failure handler, Vert.x will respond to a failed request with status code 500 and a body containing "Internal Server Error". If that response doesn't suit your needs, you'll need to register an error handler and/or one or more failure handlers.

Error handlers

You can register one error handler per status code with a router. If some failure happens while handling a request and there are no failure handlers (more about those below), then the error handler registered for the status code corresponding to the failure will handle the request:

@Test
void errorHandlerCanHandleException(VertxTestContext vertxTestContext) {
    var handlerExecuted = vertxTestContext.checkpoint();
    var errorHandlerExecuted = vertxTestContext.checkpoint();

    router.route("/")
            .handler(rc -> {
                handlerExecuted.flag();
                throw new RuntimeException(REQUEST_HANDLER_ERROR_MESSAGE);
            });
    router.errorHandler(500, rc -> {
        errorHandlerExecuted.flag();
        rc.response()
                .setStatusCode(500)
                .end(MESSAGE_FROM_ERROR_HANDLER + ": " + rc.failure().getMessage());
    });

    var response = performGetRequest("/");

    assertThat(response.statusCode()).isEqualTo(500);
    assertThat(response.body()).startsWith(MESSAGE_FROM_ERROR_HANDLER);
    assertThat(response.body()).endsWith(REQUEST_HANDLER_ERROR_MESSAGE);
}

As discussed above, the error handler has access to the exception that led to the invocation of the error handler. In this example, the error handler for status code 500 handles the error because this is the default status code when no other status code is provided.

Vert.x supports splitting up a single (large) router into multiple smaller ones using sub routers. Although error handlers can be registered for each sub router, they will simply be ignored:

@Test
void errorHandlerForSubRouterIsIgnored(Vertx vertx, VertxTestContext vertxTestContext) {
    var handlerExecuted = vertxTestContext.checkpoint();
    var rootErrorHandlerExecuted = vertxTestContext.checkpoint();

    var subRouter = Router.router(vertx);
    subRouter.errorHandler(500, rc ->
            vertxTestContext.failNow("Error handler for sub router should not be reached"));
    subRouter.route("/route")
            .handler(rc -> {
                handlerExecuted.flag();
                throw new RuntimeException(REQUEST_HANDLER_ERROR_MESSAGE);
            });

    router.route("/sub/*")
            .subRouter(subRouter);

    router.errorHandler(500, rc -> {
        rootErrorHandlerExecuted.flag();
        rc.response()
                .setStatusCode(500)
                .end(MESSAGE_FROM_ERROR_HANDLER + ": " + rc.failure().getMessage());
    });

    var response = performGetRequest("/sub/route");

    assertThat(response.statusCode()).isEqualTo(500);
    assertThat(response.body()).startsWith(MESSAGE_FROM_ERROR_HANDLER);
    assertThat(response.body()).endsWith(REQUEST_HANDLER_ERROR_MESSAGE);
}

Failure handlers

In some cases, you may want more fine-grained control over how errors are handled. This is where failure handlers come in. One or more failure handlers can be registered per route. They will handle errors in the order in which they are registered, until a handler successfully handles the error or an exception is thrown.

Like error handlers, failure handlers have access to the exception that led to their invocation. They also have access to the status code:

@Test
void failureHandlerCanHandleFailWithStatusCodeAndException(VertxTestContext vertxTestContext) {
    var handlerExecuted = vertxTestContext.checkpoint();
    var failureHandlerExecuted = vertxTestContext.checkpoint();

    router.route("/")
            .handler(rc -> {
                handlerExecuted.flag();
                rc.fail(418, new RuntimeException(REQUEST_HANDLER_ERROR_MESSAGE));
            })
            .failureHandler(rc -> {
                failureHandlerExecuted.flag();
                rc.response()
                        .setStatusCode(rc.statusCode())
                        .end(MESSAGE_FROM_FAILURE_HANDLER + ": " + rc.failure().getMessage());
            });

    var response = performGetRequest("/");

    assertThat(response.statusCode()).isEqualTo(418);
    assertThat(response.body()).startsWith(MESSAGE_FROM_FAILURE_HANDLER);
    assertThat(response.body()).endsWith(REQUEST_HANDLER_ERROR_MESSAGE);
}

Once an failure handler has handled a failure successfully, no error handler will be invoked:

@Test
void errorHandlerIsIgnoredWhenFailureHandlerHandledFailure(VertxTestContext vertxTestContext) {
    var handlerExecuted = vertxTestContext.checkpoint();
    var failureHandlerExecuted = vertxTestContext.checkpoint();

    router.route("/")
            .handler(rc -> {
                handlerExecuted.flag();
                throw new RuntimeException(REQUEST_HANDLER_ERROR_MESSAGE);
            })
            .failureHandler(rc -> {
                failureHandlerExecuted.flag();
                rc.response()
                        .setStatusCode(rc.statusCode())
                        .end(MESSAGE_FROM_FAILURE_HANDLER + ": " + rc.failure().getMessage());
            });
    router.errorHandler(500, rc -> vertxTestContext.failNow("Error should not reach error handler"));

    var response = performGetRequest("/");

    assertThat(response.statusCode()).isEqualTo(500);
    assertThat(response.body()).startsWith(MESSAGE_FROM_FAILURE_HANDLER);
    assertThat(response.body()).endsWith(REQUEST_HANDLER_ERROR_MESSAGE);
}

If a failure handler is unable to handle a certain failure, it can let it be handled by the next failure handler:

@Test
void failureHandlerCanDeferToNextFailureHandler(VertxTestContext vertxTestContext) {
    var handlerExecuted = vertxTestContext.checkpoint();
    var firstFailureHandlerExecuted = vertxTestContext.checkpoint();
    var secondFailureHandlerExecuted = vertxTestContext.checkpoint();

    router.route("/")
            .handler(rc -> {
                handlerExecuted.flag();
                throw new RuntimeException(REQUEST_HANDLER_ERROR_MESSAGE);
            })
            .failureHandler(rc -> {
                firstFailureHandlerExecuted.flag();
                rc.next();
            })
            .failureHandler(rc -> {
                secondFailureHandlerExecuted.flag();
                rc.response()
                        .setStatusCode(rc.statusCode())
                        .end(MESSAGE_FROM_FAILURE_HANDLER + ": " + rc.failure().getMessage());
            });

    var response = performGetRequest("/");

    assertThat(response.statusCode()).isEqualTo(500);
    assertThat(response.body()).startsWith(MESSAGE_FROM_FAILURE_HANDLER);
    assertThat(response.body()).endsWith(REQUEST_HANDLER_ERROR_MESSAGE);
}

If handling a failure leads to an exception, the handling of the original failure is taken over by the error handler:

@Test
void exceptionInFailureHandlerIsIgnoredByErrorHandler(VertxTestContext vertxTestContext) {
    var handlerExecuted = vertxTestContext.checkpoint();
    var failureHandlerExecuted = vertxTestContext.checkpoint();
    var errorHandlerExecuted = vertxTestContext.checkpoint();

    router.route("/")
            .handler(rc -> {
                handlerExecuted.flag();
                throw new RuntimeException(REQUEST_HANDLER_ERROR_MESSAGE);
            })
            .failureHandler(rc -> {
                failureHandlerExecuted.flag();
                throw new RuntimeException(FAILURE_HANDLER_ERROR_MESSAGE);
            });

    router.errorHandler(500, rc -> {
        errorHandlerExecuted.flag();
        rc.response()
                .setStatusCode(500)
                .end(MESSAGE_FROM_ERROR_HANDLER + ": " + rc.failure().getMessage());
    });

    var response = performGetRequest("/");

    assertThat(response.statusCode()).isEqualTo(500);
    assertThat(response.body()).startsWith(MESSAGE_FROM_ERROR_HANDLER);
    assertThat(response.body()).endsWith(REQUEST_HANDLER_ERROR_MESSAGE);
}

If there is no error handler registered for status code 500, an exception thrown in a failure handler will lead to an internal server error.

We saw above that error handlers registered on sub router are ignored. Failure handlers registered for routes on a sub router function as expected, however. The failure handler registered for one of the routes of a sub router can either return a response itself or fall back to the failure handler of another matching route:

@Test
void failureHandlerForSubRouterCanFallBackToFailureHandlerForRoot(Vertx vertx, VertxTestContext vertxTestContext) {
    var handlerExecuted = vertxTestContext.checkpoint();
    var rootFailureHandlerExecuted = vertxTestContext.checkpoint();
    var subFailureHandlerExecuted = vertxTestContext.checkpoint();

    var subRouter = Router.router(vertx);
    subRouter.route("/route")
            .handler(rc -> {
                handlerExecuted.flag();
                throw new RuntimeException(REQUEST_HANDLER_ERROR_MESSAGE);
            })
            .failureHandler(rc -> {
                subFailureHandlerExecuted.flag();
                rc.next();
            });

    router.route("/sub/*")
            .subRouter(subRouter);

    router.route()
            .failureHandler(rc -> {
                rootFailureHandlerExecuted.flag();
                rc.response()
                        .setStatusCode(500)
                        .end(MESSAGE_FROM_FAILURE_HANDLER + ": " + rc.failure().getMessage());
            });

    var response = performGetRequest("/sub/route");

    assertThat(response.statusCode()).isEqualTo(500);
    assertThat(response.body()).startsWith(MESSAGE_FROM_FAILURE_HANDLER);
    assertThat(response.body()).endsWith(REQUEST_HANDLER_ERROR_MESSAGE);
}

Conclusion

As we've seen, error handlers are pretty straightforward. There can be only one error handler per status code, practically speaking, and this handler will handle each error for the given status code if that error has not been handled otherwise.

There's a little more to say about failure handlers. There can be multiple error handlers per route, which will handle errors in the order in which the handlers are registered. In case of overlapping routes (multiple routes that match the path of a given request), the failure handlers for each of these routes are invoked in the order in which the routes are registered. Each failure handler can decide to let the next failure handler handle an error.

I hope this post provides a useful addition to Vert.x's official documentation. If you want to experiment a little yourself, clone and browse https://github.com/ljpengelen/vertx-error-and-failure-handlers for some inspiration and a nice starting point.

To transduce or not to transduce?

Luc Engelen — Wed, 20 Nov 2024 23:59:59 +0000

Transducers are to Clojure what monads are to Haskell: an almost endless source of inspiration for blog posts and discussions. I've heard and read about transducer in the past, but never got around to taking a real close look for myself. Now that I did take a closer look, I also learned some interesting things about Java's volatile keyword, the Java memory model, and the Java Concurrency Stress tests.

Lazy seqs

Before we look into transducers, let's talk about lazy seqs first. Consider the following expression:

(->> (range)
     (filter even?)
     (drop 5)
     (take 5))

This expression can be read as follows:

start with the (infinite) seq of all natural numbers,
keep only the even numbers,
drop the first five, and
take the following five.

The end result of the expression is a lazy seq containing the numbers 10, 12, 14, 16, and 18. Lazy evaluation is part of the foundation of Clojure. Without lazy evaluation, evaluating expressions like the one above would be impossible. You cannot eagerly construct the set of all natural numbers, because that would take an infinite amount of time and storage space.

Although lazy evaluation is what makes expressions like the one above possible, it also has its downsides. One downside is that each step in the transformation pipeline above leads to an intermediate lazy seq. They are created when calculating the end result and then immediately discarded.

Once the value of a certain element of a lazy seq has been calculated, this value is cached. When working with large lazy seqs while keeping a reference to the final result (either deliberately or by accident), memory use can be significant.

Consider the following expression, for example:

(let [r (range 3e7)]
    [(last r) (first r)])

This expression takes the numbers 0 to 30 million minus one and returns a vector containing the last and first of those numbers. After evaluating (last r), all of these numbers are cached in memory, which takes a large amount of space. Since all but the first number are never used again, this space is wasted. We'll come back to this example below.

Because of these downsides (and more), some say that lazy evaluation should be avoided as much as possible. One way of avoiding lazy evaluation as much as possible would be to default to using transducers.

A first look at transducers

The following example shows an alternative expression for calculating the 6th to 10th even number:

(into []
      (comp
       (filter even?)
       (drop 5)
       (take 5))
      (range))

This example uses a transducer, constructed by composing three other transducers, and returns a vector containing the numbers 10, 12, 14, 16, and 18. This end result is computed eagerly, and no intermediate lazy seqs are created.

Although there are some syntactical similarities between this expression and the one at the start of this post, something completely different is going on. Both this example and the one at the start of this post contain the expressions (filter even?), (drop 5), and (take 5). However, after macro expansion, these syntactical similarities are gone:

(clojure.core/take 5
    (clojure.core/drop 5
        (clojure.core/filter clojure.core/even? (clojure.core/range))))

Once the threading macro ->> is out of the picture, it is clear that the functions filter, drop, and take operate on seqs in the first example. In the example demonstrating transducers, (filter even?), (drop 5), and (take 5) each return a transducer.

It is not hard to imagine what (filter even? [1 2 3]) returns. Imagining what (filter even?) could return and how that could be put to good use is a bit more difficult.

Transducers in terms of reducing functions

A reducing function is a function that takes an intermediate result and a new input, and returns a new result. For example, + is a reducing function that takes an intermediate sum and a new number and returns a new sum. The function conj is also a reducing function, which takes an intermediate collection and a new value and returns a new collection that includes the new value. Generally speaking, reducing functions are used to construct a single value from a number of values, one step at a time. They're used to reduce multiple values into a single value.

Depending on how they're supposed to be used, some reducing functions in Clojure also have to support taking no arguments. For example, because (+) evaluates to 0, (reduce + []) can be evaluated too, and will evaluate to 0.

Clojure's documentation describes transducers as a transformation from one reducing function into another. For that statement to be 100% correct, an additional requirement for reducing functions is needed. Apart from accepting no arguments or two arguments, they should also accept a single argument.

Consider the following function, which is a slimmed down variant of Clojure's filter function that takes a predicate and returns a transducer:

(defn filter [pred]
  (fn [rf]
    (fn
      ([] (rf))
      ([result] (rf result))
      ([result input]
        (if (pred input)
          (rf result input)
          result)))))

You'll notice that the transducer returned by filter is a function that takes a reducing function rf and returns a function that takes either 0, 1, or 2 arguments. You'll also notice that the reducing function rf itself is called with 0, 1, or 2 arguments. The most important thing to note, however, is that this transducer will work as expected, regardless of what function is provided as reducing function. The following examples illustrate this.

(reduce ((filter even?) +) 0 [1 2 3 4 5 6]) ;; evaluates to 12
(reduce ((filter even?) str) "" [1 2 3 4 5 6]) ;; evaluates to "246"
(reduce ((filter even?) *) 1 [1 2 3 4 5 6]) ;; evaluates to 48

These examples are only provided to illustrate that any reducing function can be used in combination with a transducer. This is not how you'd use transducers in practice.

Because transducers are simply functions that transform reducing functions into reducing functions, they can be composed with comp, as we've seen in one of the examples above.

The fact that transducers do not care at all which reducing function they're wrapping is exactly the reason why they were added to the language. Whereas the traditional implementations of functions like map and filter operate on collections and return collections, transducers are much more widely applicable. They can be used to implement a variety of processes that take input one value at a time, perform some operation on each of these values, and combine the result somehow.

Creating a stateful transducer

The transducer returned by the function filter we looked at earlier was stateless. It processes input value by value, without maintaining any state concerning previous values. To get a feel for stateful transducers, I created drop-nth, the twin brother of take-nth.

(defn drop-nth [n]
  (fn [rf]
    (let [nv (volatile! -1)]
      (fn
        ([] (rf))
        ([result] (rf result))
        ([result input]
         (let [i (vswap! nv inc)]
           (if (zero? (rem i n))
             result
             (rf result input))))))))

The function drop-nth takes a number n and returns a transducer that leaves out every nth value it receives as input from the output. If this transducer is called without arguments, there's nothing for it to do, so it calls the reducing function without arguments. If this transducer is called with a single argument, there's also nothing for it to do, so it calls the reducing function with the single argument. If the transducer is called with two arguments, it checks its local state to see whether or not the new input value should be included in the result. This is where it gets interesting.

The contract for transducers says that a transducer may be invoked by different threads, but not at the same time. A given transducer could be used to process some values on one thread for some time and then later to process some other values on another thread. Each of these threads should see the current value of nv, the local state of the transducer. This is where volatile and vswap! come into play.

Usually, atoms are used to share state between threads in Clojure. However, to keep transducers as performant as possible, volatiles where introduced for state kept by transducers. The JVM will ensure that the value of a volatile field is always read from main memory and not from the cache maintained by a thread. A volatile does not provide the atomicity guarantees that an atom provides, but that is acceptable given the contract for transducers mentioned above.

Volatile

The following Java application demonstrates the effect of the volatile keyword:

public class VolatileDemo {

    private static volatile boolean STOP_RUNNING_VOLATILE;
    private static boolean STOP_RUNNING_NON_VOLATILE;

    public static void main(String[] args) throws InterruptedException {
        try (var executorService = Executors.newCachedThreadPool()) {
            executorService.submit(() -> {
                var count = 0;
                while (!STOP_RUNNING_VOLATILE) {
                    count++;
                }

                System.out.println("Runnable checking volatile field terminated: " + count);
            });
            executorService.submit(() -> {
                var count = 0;
                while (!STOP_RUNNING_NON_VOLATILE) {
                    count++;
                }

                System.out.println("Runnable checking non-volatile field terminated: " + count);
            });
            Thread.sleep(10);
            STOP_RUNNING_VOLATILE = true;
            STOP_RUNNING_NON_VOLATILE = true;
        }
    }
}

The first runnable submitted to the executor service will stop increasing the counter as soon as the value of STOP_RUNNING_VOLATILE is changed to true. The second runnable will keep on increasing its counter because it's reading a cached value of STOP_RUNNING_NON_VOLATILE.

Another stateful transducer

The transducer drop-nth had nothing to do once the end of its input was reached, but the following transducer does:

(defn strings-to-the-back [rf]
  (let [stringsv (volatile! (java.util.ArrayList.))]
    (fn
      ([] (rf))
      ([result]
       (let [^java.util.ArrayList strings @stringsv
             result (if (.isEmpty strings)
                      result
                      (let [v (vec strings)]
                        (.clear strings)
                        (vreset! stringsv strings)
                        (reduce rf result v)))]
         (rf result)))
      ([result input]
       (let [^java.util.ArrayList strings @stringsv]
         (if (string? input)
           (do
             (.add strings input)
             (vreset! stringsv strings)
             result)
           (rf result input)))))))

This transducer inspects the values it receives as input and will not immediately add them to the output if they are strings. One the end of the input is reached, all strings are added to the output.

This transducer is inspired by partition-all, which looks like this at the time of writing:

(defn partition-all [^long n]
  (fn [rf]
    (let [a (java.util.ArrayList. n)]
      (fn
        ([] (rf))
        ([result]
         (let [result (if (.isEmpty a)
                        result
                        (let [v (vec (.toArray a))]
                          ;;clear first!
                          (.clear a)
                          (unreduced (rf result v))))]
           (rf result)))
        ([result input]
         (.add a input)
         (if (= n (.size a))
           (let [v (vec (.toArray a))]
             (.clear a)
             (rf result v))
           result))))))

The most notable difference between strings-to-the-back and partition-all is that the latter does not make use of a volatile. This is, however, a bug: https://clojure.atlassian.net/browse/CLJ-2146. Another difference is that the array list stored as state is converted to a vector as follows: (vec (.toArray a)). After some benchmarking, I found out that this is slightly faster than (vec a) for small lists. I don't see why this only holds for small lists, but I don't want to invest time in finding out right now.

After each update of the array list containing strings in strings-to-the-back, you'll see (vreset! stringsv strings). This may seem unnecessary, since strings is always the same object. This expression does have an effect, however. The Java memory model guarantees that when a thread reads a volatile variable, it sees not just the latest change to the volatile, but also the side effects of the code that led up to the change.

Volatile fields and visibility of related changes

There is a set of stress tests called the Java Concurrency Stress tests (jcstress) that can be used to find concurrency-related bugs in implementations of the JVM, among other things.

Running the following stress test shows that the JVM behaves exactly as documented. The observer will either see a null list or one that contains the number 42, because the assignment to the volatile field happens after the number 42 is added to the temporary list.

@JCStressTest
@State
@Outcome.Outcomes({
        @Outcome(id = "-1", expect = ACCEPTABLE, desc = "Null list"),
        @Outcome(id = "0", expect = FORBIDDEN, desc = "Empty list"),
        @Outcome(id = "42", expect = ACCEPTABLE, desc = "List containing 42"),
})
public class VolatileSaveAfterModification {

    volatile List<Integer> list;

    @Actor
    public void actor() {
        var tmpList = new ArrayList<Integer>();
        tmpList.add(42);
        list = tmpList;
    }

    @Actor
    public void observer(I_Result r) {
        var l = list;
        if (l != null) {
            if (l.isEmpty()) {
                r.r1 = 0;
            } else {
                r.r1 = l.get(0);
            }
        } else {
            r.r1 = -1;
        }
    }
}

The test report is as follows:

  RESULT     SAMPLES    FREQ      EXPECT  DESCRIPTION
      -1  24.230.598  78,97%  Acceptable  Null list
       0           0   0,00%   Forbidden  Empty list
      42  6.454.586   21,03%  Acceptable  List containing 42

The result of running the following test is very different:

@JCStressTest
@State
@Outcome.Outcomes({
        @Outcome(id = "-1", expect = ACCEPTABLE, desc = "Null list"),
        @Outcome(id = "-2", expect = ACCEPTABLE_INTERESTING, desc = "Non-empty list without item"),
        @Outcome(id = "0", expect = ACCEPTABLE_INTERESTING, desc = "Empty list"),
        @Outcome(id = "42", expect = ACCEPTABLE, desc = "List containing 42"),
})
public class VolatileSaveBeforeModification {

    volatile List<Integer> list;

    @Actor
    public void actor() {
        list = new ArrayList<>();
        list.add(42);
    }

    @Actor
    public void observer(I_Result r) {
        var l = list;
        if (l != null) {
            if (l.isEmpty()) {
                r.r1 = 0;
            } else {
                try {
                    var value = l.get(0);
                    r.r1 = value != null ? value : -1;
                } catch (Exception e) {
                    r.r1 = -2;
                }
            }
        } else {
            r.r1 = -1;
        }
    }
}

The test report is as follows:

  RESULT     SAMPLES    FREQ       EXPECT  DESCRIPTION
      -1  30.773.768  88,04%   Acceptable  Null list
      -2          49  <0,01%  Interesting  Non-empty list without item
       0      62.466   0,18%  Interesting  Empty list
      42   4.118.981  11,78%   Acceptable  List containing 42

The observer still sees a null list or one containing 42 most of the time, but it also happens that it sees an empty list or one that is not empty but does not have a first item.

Applying transducers in different contexts

As mentioned above, part of the beauty of transducers is that they can be reused in different, unrelated contexts. In the examples below, we use the same transducer to modify a vector of values as well as to transform all values communicated over a core/async channel.

(into [] strings-to-the-back [1 "2" 3]) ;; Evaluates to [1 3 "2"]

(let [c (chan 3 strings-to-the-back)]
  (>!! c 1)
  (>!! c "2")
  (>!! c 3)
  (close! c)
  (-> []
      (conj (<!! c))
      (conj (<!! c))
      (conj (<!! c)))) ;; Evaluates to [1 3 "2"]

Delayed evaluation

When discussing some of the downsides of lazy seqs, we encountered the following example:

(let [r (range 3e7)]
    [(last r) (first r)])

Due to the caching of already computed values, evaluating this expression takes a large amount of memory. In situations where you need delayed eager evaluation and no caching, eduction can come in handy. The eduction function takes zero or more transducers and a collection, and returns something that can be reduced or iterated over.

I see an eduction as something that is not yet a reduction. It can become a reduction after reducing it.

Values are computed eagerly, one at a time, and only when reducing or iterating over an eduction. Computed values are not cached and thus have to be recomputed each time an eduction is reduced or iterated over again.

The following example uses an eduction to prevent the memory issues of the previous example. In this example, the eduction function is used without any transducers.

(let [r (eduction (range 3e7))]
    [(last r) (first r)])

The image below, produced with jconsole, shows that the memory used while evaluating the second expression is much lower.

Conclusion

First of all, let's answer the question in the title of this post. Should we transduce or should we not? Of course, we should. Transducers are really interesting conceptually, they're performant, and they're reusable. Should they be used every time, everywhere? I'm not convinced about that. I don't think laziness should be avoided at all costs or that transducer-based solutions are always superior to solutions using lazy seqs.

Visit https://github.com/ljpengelen/transduce-shakespeare/ to try this at home.

Experimenting with MongoDB index creation and Spring Boot

Luc Engelen — Wed, 29 Mar 2023 23:59:59 +0000

Creating indexes for MongoDB collections with Spring Boot is easy. You annotate your entities with the correct annotations, you set spring.data.mongodb.auto-index-creation to true in your configuration file, and you're done. Indexes will be created when you start your application.

Over time, however, people will start using your application, and your MongoDB collections will grow as a result. Creating an index for an empty collections takes very little time. Creating an index for a big collection can take a while. Because of this, configuring Spring to handle index creation on startup can lead to unpleasant surprises. The startup of your application will block until the new index is created, and this can take a while for existing, large collections.

Additionally, your application will not start at all if something goes wrong while creating an index. This could happen if you try to modify an existing index, for example.

All in all, it's worthwhile to take a closer look at various ways to programmatically create, find, and delete indexes.

Experimenting

I've created a small Spring Boot application accompanied by a set of tests to experiment with index creation: https://github.com/ljpengelen/mongo-index-experiments. The application itself is not much more than a single document RandomData and a repository for this document. The class RandomData looks like this:

@Builder
@CompoundIndex(def = "{ randomString: 1, randomLong: 1 }", name = "idx0")
@Data
@Document
public class RandomData {

    @Indexed
    private String randomString;

    @Indexed
    private long randomLong;

    private boolean randomBoolean;
}

The app is configured to create indexes on startup, so once you start it, four indexes are generated: one compound index corresponding to the @CompoundIndex annotation, two single-field indexes corresponding to the @Indexed annotations, and one for the implicit ID. On my machine, the app starts in about 2 seconds. Part of the startup time is spent creating indexes, but this is almost negligible.

Now, let's insert some random data by executing the following test a few times:

@Test
void savesEntities() {
    var batchSize = 100;
    var totalNumberOfEntities = 1_000_000;
    IntStream.range(0, totalNumberOfEntities / batchSize).forEach(batchNumber -> {
        var entities = Stream.generate(ExperimentApplicationTest::randomData)
                .limit(batchSize)
                .toList();
        repository.saveAll(entities);

        if (batchNumber % 500 == 0) {
            log.info("Inserting batch number {}", batchNumber);
        }
    });
}

After inserting 3 million documents and removing the previously created indexes, the app takes around 14 seconds to start on my machine. Clearly, the time it takes to create indexes is no longer negligible.

Now that I've told you the same thing twice, it's time for some new information.

One way of creating indexes programmatically uses Spring's Mongo template:

@Test
void createsIndexViaTemplate() {
    var indexOps = mongoTemplate.indexOps(COLLECTION_NAME);

    log.info("Creating index");

    var indexDefinition = new Index();
    indexDefinition.named(INDEX_NAME)
            .on("randomBoolean", Sort.Direction.ASC)
            .on("randomString", Sort.Direction.ASC)
            .on("randomLong", Sort.Direction.ASC);

    var stopWatch = new StopWatch();
    stopWatch.start();
    indexOps.ensureIndex(indexDefinition);
    stopWatch.stop();
    log.info("Time to create index: {}", stopWatch.getTotalTimeMillis());
}

On my machine, creating this index takes around 4 seconds.

With MongoDB versions before 4.2, indices could be created in the foreground or the background. Foreground builds would be faster and would lead to more efficient index data structures, but would block access to the database during the build. Background builds would not block access to the database, but would take longer to build and be less efficient.

Starting from version 4.2, access is no longer blocked while the index is built. However, access is blocked at the start and end of the build process.

Even though access to the database is not blocked during index creation, the statement indexOps.ensureIndex(indexDefinition) does block, just like the application startup blocks during index creation.

One way of ensuring that your application is not blocked during index creation is by explicitly starting a new thread for this:

@Test
void createsIndexViaTemplateInBackground() throws InterruptedException, ExecutionException {
    var indexOps = mongoTemplate.indexOps(COLLECTION_NAME);

    var completableFuture = new CompletableFuture<Void>();
    var thread = new Thread(() -> {
        log.info("Creating index");

        var indexDefinition = new Index();
        indexDefinition.named(INDEX_NAME)
                .on("randomBoolean", Sort.Direction.ASC)
                .on("randomString", Sort.Direction.ASC)
                .on("randomLong", Sort.Direction.ASC);

        var stopWatch = new StopWatch();
        stopWatch.start();
        indexOps.ensureIndex(indexDefinition);
        stopWatch.stop();
        log.info("Time to create index: {}", stopWatch.getTotalTimeMillis());

        completableFuture.complete(null);
    });

    thread.start();

    completableFuture.get();
}

Alternatively, you could use Spring's reactive Mongo template:

@Test
void createsIndexReactively() throws InterruptedException, ExecutionException {
    var indexOps = reactiveMongoTemplate.indexOps(COLLECTION_NAME);

    log.info("Creating index");

    var indexDefinition = new Index();
    indexDefinition.named(INDEX_NAME)
            .on("randomBoolean", Sort.Direction.ASC)
            .on("randomString", Sort.Direction.ASC)
            .on("randomLong", Sort.Direction.ASC);

    var completableFuture = new CompletableFuture<Void>();
    var stopWatch = new StopWatch();
    stopWatch.start();
    indexOps.ensureIndex(indexDefinition).subscribe(name -> {
        stopWatch.stop();
        log.info("Time to create index {}: {}", name, stopWatch.getTotalTimeMillis());

        completableFuture.complete(null);
    });

    completableFuture.get();
}

If you're looking for a way to create indexes that is not Spring-specific, you could also use the Mongo client for Java:

@Test
void createsIndexViaClient() {
    var keys = new BsonDocument();
    keys.put("randomLong", new BsonInt32(1));
    keys.put("randomString", new BsonInt32(1));
    keys.put("randomBoolean", new BsonInt32(1));

    var indexOptions = new IndexOptions();
    indexOptions.name(INDEX_NAME);

    var stopWatch = new StopWatch();
    stopWatch.start();
    log.info("Creating index");
    mongoClient.getDatabase(DATABASE_NAME).getCollection(COLLECTION_NAME).createIndex(keys, indexOptions);
    stopWatch.stop();
    log.info("Time to create index: {}", stopWatch.getTotalTimeMillis());
}

The statement mongoClient.getDatabase(...).getCollection(...).createIndex(keys, indexOptions) is again a blocking statement. As you might expect, all four ways take the same amount of time to create this particular index. The hard work is done by MongoDB, not our application or any library we're using.

What's in a name?

Some of the methods above are named ensureIndex, and some are named createIndex. In practice, they all behave as you would expect a method named ensureIndex to behave. They create an index if it doesn't exist yet, and they'll just do nothing if the index is already present. In other words, the following test passes and the last indexOps.ensureIndex(indexDefinition) only takes a few milliseconds:

@Test
void canEnsureExistingIndexViaTemplate() {
    var indexOps = mongoTemplate.indexOps(COLLECTION_NAME);

    var indexDefinition = new Index();
    indexDefinition.named(INDEX_NAME)
            .on("randomBoolean", Sort.Direction.ASC)
            .on("randomString", Sort.Direction.ASC)
            .on("randomLong", Sort.Direction.ASC);

    log.info("Ensuring index");
    indexOps.ensureIndex(indexDefinition);
    log.info("Ensured index");
    indexOps.ensureIndex(indexDefinition);
    log.info("Ensured index again");
}

No updates

MongoDB does not allow you to update existing indices. If you have a non-unique index with a given name and you want a unique index with that same name, for example, you have to delete the existing index and create a new one to replace it. After deleting the existing index, performance may suffer until the replacement index is built.

Alternatively, you can introduce the replacement index with a new name. It's perfectly fine to have two indexes for the same fields as long as they have different names and one is unique and the other isn't, or one is sparse and the other isn't, etc.

Automating index creation

A basic way of creating indexes at the start of your application, without blocking, is as follows:

@Component
@Slf4j
public class RandomDataIndexCreator {

    private static final String COLLECTION_NAME = "randomData";
    private static final String DATABASE_NAME = "test";

    private final MongoIndexOperations mongoIndexOperations;

    public RandomDataIndexCreator(MongoClient mongoClient) {
        mongoIndexOperations = new MongoIndexOperations(DATABASE_NAME, COLLECTION_NAME, mongoClient);
    }

    @PostConstruct
    public void startIndexCreation() {
        var indexSpecification = MongoIndexSpecification.builder()
            .definition("{ randomBoolean: 1, randomLong: 1 }")
            .build();
        new Thread(() -> mongoIndexOperations.createIndex(indexSpecification)).start();
    }
}

The class MongoIndexOperations is a wrapper around MongoClient, but you could use MongoTemplate or ReactiveMongoTemplate too. I used MongoClient because it's Spring independent, which would make it possible to use MongoIndexOperations in non-Spring applications too. See https://github.com/ljpengelen/mongo-index-experiments/blob/main/src/main/java/nl/cofx/mongo/indices/experiment/operations/MongoIndexOperations.java for the complete implementation.

It could happen that some of the indexes you need are already present on some deployment environments, for example because someone created them manually. If you know the names of these indexes, you can just issue create statements like the one above. If the index already exists, nothing will happen, as discussed above. If it doesn't exist, it will be created.

If the naming is not consistent across deployment environments, things are a little trickier. In such cases, you first have to determine whether a given index exists, regardless of the name, and only create it when it doesn't exist.

@PostConstruct
public void startIndexCreation() {
    var indexSpecification = MongoIndexSpecification.builder()
        .definition("{ randomBoolean: 1, randomLong: 1 }")
        .build();
    new Thread(() -> {
        if (mongoIndexOperations.findIndex(indexSpecification) == null) {
            mongoIndexOperations.createIndex(indexSpecification.toBuilder()
                .name("name-that-does-not-exist-in-any-deployment-environment")
                .build());
        }
    }).start();
}

Ideally, you'd use some migration framework that ensures that each index is only created once, instead of creating it (or at least verifying its existence) each time your app starts. For SQL databases, Flyway provides such functionality. I have no experience with any open-source counterpart for MongoDB.

Conclusion

If you have a few minutes to spare, I advise you to clone https://github.com/ljpengelen/mongo-index-experiments and do some experiments yourself. The proof of the pudding is in the eating.

Tiny Utterances: a minimalistic comment system

Luc Engelen — Tue, 28 Mar 2023 23:59:59 +0000

Are you looking for a free, serverless™ comment system for a technical blog? Maybe Tiny Utterances is the tool you need. All you need to get started is a GitHub issue and a few lines of CSS and JavaScript.

Conception

Tiny Utterances started out as Tiny Giscus, a clone of Giscus. Giscus is a comment system based on GitHub Discussions. Utterances, on the other hand, is based on GitHub Issues.

I started working on a minimalistic clone of Giscus that was based on GitHub Discussions too, simply because comments feel more closely related to discussions than issues. However, there's only a GraphQL API to interact with Github Discussions and this API requires a personal access token for all operations. As a result, you can't really use this API client side. Although it's technically possible, it requires you to expose one of your personal access tokens. That's also technically possible, because you could create a personal access token that can only be used to read public repositories and discussion, but GitHub immediately revokes any personal access token it finds in a repository.

Long story short, I had to switch to GitHub Issues as a basis for my minimalistic comment system. Although it's not a perfect fit conceptually, it works pretty well.

For additional details, such as installation instructions, visit the documentation.

Dependency injection and loggers in Clojure

Luc Engelen — Sat, 04 Feb 2023 23:59:59 +0000

Logging functions have to be impure to be useful. If they don't change the state of the world around them by writing something somewhere, why would you use them? This makes any function that uses a logging function directly impure too. If that is something you want to avoid, you could inject a logging service and use that instead of the logging function. Let's do that and see what challenges we come across.

The protocol Logger below consists of a single method info. The constructor function create-logger returns a concrete implementation of Logger, which delegates to clojure.tools.logging/info.

(ns logging
  (:require [clojure.tools.logging :as log]))

(defprotocol Logger
  (info [this message]))

(defn create-logger []
  (reify Logger
    (info [_ message] (log/info message))))

The function add-and-log below takes a logger as its first argument and uses it to log the result of some computation. Pay close attention to the namespace.

(ns domain
  (:require [logging :refer [create-logger info]]))

(defn add-and-log [logger & args]
  (info logger (apply + args)))

(add-and-log (create-logger) 1 2 3 4)
(add-and-log (create-logger) 1 2 3 4 5)

The result of evaluating the last two expressions is as follows:

13:47:30.130 [nREPL-session-fab93eaa-9ae3-40d4-a4f1-a0605747ba5c] INFO logging - 10
13:49:22.927 [nREPL-session-fab93eaa-9ae3-40d4-a4f1-a0605747ba5c] INFO logging - 15

These two log entries contain the log level ("INFO"), the namespace from which the logging function was called ("logging"), and the log messages ("10" and "15").

Usually, it's convenient to be able to trace an entry in the logs to its origin in the code. In this example, however, we're logging messages in the namespace domain, but the log entries contain the namespace logging. This unfortunate, but it makes perfect sense. It may look like we're logging messages in the namespace domain, because that's where we call the info method of the logger, but the actual logging happens in the namespace logging, where log/info is called.

Macros to the rescue

After some head scratching and browsing through code bases and documentation, I learned that this is one of those occasions where macros come in handy. As you may know, macros can be used to transform code at compile time. The end result of this transformation is evaluated at runtime.

For example, the macro twice below takes a function and a value, and applies the function twice: once to the value and then to the result of the first application.

(defmacro twice [f x]
  `(~f (~f ~x)))

Without going into details too much, you could view the expression (~f (~f ~x)) as a template, where ~ is used as an escape symbol.

At compile time, the expression (twice inc 0) expands to the following:

(inc (inc 0))

At runtime, this evaluates to 2.

For beginners, it can be difficult to determine whether a function or a macro should be used to solve a certain problem. In fact, the macro twice could have been a function. Most people would say that if something can be implemented as a function, then it should be implemented as function, not a macro. The problem with our logger, however, is a perfect fit for macros.

Here's a new version of the Logger protocol and the corresponding constructor function:

(ns logging
  (:require [clojure.tools.logging :as log]
            [clojure.tools.logging.impl :as impl]))

(defprotocol Logger
  (-log [this ns level message throwable]))

(defn create-logger []
  (reify Logger
    (-log [_ ns level message throwable]
      (let [logger (impl/get-logger log/*logger-factory* ns)]
        (log/log* logger level throwable message)))))

This version of the protocol consists of a single method named -log, where the minus-sign indicates that the method is not meant to be called directly. (It can be called directly, but it's not meant to be.) What's most noteworthy about this method is that it takes an argument ns. The constructor function creates a logger by passing the value of ns to the logger factory of clojure.tools.logging, and that logger is then used to do the actual logging via log/log*.

This change itself doesn't bring us any closer to solving our problem, however. We still need to figure out how to pass the namespace in which we're logging something to the method -log without doing so explicitly. Part of the answer lies in *ns*, an object representing the current namespace. Using a function in the logging namespace to pass along *ns* wouldn't work however, because we would be passing along that namespace again. The second part of the answer lies in using a macro.

(defmacro log [logger level message throwable]
  `(-log ~logger ~*ns* ~level ~message ~throwable))

As mentioned above, macros will be expanded at compile time and the resulting expression will be evaluated at runtime. Because the expansion happens where the macro is applied, the value of *ns* is the namespace in which the macro is applied, not the namespace in which the macro is defined.

To provide an API that is a little more pleasant to use, the macro above is combined with the following ones (and similar ones for other log levels).

(defmacro info [logger message]
  `(log ~logger :info ~message nil))

(defmacro error [logger message throwable]
  `(log ~logger :error ~message throwable))

Now that we've defined this collection of macros, we can evaluate the following expression.

(ns domain
  (:require [logging :refer [create-logger info]]))

(info (create-logger) "a message to log")

At compile time, the expression on the last line expands to the following:

(logging/-log (create-logger) #namespace[domain] :info "a message to log" nil)

At runtime, the message "a message to log" is logged at log level "INFO", with a reference to the namespace "domain", which is exactly what we set out to achieve.

Let's put these new macros to use:

(ns domain
  (:require [logging :refer [create-logger info]]))

(defn add-and-log [logger & args]
  (info logger (apply + args)))

(add-and-log (create-logger) 1 2 3 4)
(add-and-log (create-logger) 1 2 3 4 5)

The result of evaluating the last two expressions is now as follows:

13:58:17.378 [nREPL-session-fab93eaa-9ae3-40d4-a4f1-a0605747ba5c] INFO domain - 10
13:58:18.589 [nREPL-session-fab93eaa-9ae3-40d4-a4f1-a0605747ba5c] INFO domain - 15

Only one word changed, but this can make a world of difference when looking through logs to track down bugs.

Dependency injection and protocols in Clojure

Luc Engelen — Sun, 29 Jan 2023 23:59:59 +0000

Consider the following function, which

takes a map of dependencies and a ring request,
updates a gift using data from the request, and
returns a ring response:

(defn update-gift [{:keys [datasource]} request]
  (let [{:keys [external-list-id external-gift-id]} (:path-params request)
        {:keys [name ok price description]} (:params request)]
    (when ok
      (domain/update-gift! datasource external-gift-id name price description))
    (response/redirect (str "/list/" external-list-id "/edit") :see-other)))

The function domain/update-gift! persists the changes to the database. It has a side effect, which makes it an impure function. Because update-gift uses domain/update-gift!, it's impure too.

You could argue that this fact alone is a reason to refactor this code. Generally speaking, pure functions are easier to test and easier to reason about, which are both good reasons to prefer pure functions over impure ones.

For simple apps, however, you could also argue that there's not much to reason about anyway, and refactoring may not be worth the effort. What's more, using with-redefs to replace the impure function domain/update-gift! would make testing quite straightforward.

Because this blog post is about dependency injection, we better find another reason to refactor update-gift and apply some more dependency injection. Luckily, we can pretend that we want to replace the function domain/update-gift! with a function that uses a completely different method to persist gifts. That's not something you would do with with-redefs.

Let's look at the (spoiler alert) naive approach were we introduce a parameter to inject the function domain/update-gift! directly as a function.

(defn update-gift [{:keys [datasource update-gift!]} request]
  (let [{:keys [external-list-id external-gift-id]} (:path-params request)
        {:keys [name ok price description]} (:params request)]
    (when ok
      (update-gift! datasource external-gift-id name price description))
    (response/redirect (str "/list/" external-list-id "/edit") :see-other)))

As I mentioned above, the first argument to the function update-gift is a map of dependencies. In the example above, the key update-gift! of that map should map to a function for persisting updated gifts.

The downside of this approach is that there's no static analysis that your IDE can apply to provide you with useful information about this function. In fact, it can't even tell you that the key update-gift! maps to a function at all. You yourself have to remember that update-gift! is a function that takes a datasource, an external gift ID, a name, a price, and a description, in that order. If you forget, you have to navigate to the place where you call update-gift and see what it was again that you inject under the key update-gift!.

You could argue that this is what you get when you use a dynamically typed language instead of a statically typed one, and you would be right. However, there are good reasons to prefer dynamically typed languages over statically typed ones, and there are ways around this particular problem.

Protocols to the rescue

We can use protocols to help static analysis tools a little. A protocol is a named set of named methods and their signatures. They're similar to Java's interfaces.

The following snippet shows the definition of a simple protocol named GiftService. This protocol defines a single method update-gift!, which takes a concrete implementation of the protocol as first argument together with a number of additional arguments.

(defprotocol GiftService
  (update-gift!
    [this datasource external-id name price description]
    "Update the gift with ID `external-id` with the given name, price, and description"))

There are a number of ways to create concrete implementations of protocols. The following snippet shows one way, which uses reify.

(defn create-gift-service []
  (reify GiftService
    (update-gift!
     [_ datasource external-id name price description]
     (db/update-gift! datasource {:id external-id
                                  :name name
                                  :price price
                                  :description description}))))

The snippet shows the definition of a constructor function create-gift-service, which creates a concrete implementation of the protocol GiftService by providing an implementation of the method update-gift!. This implementation ignores the gift service itself (hence the underscore) and passes its arguments to another function db/update-gift!.

In practice, most services would have more than one method, and these methods would do more than directly call a single function. The service could perform some validation, for example, or combine a number of more low-level functions that interact with a database.

Here's the same update-gift function again. This time, a gift-service is injected as a dependency.

(defn update-gift [{:keys [datasource gift-service]} request]
  (let [{:keys [external-list-id external-gift-id]} (:path-params request)
        {:keys [name ok price description]} (:params request)]
    (when ok
      (domain/update-gift! gift-service datasource external-gift-id name price description))
    (response/redirect (str "/list/" external-list-id "/edit") :see-other)))

This function is pure, like the previous version, which makes it easier to reason about and test. Because we're injecting a service and applying a method from a protocol to it, there's more information to work with for static analysis tools. The image below shows how such a tool can show the argument list and documentation of the protocol method domain/update-gift!.

Whether or not this final version is better than the first version depends a lot on the size of the app it is part of, the plans for this app, the team working on the app, etc. The point of this post is not to convince you that you should apply dependency injection where you can or that you should always use protocols when you do apply it. The point of this post is to show you that you can have your cake and eat it when it comes to dynamically typed languages and static analysis.

shadow-cljs and running tests

Luc Engelen — Fri, 27 Jan 2023 23:59:59 +0000

When I used to work on front ends based on JavaScript or TypeScript, I usually had Karma running in watch mode while developing. Each time I saved a file, all (unit) tests would run. This would give me a short feedback loop, letting me know quickly when I was unintentionally breaking things and constantly indicating whether what I was creating matched its specifications as defined by the tests. In other words, tests were used to prevent regressions, but also as a tool to quickly see whether I was building the right things.

In the last few years, I've been using Clojure and ClojureScript to create prototypes and utilities at work as well as hobby projects and apps for personal use. Because of the size and nature of these applications, I wasn't too worried about regressions. Because Clojure and ClojureScript have excellent support for REPL-driven development, the need for tests as a means for quick feedback also disappeared. As a result, I wrote a few tests for these applications, but not nearly as many as I used to.

Deep down inside, however, I knew I would have to invest some time into learning more about testing Clojure and ClojureScript applications at some point. I wouldn't want to work in a team that produced software without decent test coverage. I should hold myself to the same standard. This week, I decided to sit down and take some time to look into different ways to execute tests for ClojureScript apps powered by shadow-cljs. As you may know, shadow-cljs is one of the two de facto standard tools for creating ClojureScript apps. The other is Figwheel.

There are a number of different ways to execute tests for a shadow-cljs based ClojureScript application. This blog post covers three of them and a number of variations. There are more alternatives, but I'll probably stick with a combination of the following for now.

Running tests on the command line

shadow-cljs supports a number of build targets for building and running tests. One of them if the :node-test target, which will gather all tests from namespaces that match a given regex and produces a build that includes these tests and a test runner for executing them.

The following configuration is the absolute minimum you need to get started. Additional configuration options are described in the user guide for shadow-cljs.

...
:builds {...
         :test {:target :node-test
                :output-to "out/node-tests.js"}
         ...}
...

Given the configuration above, executing npx shadow-cljs compile test will result in the creation of a file named out/node-test.js, which can be executed with node.

npx shadow-cljs compile test
node out/node-test.js

Executing the file leads to output like this when there are no failures:

shadow-cljs - updating dependencies
shadow-cljs - dependencies updated
[:test] Compiling ...
[:test] Build completed. (60 files, 1 compiled, 0 warnings, 2,28s)

Testing rsi.multiplication-tables-test

Ran 1 tests containing 3 assertions.
0 failures, 0 errors.

When there are failures, the output will show which assertion failed and why:

[:test] Compiling ...
[:test] Build completed. (60 files, 2 compiled, 0 warnings, 2,34s)

Testing rsi.multiplication-tables-test

FAIL in (transforming-state) (rsi/multiplication_tables_test.cljs:8:11)
correct answer on time
expected: (= {:question [1 2], :score 2, :highscore 22, :mode :against-the-clock, :wrongly-answered #{}, :deadline-passed? false} (process-answer {:question [2 3], :score 1, :highscore 1, :wrongly-answered #{}} "6" [1 2]))
  actual: (not (= {:question [1 2], :score 2, :highscore 22, :mode :against-the-clock, :wrongly-answered #{}, :deadline-passed? false} {:question [1 2], :score 2, :highscore 2, :deadline-passed? false, :wrongly-answered #{}, :mode :against-the-clock}))

Ran 1 tests containing 3 assertions.
1 failures, 0 errors.

If all tests pass, the exit code is zero. If any test fails, the exit code is one. That makes running tests like this a good option for CI servers.

If you prefer running tests in a headless browser instead of node, there's also a build target for Karma. As long as your test don't touch any code that uses browser-only APIs, I'd say that running them in node is fine. Tests like the following will fail when run with node, however:

(deftest log
  (is (= 1 ((fn [] (js/alert "1") 1)))))

Especially when combining unit tests with end-to-end tests executed via something like Cypress or Etaoin, I think it's perfectly reasonable to restrict the unit tests to testing pure functions and testing browser-specific functionality with the end-to-end tests.

Functions that make use of browser-only APIs that can't be tested efficiently via end-to-end tests could be extracted into a separate library, which could then be tested via Karma. This could make sense for functions that use localStorage, sessionStorage, cookies, or a canvas, for example.

The :node-test target has an optional configuration option :autorun. When set to true, all tests will be executed automatically after creating a build. Using this option in combination with the watch build command makes it possible to automatically run all tests each time a file is changed. You can either include the :autorun option directly in your configuration, or add it later on the command line when starting the watch build:

npx shadow-cljs watch test --config-merge '{:autorun true}'

Running tests in the browser

There's another way to automatically run all tests each time a file is changed. The :browser-test build target can be used to generate a web page that shows the results of your tests. Starting a watch build for this build target will regenerate this page each time a file is changed. The configuration below is enough to get you started, but there are additional options.

...
:builds {...
         :browser-test {:target :browser-test
                        :test-dir "out/test"}
         ...}
:dev-http {...
           3001 "out/test"
           ...}
...

The configuration above will produce the web page containing test results in the folder out/test. It also sets up an HTTP server on port 3001 that will serve this page.

If all tests pass, the page will look like this:

If any of the tests fail, the page will look like this:

Essentially, you'll get the same feedback as you'd get on the command line.

Because the favicon changes from green to red when any of the tests fail, you don't need to keep a close eye on this page all the time during development. As long as you have it open in a browser tab, you'll notice the color change soon enough when something breaks.

Running tests from the REPL

For some reason, I had high hopes for this final way of running tests. It took me quite some time before I understood what I had to do to run tests from the REPL. In the end, I wonder if there will be situations where I prefer this method over the ones above.

The library cljs.test contains a macro run-all-tests, which runs all tests in all namespaces. When you start a watch build for your shadow-cljs app and execute this macro in the REPL, you'll most likely see a list of test results for all libraries used by your app. What it probably won't show are the test results for your own app.

Because the main entrypoint for your app won't refer to any of your test namespaces, these namespaces can't be found by run-all-tests. Since you don't want the main entrypoint of your app to refer to any test namespace, you'll need another way of including them in your development build.

One way of achieving this involves the cljs.user namespace. This namespace is automatically loaded in each ClojureScript REPL started by shadow-cljs. The example below shows the content of a file named cljs/user.cljs that loads the namespaces cljs.test and rsi.multiplication-tables-test. As a result, the namespace rsi.multiplication-tables-test will be found by run-all-tests.

(ns cljs.user
  (:require [cljs.test]
            [rsi.multiplication-tables-test]))

(comment
  (cljs.test/run-all-tests)
  (cljs.test/run-all-tests #"rsi.*-test"))

The last line of the snippet above shows how you can restrict run-all-tests to the namespaces containing the tests for your app. Most likely, you're not interested in seeing the test results for all your dependencies.

Many editors that support Clojure offer functionality to trigger the evaluation of custom snippets of Clojure when a certain combination of keys is pressed. You could use that functionality to evaluate something like (cljs.test/run-all-tests #"rsi.*-test") each time you want to run your tests. Make sure to evaluate test definitions after you've changed them, however, before running the tests. Otherwise, run-all-tests will execute the previous version of your tests.

Conclusion

As mentioned above, I'm not sure which combination of these methods I'll use in the future. I'll definitely run tests on the command line for CI builds. I'll probably won't be running tests in the REPL very often. Evaluating changed test definitions before running tests requires additional key presses, and there's some extra work needed to keep cljs/user.cljs up to date.

Browser beats I: synthesizing a kick drum

Luc Engelen — Mon, 25 May 2020 00:00:00 +0000

Because I wanted to gain some experience in working with Angular and NgRx, I started building a sample-based step sequencer that runs in the browser.
To do that, I had to dive into the Web Audio API.
I'll write something about that step sequencer later.
First, I want to take a look at the basics of the Web Audio API and try to synthesize a kick drum.

The basis

At the basis of most syntesized kick drums, there's a sine wave, or something that's close to a sine wave.
The function below produces a sine wave with a frequency of 55Hz that plays for the duration of ten seconds.

const play = () => {
    const audioContextClass = window.AudioContext || window.webkitAudioContext;
    const audioContext = new audioContextClass();

    const sine = audioContext.createOscillator();
    sine.type = 'sine';
    sine.frequency.value = 55;

    sine.start(audioContext.currentTime);
    sine.stop(audioContext.currentTime + 10);
}

It sounds like this:
(You might not hear it over your laptop's speakers. You'll need decent speakers or headphones that are able to reproduce low frequencies.)

When you visualize that sound, as shown below, you'll see why it's called a sine wave.
The left-hand side of the figure shows the waveform, and the right-hand side shows the sound spectrum.

The sound spectrum is almost completely empty, except for a narrow spike at the rightmost end.
This explains why you might not hear the sound over your laptop speakers, for example.
Not all speakers are capable of reproducing sounds at low frequencies.
You can emulate the frequency response of such speakers by applying a high-pass filter.
If you filter out all frequencies below 120Hz, this is what's left of our sine wave:

The graphs below further illustrate that not much is left of the original sound.

What does that mean for our synthesized kick drum?
We'll apply a trick to make your ears believe that there's still some bass to be heard, even when listening to speakers that can't reproduce low frequencies very well.
Instead of a sine wave, we'll start out with a triangle wave.

const play = () => {
    const audioContextClass = window.AudioContext || window.webkitAudioContext;
    const audioContext = new audioContextClass();

    const triangle = audioContext.createOscillator();
    triangle.type = 'triangle';
    triangle.frequency.value = 55;

    triangle.connect(audioContext.destination);

    triangle.start(audioContext.currentTime);
    triangle.stop(audioContext.currentTime + 10);
}

Without further processing, it will look like this:

It's again clear where the name comes from.
It's also clear that there's much more going on in the spectrum graph.

Unfortunately, it sounds a little abrasive, like this:

Ideally, we'd like to process this triangle wave in such a way that it sounds more like the sine wave, without cutting off too much of the high-frequency sounds.
We can do that using a wave shaper.

const distortionCurve = (amount) => {
    const numberOfSamples = 44100;
    const curve = new Float32Array(numberOfSamples);
    const deg = Math.PI / 180;
    for (let i = 0; i < numberOfSamples; ++i) {
        const x = i * 2 / numberOfSamples - 1;
        curve[i] = (3 + amount) * x * 20 * deg / ( Math.PI + amount * Math.abs(x) );
    }
    return curve;
};

const play = () => {
    const audioContextClass = window.AudioContext || window.webkitAudioContext;
    const audioContext = new audioContextClass();

    const triangle = audioContext.createOscillator();
    triangle.type = 'triangle';
    triangle.frequency.value = 55;

    const waveShaper = audioContext.createWaveShaper();
    waveShaper.curve = distortionCurve(5);

    triangle.connect(waveShaper);
    waveShaper.connect(audioContext.destination);

    triangle.start(audioContext.currentTime);
    triangle.stop(audioContext.currentTime + 10);
}

The curve I'm using above comes from a Stack Overflow answer by Kevin Ennis.
In theory, there are multiple Sigmoid functions that you could use.
I only tried this one and stuck with it because I liked the result.

Speaking of results, here are the graphs for this sound:

The triangles look a lot more like sines, and there is still something going on at the higher end of the frequency spectrum.
The resulting sound sounds like this:

The W3C spec gives a good explanation of what's actually going on when you apply a wave shaper with a certain curve.
I won't go into the details here.

What did we achieve with this detour?
If we filter out the low frequencies again to simulate cheaper speakers, we end up with the following sound:

The graphs for this filtered sound are shown below.
When you compare these to the ones for the filtered sine wave shown above, you'll notice that there's still something to hear after removing the low end.
This is enough for you ears to trick you into believing that there's actually some low end left, even when there isn't.

Make it boom

The sound we ended up with sounds a little like "WOOOOOOOOOOH".
Let's turn that into a "WOOOOM".

const play = () => {
    const audioContextClass = window.AudioContext || window.webkitAudioContext;
    const audioContext = new audioContextClass();

    const triangle = audioContext.createOscillator();
    triangle.type = 'triangle';
    triangle.frequency.value = 55;

    const waveShaper = audioContext.createWaveShaper();
    waveShaper.curve = distortionCurve(5);

    const triangleGainNode = audioContext.createGain();
    triangleGainNode.gain.value = 1;
    triangleGainNode.gain.linearRampToValueAtTime(0, audioContext.currentTime + 0.6)

    triangle.connect(waveShaper);
    waveShaper.connect(triangleGainNode);
    triangleGainNode.connect(audioContext.destination);

    triangle.start(audioContext.currentTime);
    triangle.stop(audioContext.currentTime + 1);
}

In the snippet above, you'll see that we're using a gain node to gradually fade out over the course of 600 milliseconds.
The end result sounds like this.

Now that we have something that sounds like "WOOOOM", let's make it sound like "BOOOOM".

const play = () => {
    const audioContextClass = window.AudioContext || window.webkitAudioContext;
    const audioContext = new audioContextClass();

    const triangle = audioContext.createOscillator();
    triangle.type = 'triangle';
    triangle.frequency.value = 220;
    triangle.frequency.exponentialRampToValueAtTime(55, audioContext.currentTime + 0.1);

    const waveShaper = audioContext.createWaveShaper();
    waveShaper.curve = distortionCurve(5);

    const triangleGainNode = audioContext.createGain();
    triangleGainNode.gain.value = 1;
    triangleGainNode.gain.linearRampToValueAtTime(0, audioContext.currentTime + 0.6)

    triangle.connect(waveShaper);
    waveShaper.connect(triangleGainNode);
    triangleGainNode.connect(audioContext.destination);

    triangle.start(audioContext.currentTime);
    triangle.stop(audioContext.currentTime + 1);
}

As shown above, we do that be quickly lowering the frequency of the triangle wave from 220Hz to 55Hz over the course of 100 milliseconds.
The end result sounds like this:

If you want to achieve more of a 90s Euro house vibe, you can drop down from a higher frequency.

White noise

If you look at how classic synthesizers emulate kick drums, you'll see that they'll often use a little white noise to give the kicks a little more body.
The Web Audio API doesn't provide (white) noise out of the box, but you can use an audio buffer to create your own.

const generateWhiteNoiseBuffer = (numberOfSamples) => {
    const buffer = audioContext.createBuffer(1, numberOfSamples, audioContext.sampleRate);

    const data = buffer.getChannelData(0);
    for (let i = 0; i < numberOfSamples; ++i) {
        data[i] = Math.random() * 2 - 1;
    }

    return buffer;
}

const whiteNoiseBuffer = generateWhiteNoiseBuffer(audioContext.sampleRate);

const whiteNoiseBufferSource = () => {
    const bufferSource = audioContext.createBufferSource();
    bufferSource.buffer = whiteNoiseBuffer;
    bufferSource.loop = true;
    bufferSource.loopEnd = audioContext.sampleRate
    return bufferSource;
}

Each buffer source returned by the function whiteNoiseBufferSource can only be started once.
The same holds for the oscillator nodes that we've been creating above.
The buffer returned by generateWhiteNoiseBuffer, however, can be reused.
The result sounds like this:

The next step is to apply a fade to this sound, just like we did before.

After that, we cut of most of the higher frequencies using a low pass filter.

const play = () => {

    ...

    const noise = whiteNoiseBufferSource();

    const noiseGainNode = audioContext.createGain();
    noiseGainNode.gain.value = 1;
    noiseGainNode.gain.linearRampToValueAtTime(0, audioContext.currentTime + 0.2);

    const noiseFilter = audioContext.createBiquadFilter();
    noiseFilter.type = 'lowpass';
    noiseFilter.frequency.value = 120;

    noise.connect(noiseGainNode);
    noiseGainNode.connect(noiseFilter);
    noiseFilter.connect(audioContext.destination);

    noise.start(audioContext.currentTime);
    noise.stop(audioContext.currentTime + 1);
};

The end result sounds like this:

End result

Combining the sine-like wave and the filtered white noise leads to the following result:

All you need is a handful of other instruments and you're halfway making the next big dance hit, in your browser.

Conclusion

Let's put the results of all this hard work into action.
First, open Youtube, Spotify or whatever streaming service you like to play your favorite song.
Then, visit https://ljpengelen.github.io/groovid19/kick-snare-hihat.html and press q, w, and e to drum along.
Enjoy!

Browser beats II: synthesizing a snare drum and a hi-hat

Luc Engelen — Mon, 25 May 2020 00:00:00 +0000

In the previous installment of browser beats, we used the Web Audio API to synthesize a kick drum.
This time, we'll look at snares and hi-hats.
Once you know how to synthesize kicks, snares and hi-hats are not far away.

Snare

The snare sound we'll synthesize consists of two components.
One component represents the vibrating skins of the snare drum, the other represents the vibrating snares.
For the first component, we'll use two sine-like waves, one at 185Hz and the other at 349Hz.
I took these values from a MusicTech tutorial.
An article in Sound on Sound mentions 180Hz and 330Hz.
Obviously, you should go with whatever frequencies sound best to you.

const playSnare = () => {
    const lowTriangle = audioContext.createOscillator();
    lowTriangle.type = 'triangle';
    lowTriangle.frequency.value = 185;

    const highTriangle = audioContext.createOscillator();
    highTriangle.type = 'triangle';
    highTriangle.frequency.value = 349;

    const lowWaveShaper = audioContext.createWaveShaper();
    lowWaveShaper.curve = distortionCurve(5);

    const highWaveShaper = audioContext.createWaveShaper();
    highWaveShaper.curve = distortionCurve(5);

    const lowTriangleGainNode = audioContext.createGain();
    lowTriangleGainNode.gain.value = 1;
    lowTriangleGainNode.gain.linearRampToValueAtTime(0, audioContext.currentTime + 0.1)

    const highTriangleGainNode = audioContext.createGain();
    highTriangleGainNode.gain.value = 1;
    highTriangleGainNode.gain.linearRampToValueAtTime(0, audioContext.currentTime + 0.1)

    const snareGainNode = audioContext.createGain();
    snareGainNode.gain.value = 1;

    lowTriangle.connect(lowWaveShaper);
    lowWaveShaper.connect(lowTriangleGainNode);
    lowTriangleGainNode.connect(snareGainNode);
    snareGainNode.connect(audioContext.destination);

    highTriangle.connect(highWaveShaper);
    highWaveShaper.connect(highTriangleGainNode);
    highTriangleGainNode.connect(snareGainNode);

    lowTriangle.start(audioContext.currentTime);
    lowTriangle.stop(audioContext.currentTime + 1);

    highTriangle.start(audioContext.currentTime);
    highTriangle.stop(audioContext.currentTime + 1);
};

Together, these two sound like this:

We could have used pure sines waves here.
There's no need for applying the trick we used for the kick drum.
What you're witnessing here is a sheer waste of processing power due to my unwillingness to refactor this code right now.
Let's just say that I like the slightly more metallic sound of the distorted traingle waves.

We'll use white noise again to represent the second component.
This time, we'll use a filter to cut of all frequencies below 2kHz.

const playSnare = () => {

    ...

    const noise = whiteNoiseBufferSource();

    const noiseGainNode = audioContext.createGain();
    noiseGainNode.gain.value = 1;
    noiseGainNode.gain.linearRampToValueAtTime(0, audioContext.currentTime + 0.2);

    const noiseFilter = audioContext.createBiquadFilter();
    noiseFilter.type = 'highpass';
    noiseFilter.frequency.value = 2000;

    noise.connect(noiseGainNode);
    noiseGainNode.connect(noiseFilter);
    noiseFilter.connect(snareGainNode);

    noise.start(audioContext.currentTime);
    noise.stop(audioContext.currentTime + 1);
};

The filtered noise sounds like this:

Finally, the distorted sines and the noise together sound like this:

Hi-hat

Some filtered white noise is all you need for a hi-hat.
We again cut all frequencies below 2kHz.
This time, the volume should fade to zero in 100 milliseconds.

const playHiHat = () => {
    const noise = whiteNoiseBufferSource();

    const noiseGainNode = audioContext.createGain();
    noiseGainNode.gain.value = 1;
    noiseGainNode.gain.setValueAtTime(1, audioContext.currentTime + 0.001);
    noiseGainNode.gain.linearRampToValueAtTime(0, audioContext.currentTime + 0.1);

    const noiseFilter = audioContext.createBiquadFilter();
    noiseFilter.type = 'highpass';
    noiseFilter.frequency.value = 2000;

    const hiHatGainNode = audioContext.createGain();
    hiHatGainNode.gain.value = 0.3;

    noise.connect(noiseGainNode);
    noiseGainNode.connect(noiseFilter);
    noiseFilter.connect(hiHatGainNode);
    hiHatGainNode.connect(audioContext.destination);

    hiHatGainNode.connect(analyser)

    noise.start(audioContext.currentTime);
    noise.stop(audioContext.currentTime + 1);
};

The end result sounds like this:

Conclusion

The snare and hi-hat we've produced here are pretty basic.
If you want to dig deeper to achieve prettier or more realistic results, the following articles would be good starting points:

Don't forget to put these sounds to the test by playing along with your favorite songs: https://ljpengelen.github.io/groovid19/kick-snare-hihat.html.

Reactive Java using the Vert.x toolkit

Luc Engelen — Mon, 19 Aug 2019 11:24:45 +0000

Vert.x is a toolkit for developing reactive applications on the JVM.
Although it's possible to use Vert.x with many different languages (Java, JavaScript, Groovy, Ruby, Ceylon, Scala and Kotlin), this post will use plain old Java.

The Reactive Manifesto states that reactive systems are:

responsive,
resilient,
elastic, and
message driven.

Before we consider what that means in the context of Vert.x, let's look at one of the simplest possible applications using Vert.x:

package nl.kabisa.vertx;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServerOptions;

public class Application {

    private static class HelloWorldVerticle extends AbstractVerticle {

        @Override
        public void start() {
            var options = new HttpServerOptions().setPort(8080);
            vertx.createHttpServer(options)
                    .requestHandler(request -> request.response().end("Hello world"))
                    .listen();
        }
    }

    public static void main(String[] args) {
        Vertx.vertx().deployVerticle(new HelloWorldVerticle());
    }
}

When running this application, a single verticle is deployed when the statement Vertx.vertx().deployVerticle(new HelloWorldVerticle()); is executed.
This verticle is an instance of the classHelloWorldVerticle.
Each verticle has a start and a stop method.
The start method is called when the verticle is deployed,
and the stop method is called when the verticle is undeployed.
In this example, we only provide an implementation for the start method and reuse the (empty) stop method of the class AbstractVerticle.
When an instance of HelloworldVerticle is deployed, an HTTP server is created, which listens for incoming requests on port 8080.
Each request is answered with the plain-text response "Hello world".

Responsive

By default, Vert.x creates two threads per CPU core to deploy verticles like the one above.
Each verticle is assigned to a specific thread, and all handlers of that verticle are executed on that thread sequentially.
For the example above, this means that the handler request -> request.response().end("Hello world") is always executed on the same thread.

Because the handlers for a given verticle are never executed concurrently, you don't have to worry about locking or the atomicity of actions relevant for a single verticle.
Multiple instances of the same verticle, however, can have their handlers executed at the same time.
In fact, this holds for any two verticles.
This means that if two verticles share a resource, you might still have to worry about concurrent access to that resource.

It's your responsibility as a developer to ensure that a handler cannot occupy its assigned thread for too long.
If you block a thread for too long, Vert.x will log a warning.
The Vert.x developers took at it as their responsibility to ensure that no Vert.x API call will block a thread.
As a result, a well-designed Vert.x application can handle a large amount of events using only a few threads,
ultimately making such an application responsive.

Message driven and resilient

The example below shows an application consisting of two verticles.
It illustrates Vert.x's event bus.
The event bus allows you to broadcast messages to any number of interested receivers as well as send messages to a single receiver.
The broadcasted messages end up at each of the receivers registered for an address,
whereas the messages sent directly end up at a single receiver.

In the example below, instances of the WorldVerticle are registered as consumers on the address WORLD.
Instances of the HelloVerticle send messages to this address.
If we would deploy multiple WordVerticles, each of them would receive messages in turn.

It's possible to send messages in a number of different forms, including strings, booleans, JSON objects, and JSON arrays.
Vert.x best-effort delivery, which means that message can get lost, but are never thrown away intentionally.

package nl.kabisa.vertx;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServerOptions;

public class Application {

    private static class HelloVerticle extends AbstractVerticle {

        @Override
        public void start() {
            var options = new HttpServerOptions().setPort(8080);
            vertx.createHttpServer(options)
                    .requestHandler(request ->
                            vertx.eventBus().send("WORLD", "Hello", ar -> {
                                if (ar.succeeded()) {
                                    request.response().end((String) ar.result().body());
                                } else {
                                    request.response().setStatusCode(500).end(ar.cause().getMessage());
                                }
                            }))
                    .listen();
        }
    }

    private static class WorldVerticle extends AbstractVerticle {

        @Override
        public void start() {
            vertx.eventBus().consumer("WORLD", event -> event.reply(event.body() + " world"));
        }
    }

    public static void main(String[] args) {
        var vertx = Vertx.vertx();
        vertx.deployVerticle(new WorldVerticle());
        vertx.deployVerticle(new HelloVerticle());
    }
}

The example shows that the sender of a message can specify an optional reply handler.
The reply is provided to the handler in the form of an asynchronous result, which can either be succeeded or failed.
If it succeeded, the actual reply message is available (ar.result(), as shown in the example).
Otherwise, a throwable is available that indicates what went wrong (ar.cause(), also shown in the example).

I probably don't need to tell you that this covers the message driven part of the Reactive Manifesto.
Clearly, verticles can communicate via asynchronous message passing.

In a way, the example also illustrates resilience.
If we would deploy multiple WorldVerticles and one of them would fail, the others would just keep on doing their jobs on their own thread.
Additionally, the example shows how Vert.x reminds you to think about gracefully handling failure when implementing a handler.
Many handlers receive their input in the form of an asynchronous result, which can always be succeeded or failed, as discussed above.
Finally, and perhaps paradoxically, because of the best-effort delivery of messages via the event bus, you're also forced to consciously deal with failure related to lost messages.
If it's paramount that a given type of message is always processed, you need to implement acknowledgements and retries.

Elasticity

As mentioned above, Vert.x creates two threads per available CPU core to deploy verticles like the ones shown above.
If you need to handle more events (such as HTTP requests, for example), you can run your app on a machine with more CPU cores and reap the benefits of more concurrency, without any additional programming or configuration changes.
Additionally, it's possible to scale individual components of your application by simply deploying more or fewer verticles of a certain type.
That sounds pretty elastic to me.

Let's go overboard 🚢

If you have experience with callback-based asynchronous programming, you've probably also heard of callback hell.
Callback hell is the term used to describe the type of programs that slowly but surely move to the right-hand side of your screen,
where you're dealing with callbacks inside callbacks, inside callbacks, inside callbacks, etc.

Take the following TCP client for example:

package nl.kabisa.vertx.tcp;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.google.common.primitives.Bytes;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetClient;

public class TcpClientVerticle extends AbstractVerticle {

    public static final String REQUEST_ADDRESS = "tcp.client.request";

    private static final Logger LOGGER = LogManager.getLogger();

    private EventBus eventBus;
    private NetClient authClient;
    private NetClient echoClient;

    private void handleEvent(Message<JsonObject> event) {
        authClient.connect(3001, "localhost", asyncAuthSocket -> {
            if (asyncAuthSocket.succeeded()) {
                var authSocket = asyncAuthSocket.result();
                authSocket.handler(authBuffer -> {
                    if (authBuffer.getByte(0) == 0) {
                        event.fail(0, "Invalid credentials");
                    } else if (authBuffer.getByte(0) == 2) {
                        event.fail(0, "Unexpected error");
                    } else if (authBuffer.getByte(0) == 1) {
                        var id = authBuffer.getBytes(1, authBuffer.length());

                        echoClient.connect(3002, "localhost", asyncEchoSocket -> {
                            if (asyncEchoSocket.succeeded()) {
                                var echoSocket = asyncEchoSocket.result();
                                echoSocket.handler(echoBuffer -> {
                                    if (echoBuffer.getByte(0) == 0) {
                                        event.fail(500, "Unauthenticated");
                                    } else if (echoBuffer.getByte(0) == 1) {
                                        event.reply(echoBuffer.getBuffer(1, echoBuffer.length()));
                                    } else {
                                        event.fail(500, "Unexpected response from echo service");
                                    }
                                });
                                echoSocket.write(Buffer.buffer(Bytes.concat(id, event.body().getString("body").getBytes())));
                            } else {
                                String errorMessage = "Unable to obtain socket for echo service";
                                LOGGER.error(errorMessage, asyncEchoSocket.cause());
                                event.fail(500, errorMessage);
                            }
                        });
                    } else {
                        event.fail(500, "Unexpected response from authentication service");
                    }
                });
                authSocket.write(Buffer.buffer(new byte[] { 1, 2, 3, 4 }));
            } else {
                String errorMessage = "Unable to obtain socket for authentication service";
                LOGGER.error(errorMessage, asyncAuthSocket.cause());
                event.fail(500, errorMessage);
            }
        });
    }

    @Override
    public void start() {
        LOGGER.info("Starting");

        eventBus = vertx.eventBus();
        authClient = vertx.createNetClient();
        echoClient = vertx.createNetClient();

        eventBus.consumer(REQUEST_ADDRESS, this::handleEvent);
    }
}

This verticle listens for messages on the address tcp.client.request.
Each time a message arrives, the verticle authenticates itself with some service listening on port 3001 by exchanging some bytes.
It uses the token it receives to communicate with some other service listening on port 3002.
In the end, it replies to the initial message with a buffer containing an array of bytes received from the service listening on port 3002.
You could argue that this isn't the most beautiful piece of code ever written, although beauty lies in the eyes of the beholder.

(If you want to see the callback-based implementation of the rest of this application, by my guest: https://github.com/ljpengelen/vertx-demo/tree/971e33e4475a18fb7239d716a8c6d05369442b8a.)

Futures

JavaScript's answer to callback hell were promises.
Vert.x's answer to callback hell are futures.
A future represents the result of some computation that is potentially available at some later stage.
A future can either succeed or fail.
When it succeed, its result will be available.
When it fails, a throwable representing the cause of failure will be available.
You can set a handler for a future, which will be called with the asynchronous result when the future has succeeded or failed.
There are different ways to combine futures into a single future, which we'll illustrate with an example.

Suppose you want to deploy a number of verticles, and some of these verticles should only be deployed once others have been deployed successfully.
Vert.x offers a deploy method with a callback, which is called when the deployment has finished.
Without the use of futures, you could end up with code like this:

package nl.kabisa.vertx;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import io.vertx.core.Vertx;
import nl.kabisa.vertx.http.HttpServerVerticle;
import nl.kabisa.vertx.tcp.*;

public class Application {

    private static final Logger LOGGER = LogManager.getLogger();

    private static Vertx vertx;

    public static void main(String[] args) {
        vertx = Vertx.vertx();

        vertx.deployVerticle(new AuthServiceVerticle(), authServiceDeployment -> {
            if (authServiceDeployment.succeeded()) {
                vertx.deployVerticle(new ScreamingEchoServiceVerticle(), screamingEchoServiceDeployment -> {
                    if (screamingEchoServiceDeployment.succeeded()) {
                        vertx.deployVerticle(new TcpClientVerticle(), tcpClientDeployment -> {
                            if (tcpClientDeployment.succeeded()) {
                                vertx.deployVerticle(new HttpServerVerticle(), httpServerDeployment ->
                                    LOGGER.info("All verticles started successfully"));
                            }
                        });
                    }
                });
            }
        });
    }
}

This isn't pretty at all, even without the additional code you need to deal with possible failures.
Also, we're deploying the verticles one at a time, while we actually want to deploy the HttpServerVerticle once the others have been deployed successfully.

Rewriting this example using futures leads to the following:

package nl.kabisa.vertx;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import io.vertx.core.*;
import nl.kabisa.vertx.http.HttpServerVerticle;
import nl.kabisa.vertx.tcp.*;

public class Application {

    private static final Logger LOGGER = LogManager.getLogger();

    private static Vertx vertx;

    private static Future<String> deploy(Vertx vertx, Verticle verticle) {
        Future<String> future = Future.future();
        vertx.deployVerticle(verticle, future);
        return future;
    }

    public static void main(String[] args) {
        LOGGER.info("Starting");

        vertx = Vertx.vertx();

        CompositeFuture.all(
                deploy(vertx, new AuthServiceVerticle()),
                deploy(vertx, new ScreamingEchoServiceVerticle()),
                deploy(vertx, new TcpClientVerticle()))
                .compose(s -> deploy(vertx, new HttpServerVerticle()))
                .setHandler(s -> {
                            if (s.succeeded()) {
                                LOGGER.info("All verticles started successfully");
                            } else {
                                LOGGER.error("Failed to deploy all verticles", s.cause());
                            }
                        }
                );
    }
}

Here, we deploy three verticles at the same time, and deploy the last one when the deployment of all the others succeeded.
Again, beauty lies in the eyes of the beholder, but this is good enough for me.

Do you still remember the TCP client you saw above?
Here's the same client implemented using futures:

package nl.kabisa.vertx.tcp;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.google.common.primitives.Bytes;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;

public class TcpClientVerticle extends AbstractVerticle {

    public static final String REQUEST_ADDRESS = "tcp.client.request";

    private static final Logger LOGGER = LogManager.getLogger();

    private EventBus eventBus;
    private NetClient authClient;
    private NetClient echoClient;

    private Future<NetSocket> connectToAuthService() {
        Future<NetSocket> future = Future.future();

        authClient.connect(3001, "localhost", future);

        return future;
    }

    private Future<Buffer> authenticate(NetSocket authSocket) {
        Future<Buffer> future = Future.future();

        authSocket.handler(authBuffer -> {
            if (authBuffer.getByte(0) == 0) {
                future.fail("Invalid credentials");
            } else if (authBuffer.getByte(0) == 2) {
                future.fail("Unexpected error");
            } else if (authBuffer.getByte(0) == 1) {
                future.complete(authBuffer.getBuffer(1, authBuffer.length()));
            } else {
                future.fail("Unexpected response from authentication service");
            }
        });

        authSocket.write(Buffer.buffer(new byte[] { 1, 2, 3, 4 }));

        return future;
    }

    private Future<NetSocket> connectToEchoClient() {
        Future<NetSocket> future = Future.future();

        echoClient.connect(3002, "localhost", future);

        return future;
    }

    private Future<Buffer> forwardToEchoClient(NetSocket echoSocket, Buffer token, String input) {
        Future<Buffer> future = Future.future();

        echoSocket.handler(echoBuffer -> {
            if (echoBuffer.getByte(0) == 0) {
                future.fail("Unauthenticated");
            } else if (echoBuffer.getByte(0) == 1) {
                future.complete(echoBuffer.getBuffer(1, echoBuffer.length()));
            } else {
                future.fail("Unexpected response from echo service");
            }
        });
        echoSocket.write(Buffer.buffer(Bytes.concat(token.getBytes(), input.getBytes())));

        return future;
    }

    private void handleEvent(Message<JsonObject> event) {
        connectToAuthService()
                .compose(this::authenticate)
                .compose(token -> connectToEchoClient()
                        .compose(socket -> forwardToEchoClient(socket, token, event.body().getString("body"))))
                .setHandler(asyncBuffer -> {
                    if (asyncBuffer.succeeded()) {
                        event.reply(asyncBuffer.result());
                    } else {
                        event.fail(500, asyncBuffer.cause().getMessage());
                    }
                });
    }

    @Override
    public void start() {
        LOGGER.info("Starting");

        eventBus = vertx.eventBus();
        authClient = vertx.createNetClient();
        echoClient = vertx.createNetClient();

        eventBus.consumer(REQUEST_ADDRESS, this::handleEvent);
    }
}

Although I still have to look closely to see what the handleEvent method is doing exactly,
I hope we can agree that this is an improvement over the callback-based implementation.
In my opinion, it's clearer what each part of the implementation is responsible for and which parts are related.

Conclusion

By looking at these few examples, you've seen part of what Vert.x has to offer.
However, it doesn't end with what's described here.
Vert.x's documentation page offers a comprehensive list of books, manuals, and API docs that covers the complete toolkit.
There's also a page listing learning materials that should help you get started.

If you're interested in the toolkit, you should definitely play around with the example application available at https://github.com/ljpengelen/vertx-demo/.
Besides a few other verticles apart from those described here,
there are a number of tests that should give you an impression of what Vert.x has to offer.

Once you get the hang of it, developing applications with Vert.x is quite enjoyable.
As with all forms of asynchronous programming, however, I sometimes find myself in slightly annoying situations where a synchronous approach would be much easier to implement and reason about.
The question is whether you're willing to put up with a little extra work to enjoy the potential benefits of reactive systems.

Running multiple Docker containers in parallel with Jenkins

Luc Engelen — Mon, 19 Aug 2019 00:00:00 +0000

This morning, I was looking for a way to run multiple Docker containers in parallel with Jenkins.
Even though this seemed like a common use case to me, it took me a while to find all information I needed and piece it together.
As you know, the only design pattern you need is copy-paste.
I wrote this post to allow you and my future self to copy-paste some useful snippets from a Jenkinsfile.

Suppose you have a Java app that requires a PostgreSQL database.
If you want to run a few integration tests for that app, you'll probably need this database to be accessible as well.
To make it possible for Jenkins to run these integration tests for you, you could just install PostgreSQL on the machine running Jenkins, create the necessary databases and users, and call it a day.
However, I'd rather keep each app in its own Docker container and only have apps running when they're needed.

Consider the following Jenkinsfile:

def withDockerNetwork(Closure inner) {
  try {
    networkId = UUID.randomUUID().toString()
    sh "docker network create ${networkId}"
    inner.call(networkId)
  } finally {
    sh "docker network rm ${networkId}"
  }
}

pipeline {
  agent none

  stages {
    stage("test") {
      agent any

      steps {
        script {
          def database = docker.build("database", "database")
          def app = docker.build("app", "-f dockerfiles/ci/Dockerfile .")

          withDockerNetwork{ n ->
            database.withRun("--network ${n} --name database") { c ->
              app.inside("""
                --network ${n}
                -e 'SPRING_DATASOURCE_URL=jdbc:postgresql://database:5432/test'
              """) {
                sh "mvn verify"
              }
            }
          }
        }
      }
    }
  }
}

The function withDockerNetwork (copy-pasted from Ryan Desmon) creates and eventually deletes a Docker network with a random name.
After creating the network, it calls a block of code of your choice and provides it with this random name.
After the block of code has finished, the network is deleted.

The statement docker.build("database", "database") builds a Docker image named "database" with the context database.
The statement docker.build("app", "-f dockerfiles/ci/Dockerfile .") builds a Docker image named "app" from the Dockerfile dockerfiles/ci/Dockerfile with context ..

Once both images are built, containers based on these images are started and connected to the same network, allowing them to communicate.
The arguments --network ${n} are used to connect both containers to the network.
The container for the database is given a name explicitly with the argument --name database, so that we can point the app to it.
The latter is achieved by setting an environment variable with the argument -e 'SPRING_DATASOURCE_URL=jdbc:postgresql://database:5432/test'.
This last step is specific to Spring.
You'll probably need to do something completely different for your own use case.

Once both containers are running, the tests for the app are executed by the step sh "mvn verify".
This step is specific to Java and Maven and is again unrelated to running containers in parallel.

If you want to see this in action, take a look at https://github.com/ljpengelen/java-meetup-jwt.
The example above is a simplified version of the Jenkinsfile used for this project.