Extending KETE

Add custom destinations, serializers, matchers, and certificate loaders to KETE.

---

Adding a New Destination

1. Create the Config Class

Each destination needs a configuration class extending DestinationConfig:

package io.github.fortunen.kete.destinations.mydestination;

import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import io.github.fortunen.kete.DestinationConfig;

@Data
@NoArgsConstructor
@EqualsAndHashCode(callSuper = true)
public class MyDestinationConfig extends DestinationConfig {

    private String host;
    private int port = 8080;
    private String topic;

    // Lombok generates getters/setters
}

2. Create the Destination Class

package io.github.fortunen.kete.destinations.mydestination;

import io.github.fortunen.kete.Component;
import io.github.fortunen.kete.Destination;
import io.github.fortunen.kete.EventMessage;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor(force = true)
@EqualsAndHashCode(callSuper = true)
@Component(name = "my-destination")
public class MyDestination extends Destination<MyDestinationConfig> {

    private MyClient client;

    @Override
    protected void doInitialize() {
        // config is already set and validated
        client = new MyClient(config.getHost(), config.getPort());
        client.connect();

        // tls is available via config.getTls() if configured
        if (config.getTls().isEnabled()) {
            var sslContext = config.getTls().getKeyStoreAndTrustStoreSSLContext();
            client.enableTls(sslContext);
        }
    }

    @Override
    protected void doSend(EventMessage message) {
        client.publish(config.getTopic(), message.eventBody());
    }

    @Override
    public boolean isHealthy() {
        return client != null && client.isConnected();   // cheap state check, no I/O — used by the pool
    }

    @Override
    public void close() {
        if (client != null) {
            client.disconnect();
        }
    }
}

3. Key Points

Aspect	Details
@Component(name = "xxx")	The name becomes destination.kind value
Scope	Destinations are TRANSIENT (a new instance for every pooled object, up to pool.max-total per route)
config field	Automatically populated before doInitialize() is called
doInitialize()	Called each time the pool creates an instance; set up connections here
doSend(EventMessage)	Called for each event, send the message here
isHealthy()	Cheap connection-state check used by the pool on borrow/return; must not perform I/O
close()	Called by the pool when the instance is destroyed (invalidated, evicted or pool closed)

4. Use the Destination

kete.routes.my-route.destination.kind=my-destination
kete.routes.my-route.destination.host=example.com
kete.routes.my-route.destination.port=9090
kete.routes.my-route.destination.topic=events

---

Adding a New Serializer

1. Create the Class

package io.github.fortunen.kete.serializers;

import io.github.fortunen.kete.Component;
import io.github.fortunen.kete.Serializer;
import org.keycloak.events.Event;
import org.keycloak.events.admin.AdminEvent;

@Component(name = "my-format", scope = Component.SINGLETON)
public class MySerializer extends Serializer {

    private String contentType = "application/my-format";

    @Override
    public byte[] serialize(Event event) {
        return formatEvent(event).getBytes();
    }

    @Override
    public byte[] serialize(AdminEvent adminEvent) {
        return formatAdminEvent(adminEvent).getBytes();
    }

    private String formatEvent(Event event) {
        // Your serialization logic
        return "...";
    }

    private String formatAdminEvent(AdminEvent adminEvent) {
        // Your serialization logic
        return "...";
    }
}

2. Key Points

Aspect	Details
Scope	SINGLETON (one instance shared across all routes)
contentType	Set the MIME type via field initialization
Thread-safe	Serializers must be thread-safe

3. Use the Serializer

kete.routes.my-route.serializer.kind=my-format

---

Adding a New Matcher

1. Create the Class

package io.github.fortunen.kete.matchers;

import io.github.fortunen.kete.Component;
import io.github.fortunen.kete.Matcher;

@Component(name = "my-matcher")
public class MyMatcher extends Matcher {

    private MyPatternEngine engine;

    @Override
    public void initialize() {
        // pattern field is inherited, contains the user's pattern
        this.engine = MyPatternEngine.compile(pattern);
    }

    @Override
    public boolean matches(String eventType) {
        return engine.matches(eventType);
    }
}

2. Key Points

Aspect	Details
pattern field	Inherited, contains the user's pattern
not field	Inherited, indicates if result should be negated
accept() method	Already implemented, calls matches() XOR not
Matching	Case-insensitive by convention

3. Use the Matcher

kete.routes.my-route.event-matchers.filter=my-matcher:PATTERN_HERE
kete.routes.my-route.event-matchers.exclude=my-matcher:not:PATTERN_HERE

---

Adding a New Certificate Loader

1. Create the Class

package io.github.fortunen.kete.certificateloaders;

import io.github.fortunen.kete.CertificateLoader;
import io.github.fortunen.kete.Component;
import java.security.KeyStore;

@Component(name = "my-loader")
public class MyCertificateLoader extends CertificateLoader {

    @Override
    public void initialize() {
        // Validate configuration properties
    }

    @Override
    @SneakyThrows
    public void loadKeyStore(KeyStore keyStore, char[] password) {
        // Load certificates/keys into the keyStore
        String source = configuration.getString("source");
        // ... parse and add entries to keyStore
    }
}

2. Key Points

Aspect	Details
configuration	Access loader-specific properties
keyStore parameter	Pre-initialized, add entries to it
password parameter	Use for private key entries

3. Use the Loader

kete.routes.my-route.destination.tls.trust-store.loader.kind=my-loader
kete.routes.my-route.destination.tls.trust-store.loader.source=/path/to/source

---

Adding a Content Encoding or Content Transfer Encoding

Content encodings (ContentEncoding: encode(byte[]) / decode(byte[])) and content transfer encodings (ContentTransferEncoding, same two methods) are components as well:

@Component(name = "my-encoding")
public class MyContentEncoding extends ContentEncoding {

    @Override
    public byte[] encode(byte[] bytes) { ... }

    @Override
    public byte[] decode(byte[] bytes) { ... }
}

The component name becomes the value of destination.content-encoding / destination.content-transfer-encoding. Encodings are only applied by the destinations that call encodePayload() (see Content Encodings).

Component Discovery

Components are discovered automatically at startup:

1. IocUtils scans the io.github.fortunen.kete package
2. Classes annotated with @Component are registered
3. SINGLETON components are instantiated immediately
4. TRANSIENT components are instantiated on demand

No additional registration is required—just add the @Component annotation.

---

EventMessage Reference

The EventMessage record carries event data through the pipeline:

public record EventMessage(
    String realm,           // Keycloak realm name
    String eventId,         // Unique event identifier
    byte[] eventBody,       // Serialized event content
    String eventType,       // Event type (e.g., LOGIN, LOGOUT)
    String contentType,     // MIME type from serializer
    String resourceType,    // Admin event resource type
    String kind,            // "EVENT" or "ADMIN_EVENT"
    String operationType,   // Admin event operation
    String result           // Event result (e.g., SUCCESS, ERROR)
) {
    // Helper methods — case conversions (cached, except eventId)
    String kindLowerCase();             // "event" or "admin_event"
    String kindUpperCase();             // "EVENT" or "ADMIN_EVENT"
    String kindKebabCase();             // "event" or "admin-event"
    String kindPascalCase();            // "Event" or "AdminEvent"
    String kindCamelCase();             // "event" or "adminEvent"
    String eventIdLowerCase();          // Lowercase event ID (uncached)
    String eventIdUpperCase();          // Uppercase event ID (uncached)
    String realmLowerCase();            // "master" or "my_realm"
    String realmUpperCase();            // "MASTER" or "MY_REALM"
    String realmKebabCase();            // "master" or "my-realm"
    String realmPascalCase();           // "Master" or "MyRealm"
    String realmCamelCase();            // "master" or "myRealm"
    String eventTypeLowerCase();        // "login_error"
    String eventTypeUpperCase();        // "LOGIN_ERROR"
    String eventTypeKebabCase();        // "login-error"
    String eventTypePascalCase();       // "LoginError"
    String eventTypeCamelCase();        // "loginError"
    String resourceTypeLowerCase();     // "realm_role"
    String resourceTypeUpperCase();     // "REALM_ROLE"
    String resourceTypeKebabCase();     // "realm-role"
    String resourceTypePascalCase();    // "RealmRole"
    String resourceTypeCamelCase();     // "realmRole"
    String operationTypeLowerCase();    // "create"
    String operationTypeUpperCase();    // "CREATE"
    String operationTypeKebabCase();    // "create"
    String operationTypePascalCase();   // "Create"
    String operationTypeCamelCase();    // "create"
    String resultLowerCase();           // "success"
    String resultUpperCase();           // "SUCCESS"
    String resultKebabCase();           // "success"
    String resultPascalCase();          // "Success"
    String resultCamelCase();           // "success"
    byte[] kindBytes();                // UTF-8 bytes for headers
    byte[] eventTypeBytes();            // UTF-8 bytes for headers
    byte[] contentTypeBytes();          // UTF-8 bytes for headers
}

---

Testing New Components

Unit Tests

Follow the project's test patterns:

package io.github.fortunen.kete.destinations.mydestination;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;

import org.junit.jupiter.api.Test;

public class doInitializeTests {

    @Test
    public void shouldConnectSuccessfully() {

        // arrange

        var config = new MyDestinationConfig();
        config.setHost("localhost");
        config.setPort(8080);

        var destination = new MyDestination();
        destination.setConfig(config);

        // act

        destination.initialize();

        // assert

        assertThat(destination.getClient()).isNotNull();

        // cleanup

        destination.close();
    }

    @Test
    public void shouldThrowWhenHostMissing() {

        // arrange

        var config = new MyDestinationConfig();
        config.setPort(8080);
        // host not set

        var destination = new MyDestination();
        destination.setConfig(config);

        // act

        var thrown = catchThrowable(() -> destination.initialize());

        // assert

        assertThat(thrown)
            .isInstanceOf(IllegalStateException.class)
            .hasMessage("host is required");
    }
}

Integration Tests

Use Testcontainers for integration testing:

@Testcontainers
class MyDestinationIntegrationTests {

    @Container
    static GenericContainer<?> myService = new GenericContainer<>("my-service:latest")
        .withExposedPorts(8080);

    @Test
    void shouldSendToRealService() {
        var config = new MyDestinationConfig();
        config.setHost(myService.getHost());
        config.setPort(myService.getMappedPort(8080));

        var destination = new MyDestination();
        destination.setConfig(config);
        destination.initialize();

        // ... test with real service

        destination.close();
    }
}

---

See Also

• Component Scopes — SINGLETON vs TRANSIENT
• Architecture — System design overview
• Test Patterns — Testing conventions