Removed the ContextFilter and replaced it with a standard function

Author: wolpert

lib/src/main/java/com/codeheadsystems/microbus/Bus.java

@@ -1,6 +1,8 @@
 package com.codeheadsystems.microbus;
 
+import java.util.Map;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 public interface Bus<D> {
 
@@ -10,9 +12,9 @@ public interface Bus<D> {
    *
    * @param filler for a context.
    */
-  void register(ContextFiller<D> filler);
+  void registerContextFiller(Function<D, Map<String, Object>> filler);
 
-  void unregister(ContextFiller<D> filler);
+  void unregisterContextFiller(Function<D, Map<String, Object>> filler);
 
   void publish(Message<D> message);
 

lib/src/main/java/com/codeheadsystems/microbus/ContextFiller.java

@@ -5,6 +5,9 @@
 import java.util.UUID;
 import java.util.function.Supplier;
 
+/**
+ * Uses a normal random object for the UUID. Create your own with a SecureRandom if needed.
+ */
 public class DefaultUUIDSupplier implements Supplier<UUID> {
 
   private final TimeBasedEpochGenerator generator;

lib/src/main/java/com/codeheadsystems/microbus/DefaultUUIDSupplier.java

@@ -7,15 +7,21 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * Basic implementation for a buss.
+ *
+ * @param <D> data used for the messages. This can be as simple or complex as you want.
+ */
 public class MicroBus<D> implements Bus<D> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(MicroBus.class);
 
   private final List<Consumer<Message<D>>> subscribers = new CopyOnWriteArrayList<>();
-  private final List<ContextFiller<D>> contextFillers = new CopyOnWriteArrayList<>();
+  private final List<Function<D, Map<String, Object>>> contextFillers = new CopyOnWriteArrayList<>();
   private final ContextFactory contextFactory;
   private final ExecutorService executorService;
 
@@ -31,13 +37,13 @@ public MicroBus(ContextFactory contextFactory,
   }
 
   @Override
-  public void register(ContextFiller<D> filler) {
+  public void registerContextFiller(Function<D, Map<String, Object>> filler) {
     LOGGER.debug("register({})", filler);
     contextFillers.add(Objects.requireNonNull(filler));
   }
 
   @Override
-  public void unregister(ContextFiller<D> filler) {
+  public void unregisterContextFiller(Function<D, Map<String, Object>> filler) {
     LOGGER.debug("unregister({})", filler);
     contextFillers.remove(filler);
   }
@@ -67,9 +73,12 @@ public void unsubscribe(Consumer<Message<D>> subscriber) {
   private void dispatch(Message<D> message) {
     contextFactory.withRunnable(() -> {
       LOGGER.debug("[{}] dispatch({})", contextFactory.contextUuid(), message.uuid());
+      // Fill up the context with any fillers first. Store the results we get.
       Map<String, Object> properties = contextFactory.currentContext()
-          .map(ctx ->  fillContext(message, ctx))
-          .orElseThrow();
+          .map(ctx -> fillContext(message, ctx))
+          .orElseThrow(); // should never happen.
+      // Now call each subscriber with the message and note context.
+      // Remember we have to copy the properties from the current context if we're doing async stuff.
       for (Consumer<Message<D>> subscriber : subscribers) {
         if (message.async()) {
           dispatchAsync(message, subscriber, properties);
@@ -111,10 +120,10 @@ private void callSubscriber(final Message<D> message,
   private Map<String, Object> fillContext(final Message<D> message,
                                           final Context context) {
     LOGGER.debug("[{}] fillContext({})", contextFactory.contextUuid(), message.uuid());
-    for (ContextFiller<D> filler : contextFillers) {
+    for (Function<D, Map<String, Object>> filler : contextFillers) {
       try {
         LOGGER.trace("[{}] filling context from {} for {}", contextFactory.contextUuid(), filler, message.uuid());
-        filler.fillContext(message.data(), context);
+        context.properties().putAll(filler.apply(message.data()));
       } catch (Exception e) {
         LOGGER.error("ContextFiller threw exception for message {}: {}", message.uuid(), filler, e);
       }

lib/src/main/java/com/codeheadsystems/microbus/MicroBus.java

@@ -2,7 +2,9 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 
+import java.lang.reflect.Field;
 import java.util.Optional;
+import java.util.Stack;
 import java.util.UUID;
 import java.util.function.Supplier;
 import org.junit.jupiter.api.BeforeEach;
@@ -145,4 +147,38 @@ void clearAllContexts_insideWithSupplier_finallySurvivesEmptyStack() {
     assertEquals(Optional.empty(), contextFactory.currentContext());
   }
 
+  @Test
+  void withContextSet_emptyStack_logsWarningButDoesNotThrow() {
+    // The finally block handles the case where the stack was cleared during execution.
+    // This can't happen through the public API alone, so we use reflection.
+    contextFactory.withRunnable(() -> {
+      getStack().clear();
+    });
+    assertEquals(Optional.empty(), contextFactory.currentContext());
+  }
+
+  @Test
+  void withContextSet_mismatchedContext_logsWarningButDoesNotThrow() {
+    // The finally block handles the case where a different context is on top of the stack.
+    // This can't happen through the public API alone, so we use reflection.
+    contextFactory.withRunnable(() -> {
+      Stack<Context> stack = getStack();
+      stack.pop();
+      stack.push(new Context(UUID.randomUUID()));
+    });
+    assertEquals(Optional.empty(), contextFactory.currentContext());
+  }
+
+  @SuppressWarnings("unchecked")
+  private Stack<Context> getStack() {
+    try {
+      Field field = ContextFactory.class.getDeclaredField("threadLocalContext");
+      field.setAccessible(true);
+      ThreadLocal<Stack<Context>> threadLocal = (ThreadLocal<Stack<Context>>) field.get(contextFactory);
+      return threadLocal.get();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
 }

lib/src/test/java/com/codeheadsystems/microbus/ContextFactoryTest.java

@@ -4,12 +4,14 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -184,13 +186,13 @@ void subscribe_nullSubscriber_throws() {
   }
 
   @Test
-  void register_nullFiller_throws() {
-    assertThrows(NullPointerException.class, () -> bus.register(null));
+  void registerContextFiller_nullFiller_throws() {
+    assertThrows(NullPointerException.class, () -> bus.registerContextFiller(null));
   }
 
   @Test
-  void register_fillerPopulatesContextBeforeSubscribers() {
-    bus.register((data, context) -> context.properties().put("key", data.toUpperCase()));
+  void registerContextFiller_fillerPopulatesContextBeforeSubscribers() {
+    bus.registerContextFiller(data -> Map.of("key", data.toUpperCase()));
 
     List<Object> captured = new ArrayList<>();
     bus.subscribe(msg -> captured.add(contextFactory.currentContext().orElseThrow().properties().get("key")));
@@ -201,9 +203,9 @@ void register_fillerPopulatesContextBeforeSubscribers() {
   }
 
   @Test
-  void register_multipleFillers_allInvoked() {
-    bus.register((data, context) -> context.properties().put("a", "1"));
-    bus.register((data, context) -> context.properties().put("b", "2"));
+  void registerContextFiller_multipleFillers_allInvoked() {
+    bus.registerContextFiller(data -> Map.of("a", "1"));
+    bus.registerContextFiller(data -> Map.of("b", "2"));
 
     List<Object> capturedA = new ArrayList<>();
     List<Object> capturedB = new ArrayList<>();
@@ -220,9 +222,9 @@ void register_multipleFillers_allInvoked() {
   }
 
   @Test
-  void register_fillerException_doesNotStopOtherFillersOrSubscribers() {
-    bus.register((data, context) -> { throw new RuntimeException("filler boom"); });
-    bus.register((data, context) -> context.properties().put("survived", true));
+  void registerContextFiller_fillerException_doesNotStopOtherFillersOrSubscribers() {
+    bus.registerContextFiller(data -> { throw new RuntimeException("filler boom"); });
+    bus.registerContextFiller(data -> Map.of("survived", true));
 
     List<Object> captured = new ArrayList<>();
     bus.subscribe(msg -> captured.add(contextFactory.currentContext().orElseThrow().properties().get("survived")));
@@ -233,15 +235,15 @@ void register_fillerException_doesNotStopOtherFillersOrSubscribers() {
   }
 
   @Test
-  void unregister_removesFiller() {
-    ContextFiller<String> filler = (data, context) -> context.properties().put("key", "value");
-    bus.register(filler);
+  void unregisterContextFiller_removesFiller() {
+    Function<String, Map<String, Object>> filler = data -> Map.of("key", "value");
+    bus.registerContextFiller(filler);
 
     List<Object> captured = new ArrayList<>();
     bus.subscribe(msg -> captured.add(contextFactory.currentContext().orElseThrow().properties().get("key")));
 
     bus.publish(new Message<>("first", false, UUID_1));
-    bus.unregister(filler);
+    bus.unregisterContextFiller(filler);
     bus.publish(new Message<>("second", false, UUID_1));
 
     assertEquals(2, captured.size());
@@ -250,9 +252,9 @@ void unregister_removesFiller() {
   }
 
   @Test
-  void register_async_fillerPopulatesContext() throws Exception {
+  void registerContextFiller_async_fillerPopulatesContext() throws Exception {
     CountDownLatch latch = new CountDownLatch(1);
-    bus.register((data, context) -> context.properties().put("async-key", data));
+    bus.registerContextFiller(data -> Map.of("async-key", data));
 
     List<Object> captured = new ArrayList<>();
     bus.subscribe(msg -> {