Extension Points

Every key bean in soda carries @ConditionalOnMissingBean. That means: define a bean with the same name yourself, and the framework's steps aside. This page shows how to use each hook.

Extension points at a glance:

What to replace	Define this bean	Package
Event bus	`EventBus`	`com.hibuka.soda.cqrs.event`
Event serializer	`EventSerializer`	`com.hibuka.soda.event`
Command bus	`CommandBus`	`com.hibuka.soda.cqrs.command`
Query bus	`QueryBus`	`com.hibuka.soda.cqrs.query`
Facade	`BusFacade`	`com.hibuka.soda.bus.facade`
Idempotency service	`IdempotencyService`	`com.hibuka.soda.event.redis.service`
Write repository	Implement `WriteRepository<T>`	`com.hibuka.soda.domain.repository`

Custom event bus

The EventBus interface has just 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;
}

To swap in your own implementation (hook up Kafka or Pulsar, or just write audit logs), define a bean:

java

package com.example.demo.infrastructure.eventbus;

import com.hibuka.soda.cqrs.event.EventBus;
import com.hibuka.soda.cqrs.event.EventHandler;
import com.hibuka.soda.domain.event.DomainEvent;
import com.hibuka.soda.foundation.error.BaseException;
import org.springframework.stereotype.Component;

@Component
public class KafkaEventBus implements EventBus {

    @Override
    public void publish(DomainEvent event) throws BaseException {
        // Publish to Kafka
    }

    @Override
    public void subscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException {
        // Register subscription
    }

    @Override
    public void unsubscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException {
        // Unsubscribe
    }
}

Because the beans in both SimpleEventBusAutoConfiguration and RedisEventBusAutoConfiguration are @ConditionalOnMissingBean, yours takes effect automatically and the framework no longer creates SimpleEventBus.

Tip

But watch out: redisStreamEventBus in RedisEventBusAutoConfiguration is @Primary. If your KafkaEventBus coexists with it, injecting EventBus gets you the @Primary one. For your bean to win, either don't pull in the soda-event-redis module, put @Primary on your bean instead, or inject precisely with @Qualifier.

Three obligations of a custom bus:

It must be able to deserialize events. A cross-process bus needs to know the type and content of each event.
It must respect transaction boundaries. Follow the pattern in RedisStreamEventBus.publish: register afterCommit when a transaction is active, send immediately when there is none.
It must be fault-tolerant. Follow the pattern in RepositoryEventAspect.publishEvents: a failed publish for one event only gets logged and does not affect the other events.

Custom event serializer

com.hibuka.soda.event.EventSerializer in soda-core-base:

java

public interface EventSerializer {
    String toJson(Object obj);

    <T> T fromJson(String json, Class<T> type);

    class SerializationException extends RuntimeException {
        public SerializationException(String message, Throwable cause) {
            super(message, cause);
        }
    }
}

This abstraction exists to decouple event serialization from any specific JSON library. The default implementation is JacksonEventSerializer (in soda-event-redis), which calls ObjectMapper's writeValueAsString / readValue through reflection.

To swap in fastjson, Gson, Kryo, or your own binary protocol:

java

package com.example.demo.infrastructure.event;

import com.alibaba.fastjson.JSON;
import com.hibuka.soda.event.EventSerializer;
import org.springframework.stereotype.Component;

@Component
public class FastjsonEventSerializer implements EventSerializer {

    @Override
    public String toJson(Object obj) {
        try {
            return JSON.toJSONString(obj);
        } catch (Exception e) {
            throw new SerializationException("Failed to serialize event", e);
        }
    }

    @Override
    public <T> T fromJson(String json, Class<T> type) {
        try {
            return JSON.parseObject(json, type);
        } catch (Exception e) {
            throw new SerializationException("Failed to deserialize event", e);
        }
    }
}

java

@Bean
@ConditionalOnMissingBean
public EventSerializer eventSerializer(ApplicationContext applicationContext) {
    return new JacksonEventSerializer(applicationContext);
}

Warning

When you swap serializers, make sure both ends use the same format. If the producer uses fastjson and the consumer Jackson, cross-service deserialization will fail. The @JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, property = "@class") on event classes is there for Jackson — with any other library, handling of the @class field is up to you.

How dual Jackson support works

This is the biggest technical change in 17.0.1: one jar supports both Spring Boot 3.2.x (Jackson 2) and Spring Boot 4.x (Jackson 3).

The root of the problem: Jackson 3 moved its packages from com.fasterxml.jackson.* to tools.jackson.*. If soda-core imported ObjectMapper directly, supporting both versions at once would be impossible.

The solution is reflection. JacksonEventSerializer contains not a single Jackson import:

java

public class JacksonEventSerializer implements EventSerializer {

    private final ApplicationContext applicationContext;
    private volatile Object objectMapper;
    private volatile Method writeValueAsString;
    private volatile Method readValue;

    public JacksonEventSerializer(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    private Object getObjectMapper() {
        if (objectMapper == null) {
            synchronized (this) {
                if (objectMapper == null) {
                    objectMapper = resolveObjectMapper();
                    try {
                        this.writeValueAsString = objectMapper.getClass().getMethod("writeValueAsString", Object.class);
                        this.readValue = objectMapper.getClass().getMethod("readValue", String.class, Class.class);
                    } catch (NoSuchMethodException e) {
                        throw new SerializationException("ObjectMapper missing required methods", e);
                    }
                }
            }
        }
        return objectMapper;
    }
    ...
}

The ObjectMapper is looked up by name from the Spring container:

java

private Object resolveObjectMapper() {
    String[] names = {"jacksonObjectMapper", "jacksonJsonMapper", "jsonMapper", "objectMapper"};
    for (String name : names) {
        try {
            if (applicationContext.containsBean(name)) {
                return applicationContext.getBean(name);
            }
        } catch (Exception ignored) {}
    }
    // Fallback: scan by class name
    for (String beanName : applicationContext.getBeanDefinitionNames()) {
        try {
            Object bean = applicationContext.getBean(beanName);
            String className = bean.getClass().getName();
            if (className.contains("ObjectMapper") || className.contains("JsonMapper")) {
                return bean;
            }
        } catch (Exception ignored) {}
    }
    throw new IllegalStateException("No ObjectMapper bean found. Ensure Jackson is on the classpath.");
}

Lookup order:

By name: jacksonObjectMapper → jacksonJsonMapper → jsonMapper → objectMapper
If no name matches, walk every bean and treat one whose class name contains ObjectMapper or JsonMapper as a hit
If neither finds anything, throw IllegalStateException("No ObjectMapper bean found. Ensure Jackson is on the classpath.")

The ObjectMapper is lazy — resolved at first serialization, not when the bean is created. That gives Spring Boot's Jackson auto-configuration time to finish.

Tip

Since the fallback matches by class name, you really want exactly one ObjectMapper in your container. With several (say one for web, one for Redis), which one gets picked is undefined — in that case define an EventSerializer bean explicitly, or give the intended ObjectMapper one of the names listed above.

Note

soda-core-base does not depend on Jackson. Jackson is used only indirectly, through the EventSerializer abstraction, inside soda-event-redis. So a project running just the in-process event bus is perfectly fine without Jackson on the classpath.

Custom write repository

The write repository needs no "extension point" — it is an interface meant for you to implement in the first place:

java

public interface WriteRepository<T extends AbstractAggregateRoot> {
    boolean save(T aggregate) throws BaseException;
    boolean update(T aggregate) throws BaseException;
    boolean delete(T aggregate) throws BaseException;
    boolean operate(T aggregate, int operationType) throws BaseException;
}

Implement it and hand the implementation class to Spring. Two rules to keep: the repository type (interface or implementation class) must have Repository in its name, and the method names must be these four — that is exactly what RepositoryEventAspect's pointcut *..*Repository+ matches. The implementation class, as a subtype of the interface, gets intercepted no matter what it is called.

Custom idempotency service

Idempotency is a feature of the Redis event bus. IdempotencyService is defined in com.hibuka.soda.event.redis.service:

java

public interface IdempotencyService {
    enum ProcessingStatus {
        PROCESSING, SUCCESS, FAILED
    }

    boolean beginProcessing(String eventId);

    void markAsSuccess(String eventId, Map<String, Object> handlerResults);

    void markAsFailed(String eventId, String error);

    ProcessingStatus getStatus(String eventId);

    void cleanupExpiredStatus();
}

The default implementation is RedisIdempotencyServiceImpl, which stores state in a Redis Hash:

Hash field	Content
`status`	`PROCESSING` / `SUCCESS` / `FAILED`
`processedAt`	Processing timestamp
`handlerResults`	Per-`EventHandler` results
`error`	Failure reason

The Redis key is idempotency.redisKeyPrefix + ":" + eventId, with default prefix soda-events-idempotency and default expiry of 24 hours.

Note

RedisStreamEventBus instantiates RedisIdempotencyServiceImpl directly instead of injecting it from the container:

java

this.idempotencyService = new RedisIdempotencyServiceImpl(redisTemplate, idempotencyProperties);

So defining an IdempotencyService bean does not replace it automatically. To swap the implementation you must subclass RedisStreamEventBus yourself, or implement EventBus yourself.

The idempotency logic lives on the consumer side:

java

IdempotencyService.ProcessingStatus status = idempotencyService.getStatus(eventId);
if (status == IdempotencyService.ProcessingStatus.SUCCESS) {
    // Already handled successfully, skip
} else if (status == IdempotencyService.ProcessingStatus.PROCESSING) {
    // Currently processing, skip
}

The rules of beginProcessing:

Status SUCCESS → returns false, skip
Status PROCESSING → returns false, skip
Status FAILED → returns true, retry allowed
No record → writes PROCESSING with an expiry, returns true

Warning

Idempotency is off by default: soda.event.redis.stream.idempotency.enabled defaults to false. Turn it on for cross-service scenarios — Redis Stream's at-least-once delivery semantics mean the same event may be consumed more than once.

The async thread pool

soda-core's async thread pool is configured through AsyncConfig, with prefix scoda.ddd.cqrs.async:

java

@ConfigurationProperties(prefix = "scoda.ddd.cqrs.async")
public class AsyncConfig {
    private int corePoolSize = 8;
    private int maxPoolSize = 16;
    private int queueCapacity = 100;
    private String threadNamePrefix = "cqrs-async-";
}

Warning

The prefix is scoda.ddd.cqrs.async, not soda.ddd.cqrs.async. README-CN says the latter — that is wrong, and it has no effect. The correct form:

yaml

scoda:
  ddd:
    cqrs:
      async:
        core-pool-size: 16
        max-pool-size: 32
        queue-capacity: 200
        thread-name-prefix: "cqrs-async-"

The pool itself is created in SpringBusAutoConfiguration:

java

@Bean("cqrsAsyncExecutor")
@ConditionalOnMissingBean(name = "cqrsAsyncExecutor")
public Executor cqrsAsyncExecutor(AsyncConfig asyncConfig) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(asyncConfig.getCorePoolSize());
    executor.setMaxPoolSize(asyncConfig.getMaxPoolSize());
    executor.setQueueCapacity(asyncConfig.getQueueCapacity());
    executor.setThreadNamePrefix(asyncConfig.getThreadNamePrefix());
    executor.setKeepAliveSeconds(60);
    executor.setRejectedExecutionHandler((r, executor1) -> {
        logger.warn("[BusAutoConfiguration] Thread pool is full, rejecting task: {}", r.getClass().getName());
        if (!executor1.isShutdown()) {
            try {
                r.run();
            } catch (Exception e) {
                logger.error("[BusAutoConfiguration] Error executing task in caller thread: {}", e.getMessage(), e);
            }
        }
    });
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(60);
    executor.initialize();
    return executor;
}

The rejection policy runs the task on the caller thread (the CallerRunsPolicy idea), so no task is lost — but the calling thread gets blocked. Keep this in mind in production: when the queue fills up, web request threads stall with it.

Tip

To use your own thread pool, define an Executor bean named cqrsAsyncExecutor — @ConditionalOnMissingBean(name = "cqrsAsyncExecutor") makes the framework back off.

Custom BusFacade

BusFacade is @ConditionalOnMissingBean too. To change the send logic (add metrics, alter the recursion policy), subclass it:

java

package com.example.demo.infrastructure.bus;

import com.hibuka.soda.bus.facade.BusFacade;
import com.hibuka.soda.cqrs.command.CommandBus;
import com.hibuka.soda.cqrs.query.QueryBus;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;

@Component
@Primary
public class MetricsBusFacade extends BusFacade {
    public MetricsBusFacade(CommandBus commandBus, QueryBus queryBus, Executor cqrsAsyncExecutor) {
        super(commandBus, queryBus, cqrsAsyncExecutor);
    }

    @Override
    public <C extends Command<R>, R> R sendTransactCommand(C command) throws BaseException {
        // Add metrics, then call super
        return super.sendTransactCommand(command);
    }
}

Warning

When subclassing BusFacade: its constructor is generated by @RequiredArgsConstructor with parameters (CommandBus, QueryBus, Executor) — the order cannot change, and the third argument must be the thread pool carrying @Qualifier("cqrsAsyncExecutor").