soda docs v17.0.1
/

Event Bus & Aspect

Nearly all of soda's automation comes from two @Aspect classes:

AspectClassInterceptsDoes
CQRS aspectCqrsAroundHandlerthe handle methods of Command / Query / Event, plus CommandBus.sendcontext propagation, recursion protection, chain logging
Repository aspectRepositoryEventAspectsave / update / delete / operate on types named *Repository and their subtypestakes the domain events, publishes them after the transaction commits

Understand these two aspects and you understand most of soda's runtime behavior.

CqrsAroundHandler

com.hibuka.soda.bus.interceptor.CqrsAroundHandler. It's an @Aspect class, registered by SpringBusAutoConfiguration:

java
@Bean
@ConditionalOnMissingBean
public CqrsAroundHandler cqrsAroundHandler(EventBus eventBus, EventProperties eventProperties) {
    return new CqrsAroundHandler(eventBus, eventProperties);
}

It has two @Around pointcuts.

Pointcut one: intercept every Handler

text
execution(* com.hibuka.soda.cqrs.command.CommandHandler+.handle(..)) ||
execution(* com.hibuka.soda.cqrs.query.QueryHandler+.handle(..)) ||
execution(* com.hibuka.soda.cqrs.event.EventHandler+.handle(..))

+ is an AspectJ type pattern meaning “including subtypes”. So your CreateProducerCommandHandler implements CommandHandler<...> gets intercepted.

This pointcut does three things.

1. Establish context. Based on the type of the first argument, write context into DomainEventContext (ThreadLocal):

java
if (firstArg instanceof BaseCommand) {
    BaseCommand cmd = (BaseCommand) firstArg;
    requestId = cmd.getRequestId();
    username = cmd.getUserName();

    DomainEventContext.setRequestId(requestId);
    DomainEventContext.setUserName(username);
    DomainEventContext.setJti(cmd.getJti());
    DomainEventContext.setAuthorities(cmd.getAuthorities());
    if (cmd.getCallerUid() != null) {
        DomainEventContext.setCallerUid(String.valueOf(cmd.getCallerUid()));
    }
    if (cmd.getHopCount() != null) {
        DomainEventContext.setHopCount(cmd.getHopCount());
    }
}

The BaseQuery branch only reads requestId and userName and writes no context — the read chain produces no events and needs no context.

The AbstractDomainEvent branch restores context so that new commands sent from an EventHandler inherit the chain information.

2. Clean up context. In finally:

java
} finally {
    // Clean up context to prevent memory leaks
    if (firstArg instanceof BaseCommand || firstArg instanceof AbstractDomainEvent) {
        DomainEventContext.clear();
    }
}
Important

This clear() is mandatory. DomainEventContext is a ThreadLocal — if thread pools reuse threads without cleanup, one request's requestId leaks into the next request's events, and during debugging you'll see ghostly phenomena like “user A's request triggered a data change for user B”.

3. Write chain logs. Fixed format:

text
requestId:{},method:{}-{}, {}:{}, result:{}, duration:{}-{},username:{}

That is: requestId:abc-123,method:com.example.demo.app.handler.CreateProducerCommandHandler-handle, CreateProducerCommand:{"name":"x"}, result:{"uid":"..."}, duration:12-0,username:alice.

Two duration fields: the first is the method execution time, the second is the time after log serialization.

The special branch in Redis mode

When soda.event.bus-type=redis, the aspect skips EventHandlers:

java
boolean streamEnabled = isStreamEnabled();
boolean isEventHandler = firstArg instanceof AbstractDomainEvent;
boolean isStreamConsumer = false;
boolean isAsyncStreamThread = Thread.currentThread().getName().startsWith("SimpleAsyncTaskExecutor");
if (isEventHandler && streamEnabled && !isStreamConsumer && !isAsyncStreamThread) {
    logger.info("[CqrsAroundHandler] Stream enabled, skip EventHandler on thread: {}", Thread.currentThread().getName());
    return null;
}

The reason: in Redis mode events are already written to the Stream, and Redis consumer threads handle deserialization and delivery. Processing locally as well would double-consume.

Note

isStreamConsumer is always false in the current code — it's a reserved switch. In other words, the real check is “is the EventHandler running on a SimpleAsyncTaskExecutor thread”. When you run async EventHandlers on the default cqrs-async- thread pool (ThreadPoolTaskExecutor), this thread-name check misses and an event may be skipped once.

Pointcut two: intercept CommandBus.send

text
execution(* com.hibuka.soda.cqrs.command.CommandBus+.send(..))

This pointcut does two things.

1. Complete the command's context. If the new command's requestId is null, fill it from DomainEventContext:

java
if (cmd.getRequestId() == null) {
    String requestId = DomainEventContext.getRequestId();
    if (requestId != null) {
        cmd.setRequestId(requestId);
        cmd.setUserName(DomainEventContext.getUserName());
        cmd.setJti(DomainEventContext.getJti());
        cmd.setAuthorities(DomainEventContext.getAuthorities());
        String callerUid = DomainEventContext.getCallerUid();
        if (callerUid != null) {
            try {
                cmd.setCallerUid(callerUid);
            } catch (Exception ignored) {}
        }
    }
}

This is what keeps the “event → command” chain going: when an EventHandler calls busFacade.sendCommand(newCmd), the new command automatically inherits the event's context.

2. Async recursion protection. Every send bumps the command's hopCount by one, and over 20 throws:

java
Integer hopCount = DomainEventContext.getHopCount();
int currentHop = (hopCount != null) ? hopCount : 0;
if (currentHop > 20) {
    logger.error("[CqrsAroundHandler] Async recursion too deep! hopCount={}, command={}", currentHop, cmd.getClass().getSimpleName());
    throw new IllegalStateException("Async recursion too deep! hopCount=" + currentHop);
}
cmd.setHopCount(currentHop + 1);
Important

The 20-hop limit is the async protection. Synchronous recursion is guarded by BusFacade's MAX_RECURSION_DEPTH = 10. The two mechanisms work independently. Details: Recursion Protection.

3. Publish events produced by the command. The third job of pointcut two, easy to overlook:

java
if (result instanceof DomainEvents) {
    for (AbstractDomainEvent event : ((DomainEvents) result).getDomainEvents()) {
        sycBaseInfo((BaseCommand) args[0], event);
        if (streamEnabled) {
            // Redis mode: skip local publish, hand it to the repository aspect
        } else {
            eventBus.publish(event);
        }
    }
} else if (result instanceof DomainEvent) {
    // the command returns a domain event directly
    ...
}

That is: if the aggregate root a CommandHandler returns still carries unpublished domain events, the aspect publishes them right away.

Warning

Doesn't this double-publish with the repository aspect? No — the repository aspect takes the events and calls clearDomainEvents() at save time. By the time pointcut two inspects the return value, the event list is already empty.

But if you manually addPendingEvent after save in a CommandHandler (or call update instead of save), events can go out twice. The rule: events are published only by the repository aspect — never call eventBus.publish manually in a Handler.

RepositoryEventAspect

com.hibuka.soda.event.spring.RepositoryEventAspect. Registered by RepositoryEventAutoConfiguration:

java
@Configuration
@EnableAspectJAutoProxy
@ConditionalOnBean(EventBus.class)
@AutoConfigureAfter(name = {
        "com.hibuka.soda.event.spring.SpringEventBusAutoConfiguration",
        "com.hibuka.soda.event.redis.RedisEventBusAutoConfiguration"
})
public class RepositoryEventAutoConfiguration {
    @Bean
    public RepositoryEventAspect repositoryEventAspect(EventBus eventBus, ObjectProvider<CommandContext> commandContextProvider) {
        return new RepositoryEventAspect(eventBus, commandContextProvider);
    }
}

@ConditionalOnBean(EventBus.class) means: with no EventBus bean in the container, this aspect isn't registered at all, and repository calls are just ordinary method calls.

The pointcut

text
execution(* *..*Repository+.save(..)) ||
execution(* *..*Repository+.update(..)) ||
execution(* *..*Repository+.delete(..)) ||
execution(* *..*Repository+.operate(..))
Warning

This expression has three critical points, and each is a trap:

  1. *..*Repository+ matches types whose name ends with Repository, plus their subtypes. So a repository interface named ProducerWriteRepository is enough — an implementation class named ProducerRepoImpl still gets intercepted. But when neither the interface nor the implementation name contains Repository (a ProducerStore, say), the pointcut misses and events are silently lost.
  2. Method names must be save / update / delete / operate. Rename to persist or insert and it silently fails the same way.
  3. *..* means any package. This means any class named XxxRepository in any of your modules matches. The aspect checks whether the first argument is an AbstractAggregateRoot subclass and skips it otherwise — but if you happen to have a utility class called OrderRepository with a save(String) method, this aspect will see it too (it just gets skipped, no impact).

Debugging mantra: if events aren't going out, first check the repository type name contains Repository, then check the method names.

Execution flow

java
public Object aroundSave(ProceedingJoinPoint joinPoint) throws Throwable {
    Object result = joinPoint.proceed();          // ① run the real save first
    Object[] args = joinPoint.getArgs();

    if (args != null && args.length > 0 && args[0] instanceof AbstractAggregateRoot) {
        AbstractAggregateRoot agg = (AbstractAggregateRoot) args[0];
        List<AbstractDomainEvent> events = List.copyOf(agg.getDomainEvents());

        if (!events.isEmpty()) {
            agg.clearDomainEvents();                // ② take the events and clear
            enrichEventsWithContext(events);        // ③ top up context

            if (TransactionSynchronizationManager.isSynchronizationActive()) {
                // ④ transaction present: register an afterCommit callback
                TransactionSynchronizationManager.registerSynchronization(
                        new TransactionSynchronization() {
                            @Override
                            public void afterCommit() {
                                publishEvents(events);
                            }
                        });
            } else {
                // ⑤ no transaction: publish immediately
                publishEvents(events);
            }
        }
    }
    return result;
}

Key points:

  • Run save first, then take the events. The order cannot flip — if save throws, the events are never taken and never published.
  • List.copyOf makes a snapshot copy, then clears the aggregate's list. So even if something else modifies the events before publication, it doesn't matter.
  • Transaction present → afterCommit; no transaction → publish immediately.

Transaction callback

java
if (TransactionSynchronizationManager.isSynchronizationActive()) {
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
        @Override
        public void afterCommit() {
            publishEvents(events);
        }
    });
} else {
    publishEvents(events);
}

isSynchronizationActive() asks “is the current thread in transaction synchronization” — i.e. has a transaction started. Inside a Spring-managed @Transactional method it's true.

Important

On rollback afterCommit does not run — the events are never published. This is how “publish events only after the transaction commits” is implemented: the phantom of “the business failed but the event went out” cannot happen.

Conversely, if you don't add @Transactional, events are published the moment save returns. If the save succeeded but later logic failed, the events are already out — data and events are inconsistent. So always open a transaction on the write chain.

Context top-up

enrichEventsWithContext has two priority levels:

java
// Priority 1: CommandContext (Legacy)
if (ctx != null) {
    events.forEach(e -> {
        e.setRequestId(ctx.getRequestId());
        e.setJti(ctx.getJti());
        String[] authoritiesArray = ctx.getAuthorities();
        if (authoritiesArray != null) {
            e.setAuthorities(String.join(",", authoritiesArray));
        }
        e.setUserName(ctx.getUserName());
        ...
    });
}

// Priority 2: DomainEventContext (New, supports hopCount and context propagation)
String requestId = DomainEventContext.getRequestId();
if (requestId != null) {
    events.forEach(e -> {
        if (e.getRequestId() == null) {
            e.setRequestId(requestId);
            ...
        }
        // Always update hopCount from ThreadLocal if available
        Integer hopCount = DomainEventContext.getHopCount();
        if (hopCount != null) {
            e.setHopCount(hopCount);
        }
    });
}
  • CommandContext is the legacy mechanism; its authorities is a String[], joined with commas.
  • DomainEventContext is the new mechanism, with hopCount on top. Its fields are filled only when empty (if (e.getRequestId() == null)), but hopCount is overwritten unconditionally — because it backs async recursion protection.

Note CommandContext has higher priority: once present, it unconditionally overwrites the event's context fields.

Publishing

java
private void publishEvents(List<AbstractDomainEvent> events) {
    for (AbstractDomainEvent event : events) {
        try {
            eventBus.publish(event);
        } catch (Exception e) {
            log.error("Failed to publish domain event: {}", event, e);
        }
    }
}
Note

A single failed event publication doesn't affect the other events and doesn't roll back the transaction. The exception is caught and only logged. This means: when event delivery fails, your business data is already committed but downstream never hears about it. In cross-service scenarios you need Redis mode's dead-letter stream or a compensation job as the safety net.

The two event buses

The EventBus interface has three methods:

java
public interface EventBus {
    void publish(DomainEvent event) throws BaseException;
    void subscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException;
    void unsubscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException;
}

SimpleEventBus

The in-process implementation. Enabled when soda.event.bus-type is unset or simple:

java
@Configuration
@ConditionalOnProperty(name = "soda.event.bus-type", havingValue = "simple", matchIfMissing = true)
public class SimpleEventBusAutoConfiguration {
    @Bean
    @ConditionalOnMissingBean
    public EventBus simpleEventBus(List<EventHandler<? extends DomainEvent>> eventHandlers) {
        return new SimpleEventBus(eventHandlers);
    }
}

Internally it's a Map<Class<? extends DomainEvent>, List<EventHandler>>, with the List a CopyOnWriteArrayList — so one event type can have multiple Handlers.

On publish, type lookup walks up the type chain:

java
public void publish(DomainEvent event) throws BaseException {
    Class<?> eventType = event.getClass();
    while (eventType != null) {
        List<EventHandler> eventHandlers = handlers.get(eventType);
        if (eventHandlers != null && !eventHandlers.isEmpty()) {
            for (EventHandler handler : eventHandlers) {
                handler.handle(event);
            }
        }

        // Check all interfaces implemented by this class
        for (Class<?> interfaceType : eventType.getInterfaces()) {
            if (DomainEvent.class.isAssignableFrom(interfaceType)) {
                List<EventHandler> interfaceHandlers = handlers.get(interfaceType);
                ...
            }
        }

        eventType = eventType.getSuperclass();
    }
}

In other words, Handler implements EventHandler<AbstractDomainEvent> receives every domain event; Handler implements EventHandler<ProducerEvent> receives only ProducerEvent and its subclasses.

Tip

Writing an EventHandler<DomainEvent> as a “global audit handler” is a very convenient pattern — it captures all events.

RedisStreamEventBus

The cross-service implementation. Enabled when soda.event.bus-type=redis; requires the soda-event-redis module and a RedisTemplate on the classpath.

java
@Configuration
@ConditionalOnClass(RedisTemplate.class)
@ConditionalOnProperty(name = "soda.event.bus-type", havingValue = "redis")
public class RedisEventBusAutoConfiguration {
    @Bean
    @Primary
    @ConditionalOnBean(RedisConnectionFactory.class)
    public EventBus redisStreamEventBus(...) { ... }
}

It is @Primary — with Redis present, it overrides SimpleEventBus.

It's built on Redis Streams and provides:

CapabilityDescription
Consumer groupsXGROUP CREATE, multiple consumers in one group share the load
Consumption ackXACK, acknowledged only after successful handling
Batch pullXREADGROUP, batchSize controls entries per pull
Concurrent consumptionconcurrency consumer threads, consumerName gets -0, -1 suffixes
Failure retryup to maxRetries times, delay initialRetryDelay * 2^n
Dead-letter streamwritten to deadLetterStream after retries are exhausted
Idempotencyvia IdempotencyService, dedup by eventId
Periodic cleanup@Scheduled(fixedRate = 3600000) clears expired idempotency state

The payload on publish:

java
Map<String, String> entry = new HashMap<>(3);
String eventJson = serializer.toJson(event);
entry.put("event", eventJson);
entry.put("type", event.getClass().getName());
if (context != null) {
    entry.put("context", serializer.toJson(context));
}
streamOps.add(streamKey, entry);

type stores the event class's fully qualified name — the consumer uses it to deserialize.

The Redis bus's publish is also transaction-aware:

java
if (TransactionSynchronizationManager.isActualTransactionActive()) {
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
        @Override
        public void afterCommit() {
            publishAction.run();
        }
    });
} else {
    publishAction.run();
}
Note

Note this uses isActualTransactionActive() while the repository aspect uses isSynchronizationActive(). The semantics are close but not identical — the former asks “is there a real transaction”, the latter “is the transaction synchronization mechanism active”. In practice the difference rarely matters.

The spring option doesn't actually exist

EventProperties.busType defaults to "spring", and docs often list spring / simple / redis as three choices. But in 17.0.1:

  • the soda-event-spring module has been deleted
  • no EventBus implementation has havingValue = "spring"

So configuring soda.event.bus-type=spring results in no EventBus bean at all. The consequences:

  • RepositoryEventAutoConfiguration doesn't register due to @ConditionalOnBean(EventBus.class) → the repository aspect is dead → domain events never go out
  • SpringBusAutoConfiguration.cqrsAroundHandler requires an EventBus parameter → bean creation fails
Warning

In production use simple or redis, never spring.

The safest move is to configure soda.event.bus-type=simple explicitly. Unconfigured, SimpleEventBusAutoConfiguration's matchIfMissing = true takes effect (Simple path), but writing it out is clearer — and dodges the misleading "spring" default in EventProperties.

Auto-configuration inventory

soda-core-base's META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports:

text
com.hibuka.soda.bus.configuration.SpringBusAutoConfiguration
com.hibuka.soda.bus.configuration.SimpleEventBusAutoConfiguration
com.hibuka.soda.event.spring.RepositoryEventAutoConfiguration

The same file in soda-event-redis:

text
com.hibuka.soda.event.redis.RedisEventBusAutoConfiguration

Beans registered by SpringBusAutoConfiguration:

BeanConditionDescription
commandBus@ConditionalOnMissingBeanSimpleCommandBus
queryBus@ConditionalOnMissingBeanSimpleQueryBus
cqrsAsyncExecutor@ConditionalOnMissingBean(name = ...)ThreadPoolTaskExecutor
busFacade@ConditionalOnMissingBeanBusFacade
cqrsAroundHandler@ConditionalOnMissingBeanthe CQRS aspect
eventProperties@ConditionalOnMissingBeansoda.event.* config
Tip

Every key bean is @ConditionalOnMissingBean. That is the extension point: define your own bean with the same name and the framework's is skipped. See Extension Points.

The execution order of the two aspects

In one complete write request, the aspects nest like this:

text
BusFacade.sendTransactCommand(command)
  │
  ├─ CommandBus.send(command)
  │    │
  │    ├─ CqrsAroundHandler.handleEventBus      ← outer
  │    │    ├─ backfill command context
  │    │    ├─ hopCount check, +1
  │    │    ├─ proceed()
  │    │    │    └─ SimpleCommandBus.send → finds CommandHandler → handle(command)
  │    │    │         │
  │    │    │         └─ CqrsAroundHandler.handleExecutor   ← inner
  │    │    │              ├─ write DomainEventContext
  │    │    │              ├─ proceed()
  │    │    │              │    ├─ domain.create() → addPendingEvent
  │    │    │              │    └─ producerWriteRepository.save(domain)
  │    │    │              │         └─ RepositoryEventAspect.aroundSave
  │    │    │              │              ├─ proceed() → the real insert
  │    │    │              │              ├─ take events, clear, top up context
  │    │    │              │              └─ register afterCommit
  │    │    │              ├─ clear() DomainEventContext
  │    │    │              └─ write chain log
  │    │    │
  │    │    └─ check return value: DomainEvents / DomainEvent → publish
  │    │
  │    └─ return
  │
  └─ transaction commits
       └─ afterCommit → eventBus.publish(event)
            └─ ProducerEventHandler.handle(event)
                 └─ CqrsAroundHandler.handleExecutor (again)

Note that the repository aspect sits inside the CQRS aspect. So at save time DomainEventContext is still there (not yet cleared) and the events can pick up context.

Debugging checklist

If events aren't going out, check in this order:

  1. Does the repository type name contain Repository? (either interface or implementation matching is enough)
  2. Is the method name save / update / delete / operate?
  3. Is the first argument an AbstractAggregateRoot subclass?
  4. Did the aggregate root's business method call addPendingEvent? (protected — outsiders can't)
  5. Was soda.event.bus-type misconfigured as spring?
  6. Does the EventBus bean exist? Search the startup log for [SimpleEventBusAutoConfiguration] Creating SimpleEventBus instance
  7. Did the transaction roll back? afterCommit runs only on successful commit.

These startup log lines confirm registration directly:

text
[SimpleCommandBus] Constructor called, handlers size: 1
[SimpleCommandBus] Registered handler for command: com.example.demo.CreateProducerCommand
[SimpleEventBus] Constructor called, handlers size: 1
[SimpleEventBus] Registered 1 event handler types
[CqrsAroundHandler] Bean created!

Next steps