Forem: Stephen Maina

JayNats: The Future of Message Brokers

Stephen Maina — Sun, 26 Oct 2025 18:00:42 +0000

Getting Started with JayNats: Zero-Copy Real-time CDC in Minutes

A tutorial-style blog post showcasing JayNats as a zero-copy, low-latency messaging broker with seamless CDC

If you're building real-time applications, you've probably dealt with messaging brokers and change data capture (CDC) separately. Kafka for events, Debezium for CDC, Redis for caching—the modern microservices architecture is a symphony of moving parts.

But what if I told you that you could have zero-copy, low-latency messaging with type-safe CDC in a single Java library?

Enter JayNats - a production-ready messaging broker built with Java's Foreign Function & Memory API, optimized for sub-microsecond latency with zero-copy operations. It just so happens to ship with CDC out of the box.

What is JayNats?

JayNats is, first and foremost, a zero-copy, low-latency messaging broker powered by Java's new FFM API. While it ships with CDC capabilities as a showcase feature, its core mission is providing the fastest, most efficient message passing in the JVM ecosystem.

Key Features

Zero-copy operations with FFM for unparalleled performance
Sub-microsecond latency for message processing
Built-in CDC with automatic schema discovery
Type-safe JOOQ Records - no JSON, no manual parsing
Simple, consistent API across all use cases
Production-ready with real-time demo at https://akilisha.com

Beyond CDC: JayNats as a Messaging Broker

Before diving into CDC, let's see JayNats in its primary role: zero-copy messaging. The beauty is the API remains consistent whether you're doing simple pub/sub or advanced CDC.

Event-Driven Architecture

import com.jaynats.stream.MessageBroker;
import com.jaynats.stream.SimpleMessageBroker;

MessageBroker broker = new SimpleMessageBroker();
broker.createTopic("user-events", 1).get();

// Publisher
broker.publish("user-events", "User registered: john@example.com".getBytes()).get();

// Subscribers
broker.subscribe("user-events", message -> {
    System.out.println("📧 " + new String(message.data()));
});

Microservices Communication

// Service A: Order Service
MessageBroker broker = new SimpleMessageBroker();
broker.createTopic("orders", 1).get();
broker.publish("orders", orderJson.getBytes()).get();

// Service B: Payment Service (in a different process)
broker.subscribe("orders", message -> {
    Order order = parseOrder(message.data());
    processPayment(order);
});

Pub/Sub Patterns

// Multiple subscribers to the same topic
broker.subscribe("notifications", message -> {
    emailService.send(new String(message.data()));
});

broker.subscribe("notifications", message -> {
    smsService.send(new String(message.data()));
});

// Publish once, fan out to all
broker.publish("notifications", "User logged in".getBytes()).get();

Real-Time Data Streaming

// High-frequency sensor data with zero-copy performance
broker.createTopic("sensor-data", 1).get();

void onSensorReading(double temperature) {
    byte[] data = serialize(temperature);
    broker.publish("sensor-data", data).get();  // Sub-microsecond latency!
}

broker.subscribe("sensor-data", message -> {
    double temp = deserialize(message.data());
    updateDashboard(temp);
});

Notice the pattern? Simple, consistent, powerful. The same API works for everything.

The Live Demo

Before we dive into the code, check out the live demo at https://akilisha.com. Connect your PostgreSQL database and watch changes appear in real-time with type-safe, zero-copy event processing!

Getting Started - It's Actually This Simple

The beauty of JayNats is in its simplicity. For CDC, you simply provide your JDBC credentials. That's it. Schema discovery, model generation, trigger creation, and type-safe event mapping are all handled automatically by the library.

Step 1: Add Dependencies

Gradle:

dependencies {
    implementation 'com.akilisha.oss:jaynats-stream:1.0.0'
    implementation 'com.akilisha.oss:jaynats-stream-cdc:1.0.0'
}

Maven:

<dependency>
    <groupId>com.akilisha.oss</groupId>
    <artifactId>jaynats-stream</artifactId>
    <version>1.0.0</version>
</dependency>
<dependency>
    <groupId>com.akilisha.oss</groupId>
    <artifactId>jaynats-stream-cdc</artifactId>
    <version>1.0.0</version>
</dependency>

Or clone the repository from GitHub to build locally.

Step 2: Write Some Code

import com.jaynats.stream.cdc.CDCManager;
import com.jaynats.stream.cdc.generated.tables.records.UsersRecord;

public class RealTimeApp {
    public static void main(String[] args) {
        // Just provide your database URL
        CDCManager cdc = new CDCManager(
            "jdbc:postgresql://localhost:5432/mydb",
            "output/models"
        );

        // Start CDC - the library handles EVERYTHING
        try {
            cdc.startCDC("users", "orders").get();
            System.out.println("✅ CDC started! Schema discovered, models generated, ready to go!");

            // Subscribe with type-safe JOOQ Records
            cdc.subscribeToTableTyped("users", UsersRecord.class, event -> {
                UsersRecord user = event.getData();  // ← Typed JOOQ Record!
                System.out.println("📧 Email: " + user.getEmail());
                System.out.println("👤 Name: " + user.getName());
                // Full IDE autocomplete, compile-time safety!
            });

            // Or use the flexible API for JSON access
            cdc.subscribeToTable("orders", event -> {
                String json = event.getJsonData();  // ← Raw JSON available too
                System.out.println("📦 Order: " + json);
            });

            // Keep running
            Thread.sleep(Long.MAX_VALUE);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

That's it! No manual triggers. No SQL setup. No schema management. Just JDBC credentials.

What Happens Under the Hood

When you call startCDC("users", "orders"), JayNats:

✅ Connects to your PostgreSQL database
✅ Discovers the schema for the specified tables
✅ Generates JOOQ model classes (UsersRecord, OrdersRecord, etc.)
✅ Creates database triggers automatically
✅ Sets up real-time LISTEN/NOTIFY connections
✅ Maps events to typed JOOQ Records

All transparently. Zero manual work.

Type Safety Out of the Box

The real magic is in the type safety. No more parsing JSON strings:

// Before (manual JSON parsing)
cdc.subscribeToTable("users", event -> {
    String json = event.getJsonData();
    JSONObject obj = new JSONObject(json);
    String email = obj.getString("email");  // 😰 Runtime error prone
});

// After (JOOQ Records)
cdc.subscribeToTableTyped("users", UsersRecord.class, event -> {
    UsersRecord user = event.getData();  // ← Strongly typed!
    String email = user.getEmail();      // ← IDE autocomplete!
    Integer id = user.getId();           // ← Compile-time type checking!
});  // 😎 Refactoring-safe, validation at compile time

Streaming to the Live Demo

Want to see your database changes on https://akilisha.com in real-time?

Quick Start

# Build the CDC CLI
./gradlew :jaynats-stream-cdc:build

# Run it
java -jar jaynats-stream-cdc/build/libs/jaynats-stream-cdc-1.0.0.jar \
  --url jdbc:postgresql://localhost:5432/mydb \
  --target https://akilisha.com

Visit https://akilisha.com, start the demo, and your database changes appear in real-time with zero-copy performance!

Why JayNats?

🚀 Zero-Copy Performance

Java FFM API for zero-copy memory operations
Sub-microsecond latency for message processing
Virtual threads for massive concurrency

🎯 Intelligent, Self-Aware

Automatic schema discovery - just provide credentials
Zero configuration - it figures out what to do
Type-safe by default - JOOQ Records everywhere

🏗️ Production-Ready

Single JVM process - no external dependencies
Modern Java 23 with latest features
Live demo at https://akilisha.com

🎨 Consistent API Across Use Cases

The same simple API works for everything:

// Standalone messaging
broker.publish("events", data);
broker.subscribe("events", handler);

// CDC (with zero-copy underneath)
cdc.startCDC("users", "orders");
cdc.subscribeToTableTyped("users", UsersRecord.class, handler);

Whether you're building event-driven architectures, microservices communication, or real-time database synchronization, JayNats provides the same intuitive, consistent experience. No learning curve when switching between use cases.

Advanced Use Cases

Real-time Analytics

cdc.subscribeToTableTyped("orders", OrdersRecord.class, event -> {
    OrdersRecord order = event.getData();  // ← Typed!
    updateDashboard(order.getAmount(), order.getCustomerId());
});

Event Sourcing

cdc.subscribeToAllChanges(event -> {
    eventStore.append(event);  // ← Type-safe event stream
});

Microservices Sync

cdc.subscribeToTableTyped("users", UsersRecord.class, event -> {
    UsersRecord user = event.getData();  // ← Zero JSON conversion!
    distributedCache.put(user.getId(), user);
});

The Philosophy

JayNats is intelligent and self-aware. You shouldn't have to:

❌ Manually create triggers
❌ Write SQL procedures
❌ Parse JSON strings
❌ Manage schema manually
❌ Learn different APIs for different use cases

You just provide credentials for CDC or create a broker for messaging. JayNats handles the rest, delivering typed, zero-copy events with sub-microsecond latency and a consistent, intuitive API across all use cases.

From simple pub/sub to complex CDC, the same beautiful simplicity throughout.

Next Steps

Try the live demo: https://akilisha.com
Check out the code: GitHub
Explore examples: Clone and check jaynats-examples module

JayNats: Zero-copy messaging with intelligent CDC. Just add credentials. 🚀

For the latest updates and contributions, visit GitHub

Java Signals and Effects

Stephen Maina — Fri, 18 Apr 2025 17:37:27 +0000

When developers talk about reactivity in either Java or JavaScript, from the surface level, it appears to be the same kind of conversation. The first place that the conversation may typically wander off to is perhaps streaming data using libraries like RXJS or RXJava (and more lately Project Reactor in Java).

The conversation may also wander off into asynchronous programming and yet retain the semblance of similarity, because both Java (threads) and JavaScript (the event loop) have a built-in model of asynchronous programming. But when you dig deeper, the similarities quickly begin to evaporate.

JavaScript explores other aspects of reactivity like rendering on the UI through observing state variables and reacting to their changes to render more efficiently. The equivalence of this fine level of granularity is missing in the Java universe.

This presentation is an attempt to narrow that gap and show that a different programming model has a solid place in Java. It sheds off the imperative model in favor of a reactive model which propagates through reacting to events (more accurately known as effects) happening in entities that have a relationship to each other. It's actually more accurate to call it a contemporary model of event-driven programming.

Observables done right

To tee off this presentation, consider a TodosList that contains Todo items. You wish to be able to react to the following events.

In any Todo item, when:

the title is changed
the completion status is toggled

In the TodosList, when:

a new item is added
an existing item is removed

The behaviors executed in response to these reactions are called effects, and they are set off when a signal, or more explicitly, an observed class property, is modified in the course of program execution.

Approach 1

Dialing it back to begin from first principles, here is a basic representation of the respective domain classes:

Domain class with basic fields

@Data
@AllArgsConstructor
public class Todo {

    private UUID id;
    private String title;
    private Boolean completed;
}

Domain class with collection field

public class TodosList {

    private final Collection<Todo> todos = new ArrayList<>();

    public Todo add(String title){
        Todo todo = new Todo(UUID.randomUUID(), title, false);
        todos.add(todo);
        return todo;
    }

    public void update(String id, String title){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.setTitle(title);
    }

    public void toggle(String id){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.setCompleted(!todo.getCompleted());
    }

    public void delete(String id){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todos.remove(todo);
    }
}

Observing changes to the state of an object when the events of interest described above are triggered can be achieved through different techniques, which I will review here.

Approach 1 is a basic solution that requires explicitly implementing some kind of observation mechanism, like a Listener.

public interface Listener {

    void onTitleChanged(Todo todo);

    void onCompletionChanged(Todo todo);

    void onItemAdded(Todo entity, Collection<Todo> todos);

    void onItemRemoved(Todo entity, Collection<Todo> todos);
}

Then a concrete implementation of the listener would execute the intended behavior when the events of interest are fired. Different implementations of the interface should be used to model different behavior. Below is one such implementation which only reacts to the event happening by printing its details to the console.

public class BasicListener implements Listener {

    @Override
    public void onTitleChanged(Todo todo) {
        System.out.printf("Task title changed to %s\n", todo.getTitle());
    }

    @Override
    public void onCompletionChanged(Todo todo) {
        System.out.printf("Task completion changed to %s\n", todo.getCompleted());
    }

    @Override
    public void onItemAdded(Todo entity, Collection<Todo> todos) {
        System.out.printf("Event: add, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }

    @Override
    public void onItemRemoved(Todo entity, Collection<Todo> todos) {
        System.out.printf("Event: remove, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }
}

These two classes (observable and listener implementation) represent functionality that needs to be woven together in some way, to be able to react to state changes. The easiest (and unfortunately pretty invasive) way is to add statements in the TodosList object to invoke methods in the BasicListener when the events of interest are happening. The updated TodosList would therefore look something like this.

public class TodosList {

    private final Collection<Todo> todos = new ArrayList<>();
    private final Listener listener = new BasicListener(); // require listener to accept events

    public Todo add(String title){
        Todo todo = new Todo(UUID.randomUUID(), title, false);
        todos.add(todo);
        listener.onItemAdded(todo, todos); // explicitly fire listener function
        return todo;
    }

    public void update(String id, String title){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.setTitle(title);
        listener.onTitleChanged(todo); // explicitly fire listener function
    }

    public void toggle(String id){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.setCompleted(!todo.getCompleted());
        listener.onCompletionChanged(todo); // explicitly fire listener function
    }

    public void delete(String id){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todos.remove(todo);
        listener.onItemRemoved(todo, todos); // explicitly fire listener function
    }
}

A main class would then be used to set off the orchestration of events as shown below

public class Main {

    public static void main(String[] args) {
        TodosList list = new TodosList();

        Todo t1 = list.add("wake up");
        Todo t2 = list.add("make breakfast");
        Todo t3 = list.add("watch tv");

        list.update(t2.getId().toString(), "work out");
        list.toggle(t1.getId().toString());
        list.delete(t3.getId().toString());
    }
}

Putting it all together, the main class shown above should certainly do a decent job of capturing all the targeted events and executing the prescribed effects.

If multiple listener implementations need to be invoked when these state changes happen, it would require having a collection of such listeners and then calling them all sequentially to dispatch the event's data.

public class AnotherListener implements Listener {

    @Override
    public void onTitleChanged(Todo todo) {
        System.out.printf("[**] Task title changed to %s\n", todo.getTitle());
    }

    @Override
    public void onCompletionChanged(Todo todo) {
        System.out.printf("[**] Task completion changed to %s\n", todo.getCompleted());
    }

    @Override
    public void onItemAdded(Todo entity, Collection<Todo> todos) {
        System.out.printf("[**] Event: add, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }

    @Override
    public void onItemRemoved(Todo entity, Collection<Todo> todos) {
        System.out.printf("[**] Event: remove, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }
}

The TodosList would now need to save a reference to the subscribers in a collection, and invoke them all when an event is patched.

public class TodosList {

    private final Collection<Todo> todos = new ArrayList<>();
    private final Collection<Listener> listeners = new LinkedList<>(); // use a collection to hold multiple listeners

    // convenience method to register listener
    public void addListener(Listener listener) {
        this.listeners.add(listener);
    }

    // convenience method to unregister listener
    public void removeListener(Listener listener) {
        this.listeners.remove(listener);
    }

    public Todo add(String title){
        // omitted for brevity
        listeners.forEach(l -> l.onItemAdded(todo, todos));
        return todo;
    }

    public void update(String id, String title){
        // omitted for brevity
        listeners.forEach(l -> l.onTitleChanged(todo));
    }

    public void toggle(String id){
        // omitted for brevity
        listeners.forEach(l -> l.onCompletionChanged(todo));
    }

    public void delete(String id){
        // omitted for brevity
        listeners.forEach(l -> l.onItemRemoved(todo, todos));
    }
}

Lastly, the main class would then be used to register (and perhaps even to unregister) listeners

public class Main {

    public static void main(String[] args) {
        TodosList list = new TodosList();

        // register listeners
        list.addListener(new BasicListener());
        list.addListener(new AnotherListener());

        // continue in the same way as before
        Todo t1 = list.add("wake up");
        Todo t2 = list.add("make breakfast");
        Todo t3 = list.add("watch tv");

        list.update(t2.getId().toString(), "work out");
        list.toggle(t1.getId().toString());
        list.delete(t3.getId().toString());
    }
}

The main problem with approach 1 is that the concerns of the listener (registering, unregistering and dispatching) must be manually woven into the observable, which increases the surface area of tripping up and therefore requires careful design and extensive testing. Even more significant is that the listener implementation is very tightly coupled to the Observable, and hence impossible to reuse in any other situation without major modifications.

Approach 2

A slightly more idiomatic approach would be to take advantage of Java's built-in Observer/Observable framework to offload much of the observing concerns, like registering and unregistering of listeners, to the underlying framework and instead focus on the behavior produced after events are triggered (that is, the effects).

This approach is just as intrusive as approach 1, and has actually been deprecated since java 9, and therefore I would not recommend it to anyone, but it's certainly worth examining.

@Getter
@AllArgsConstructor
public class Todo extends Observable {

    @Setter
    private UUID id;
    private String title;
    private Boolean completed;

    public void setTitle(String title) {
        this.title = title;
        setChanged();           // mark the field as dirty
        notifyObservers(this);  // inform listeners to do their thing
    }

    public void setCompleted(Boolean completed) {
        this.completed = completed;
        setChanged();           // mark the field as dirty
        notifyObservers(this);  // inform listeners to do their thing
    }
}

The setters in an Observable need to be instrumented to notify observers of some change in state. The existing Listener implementations can be repurposed into Observers by implementing Java's own Observer interface.

public class BasicListener implements Listener, Observer {

    @Override
    public void onTitleChanged(Todo todo) {
        System.out.printf("Task title changed to %s\n", todo.getTitle());
    }

    @Override
    public void onCompletionChanged(Todo todo) {
        System.out.printf("Task completion changed to %s\n", todo.getCompleted());
    }

    @Override
    public void onItemAdded(Todo entity, Collection<Todo> todos) {
        System.out.printf("Event: add, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }

    @Override
    public void onItemRemoved(Todo entity, Collection<Todo> todos) {
        System.out.printf("Event: remove, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }

    @Override
    public void update(Observable obj, Object arg) {
        if (obj instanceof Todo todo) {
            System.out.println("[Observer] received event -> todo: " + todo);
        }
        if (obj instanceof TodosList list) {
            System.out.println("[Observer] received event -> todos: " + list);
        }
    }
}

The second Observer would take similar modifications to the ones made in the first one.

public class AnotherListener implements Listener, Observer {

    @Override
    public void onTitleChanged(Todo todo) {
        System.out.printf("[**] Task title changed to %s\n", todo.getTitle());
    }

    @Override
    public void onCompletionChanged(Todo todo) {
        System.out.printf("[**] Task completion changed to %s\n", todo.getCompleted());
    }

    @Override
    public void onItemAdded(Todo entity, Collection<Todo> todos) {
        System.out.printf("[**] Event: add, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }

    @Override
    public void onItemRemoved(Todo entity, Collection<Todo> todos) {
        System.out.printf("[**] Event: remove, entity: %s\n", entity);
        todos.forEach(System.out::println);
    }

    @Override
    public void update(Observable obj, Object arg) {
        if (obj instanceof Todo todo) {
            System.out.println("[**Observer**] received event -> todo: " + todo);
        }
        if (obj instanceof TodosList list) {
            System.out.println("[**Observer**] received event -> todos: " + list);
        }
    }
}

The fact that the notifyObservers(obj) in the Observable takes just one argument of Object type makes it difficult to be expressive when using this approach. It becomes tricky to detect what attributes changed, and in which Observable these attributes were, when the change event is received over on the Listener side.

public class TodosList extends Observable {

    private final Collection<Todo> todos = new ArrayList<>();

    public Todo add(String title){
        Todo todo = new Todo(UUID.randomUUID(), title, false);
        todos.add(todo);
        setChanged();               // mark the field as dirty
        notifyObservers(todos);     // inform listeners to do their thing
        return todo;
    }

    public void update(String id, String title){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.setTitle(title);
    }

    public void toggle(String id){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.setCompleted(!todo.getCompleted());
    }

    public void delete(String id){
        Todo todo = todos.stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todos.remove(todo);
        setChanged();               // mark the field as dirty
        notifyObservers(this);      // inform listeners to do their thing
    }
}

The main class now changes pretty dramatically since the Observers need to be registered with each Observable party.

public class Main {

    public static void main(String[] args) {
        TodosList list = new TodosList();

        BasicListener basic = new BasicListener();
        AnotherListener another = new AnotherListener();
        // register listeners
        list.addObserver(basic);
        list.addObserver(another);

        Todo t1 = list.add("wake up");
        // register listeners
        list.addObserver(basic);
        list.addObserver(another);

        Todo t2 = list.add("make breakfast");
        // register listeners
        list.addObserver(basic);
        list.addObserver(another);

        Todo t3 = list.add("watch tv");
        // register listeners
        list.addObserver(basic);
        list.addObserver(another);

        // proceed in the usual manner
        list.update(t2.getId().toString(), "work out");
        list.toggle(t1.getId().toString());
        list.delete(t3.getId().toString());
    }
}

As mentioned earlier, this approach may have been cutting-edge back in its days, but in today's technology landscape, those days are far in the rearview mirror. It's certainly an improvement over approach 1 in that the observing responsibility is delegated to the underlying framework, but it lacks the critical versatility of reusability. It's therefore not easy to reuse without a lot of customization or some major refactoring.

I have skipped the details of demultiplexing events arriving in the void update(Observable obj, Objects arg) methods of the Observers because it can get unnecessarily complex detecting what attributes changed, so that the routing may be dispatched to the correct Listener methods.

Approach 3

So what else is out there that will offer Observability, but without the difficulties pointed out in approaches 1 & 2? Enter Signals. This is a concept that I have used extensively in the JavaScript ecosystem, and its non-existence in the Java universe is pretty saddening.

This approach certainly narrows that gap. The Signals library discussed here can be accessed as a maven artifact using the coordinates given below.

# using maven
<dependency>
    <groupId>com.akilisha.oss</groupId>
    <artifactId>signals</artifactId>
    <version>0.0.1</version>
</dependency>

# using gradle
implementation("com.akilisha.oss:signals:0.0.1")

Signals uses the concept of instrumenting attributes that need to be observed and the registration of listeners is then implicitly achieved at the most opportune point in time by literary accessing these observed attributes, for example during the construction phase of an Observable or in a separate factory method.

Allow me illustrate further because that explanation was certainly not exhaustive. The Todo class in this case clearly shows which attributes are candidates for observation - title and completed.

@Getter
@AllArgsConstructor
public class Todo {

    private final Signal<String> title = Signals.signal("");
    private final Signal<Boolean> completed = Signals.signal(false);
    @Setter
    private UUID id;

    public Todo(String title) {
        this(UUID.randomUUID(), title, false);
    }

    public Todo(UUID id, String title, Boolean completed) {
        this.id = id;
        this.title.value(title);
        this.completed.value(completed);
    }

    @Override
    public String toString() {
        return "Todo{" +
                "title=" + title.value() +
                ", completed=" + completed.value() +
                ", id=" + id +
                '}';
    }
}

It's always convenient in the majority of cases to work with DTOs (Java Records or POJOs) to complement domain classes that are instrumented with Signal attributes. Although not used in this presentation, TodoItem is nonetheless an example of such a data transfer object. They are a totally optional but usually very convenient.

public record TodoItem (UUID id, String title, Boolean completed){

    public TodoItem(String title){
        this(UUID.randomUUID(), title, false);
    }
}

Now instead of explicitly implementing Listener interfaces, the effects of changes to the title and completed attributes of a Todo class can be captured during construction of the Todo objects in a factory method. Each call to the .observe() method will return a Subscription object which can be stored and then used later on to cancel the captured effect (similar to unsubscribing a listener). In this presentation, I will ignore the subscriptions so that I may focus on demonstrating effects.

@Getter
@AllArgsConstructor
public class Todo {

    // code omitted from brevity

    public static Todo from(String title){
        Todo todo = new Todo(title);
        // observe title attribute - multiple Observer effects can be captured here
        Signals.observe(() -> System.out.printf("Task title changed to %s\n", todo.getTitle().get()));
        // observe completed attribute - multiple Observer effects can be captured here
        Signals.observe(() -> System.out.printf("Task completion changed to %s\n", todo.getCompleted().get()));
        return todo;
    }

    @Override
    public String toString() {
        // omitted for brevity
    }
}

When observing Collection or Map class attributes, the raw values are wrapped in either SignalCollection and SignalDictionary classes respectively because they have unique distinguishing characteristics that need to be handled differently.

In this case, the TodosList needs its todos Collection attribute to be Observable.

public class TodosList {

    private final SignalCollection<Todo> todos = Signals.signal(new ArrayList<>());

    public Todo add(String title){
        // using factory method to create Todo object
        Todo todo = Todo.from(title);
        todos.add(todo);
        return todo;
    }

    public void update(String id, String title){
        Todo todo = todos.value().stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.getTitle().set(title);
    }

    public void toggle(String id){
        Todo todo = todos.value().stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todo.getCompleted().set(!todo.getCompleted().value());
    }

    public void delete(String id){
        Todo todo = todos.value().stream().filter(t -> t.getId().toString().equals(id)).findFirst()
                .orElseThrow(() -> new RuntimeException("no task found with matching id"));
        todos.remove(todo);
    }
}

The magic sauce is in the choice of methods in the Signal object used to access its raw value. There are three categories of values that are Observable.

Scalar (anything that is neither a Collection nor a Map)
Collection (Lists, Sets, etc)
Dictionary (Map)

For all Scalar values the .value() and .value(arg) methods are used to access and set respectively the underlying values without triggering effects. The .get() and .set(arg) methods however will register and trigger effects respectively. An effect is the behavior triggered when an Observable attribute is changed.

For Collection values, the .value() method is used to access the underlying Collection value without triggering effects. The .get(), .forEach() and .iterator() methods will register effects. The .set(arg) and .value(arg) methods don't exists here since they serve no useful purpose in this case. So to trigger effects, only three method are currently instrumented for that purpose in the SignalCollection - add, addAll and remove.

For Dictionary values, the .value() method is equally used to access the underlying Map value without triggering effects. The .get(), .forEach() and .iterator() methods will register effects. The .set(arg) and .value(arg) methods don't exists here since they serve no useful purpose in this case. So to trigger effects, only three method are currently instrumented for that purpose in the SignalDictionary - put, putAll and remove.

When observing a Collection or a Map, the Signals.observe() method takes different arguments to accommodate the structural differences in these categories of classes. In this case, the TodosList registers an effect because todos.get() method got accessed in the constructor, and the registered handler receives an event name and the affected entity as parameters. The event name represents the name of the method which triggered the effect.

public class TodosList {

    private final SignalCollection<Todo> todos = Signals.signal(new ArrayList<>());

    public TodosList() {
        Signals.observe((event, entity) -> {
            switch (event) {
                case "add", "remove" -> System.out.printf("Event: %s, entity: %s\n", event, entity);
                default -> System.out.printf("todos size: %d\n", todos.get().size());
            }
        });
    }

    // code omitted for brevity
}

To reiterate what just happened, accessing todos.get() causes the registration of an effect function to happen, and this effect will subsequently be triggered by any add() or remove() invocations on the SignalCollection.

The main class will now look vastly cleaner than the previous times.

public class Main {

    public static void main(String[] args) {
        TodosList list = new TodosList();

        // continue as normal
        Todo t1 = list.add("wake up");
        Todo t2 = list.add("make breakfast");
        Todo t3 = list.add("watch tv");

        list.update(t2.getId().toString(), "work out");
        list.toggle(t1.getId().toString());
        list.delete(t3.getId().toString());
    }
}

And the output produced will tell the whole story of what is happening when the TodosList and TodoItems are getting updated in the main method above.

todos size: 0
Task title changed to wake up
Task completion changed to false
Event: add, entity: Todo{title=wake up, completed=false, id=4b2e720e-5510-4f35-bd13-4925ff6c6f57}
Task title changed to make breakfast
Task completion changed to false
Event: add, entity: Todo{title=make breakfast, completed=false, id=8be14779-0ec9-44c4-aa94-572d2d21aac0}
Task title changed to watch tv
Task completion changed to false
Event: add, entity: Todo{title=watch tv, completed=false, id=bd665225-8dba-421c-91d6-0b6fb78f5f75}
Task title changed to work out
Task completion changed to true
Event: remove, entity: Todo{title=watch tv, completed=false, id=bd665225-8dba-421c-91d6-0b6fb78f5f75}
The source code for the above example can be viewed in this gitlab repository for convenience.

Extending usage with a database

A followup question that may arise when using Signals is whether integrating with a database would require any special or additional handling, and the simple answer is no. It's however worth understanding that the best place to have these database interactions happening would be inside the Signals effects so that the state of the database is immediately updated to stay in sync with the latest state of the Observables.

Let's go ahead and add a H2 local database to explore this question further.

Prepare the database

Create two folders, lib and data in project folder's root. The lib folder will contain the H2 jdbc driver jar, and the data folder will contain database-generated files.

Download and add a h2 driver jar (h2-2.2.224.jar) to the lib folder and add two more files to the same folder, namely create-db.sh and init-db.sql.

In the init-db.sql file, add the sql DDL statement.

drop table if exists todos;

create table if not exists todos (
 id uuid default random_uuid() primary key,
 title varchar(64) not null,
 completed boolean default false,
 constraint uniq_title unique(title)
);

In the create-db.sh file, add a script to create the database and execute the init-db script as well.

#!/usr/bin/env bash

#DB_URL=D:\Projects\signals-usage\data\test
DB_URL=~/Projects/signals-usage/data/test

java -cp h2-2.2.224.jar org.h2.tools.Shell \
-url "jdbc:h2:$DB_URL" \
-user "sa" \
-password "sa" \
-sql "$(cat ./init-db.sql)"

Once this is done, install the royaorm helper library to make working with the database a pleasant experience. Add a jdbc.properties file in the resource path to add database configuration properties. You could by the way use any other jdbc library that suits your taste, but this discussion will simply assume royaorm to keep things light.

# using maven
<dependency>
    <groupId>com.akilisha.oss</groupId>
    <artifactId>royaorm</artifactId>
    <version>0.0.1</version>
</dependency>

# using gradle
implementation("com.akilisha.oss:royaorm:0.0.1")

Add a properties file for configuring database connectivity.

src/main/resources/jdbc.properties

#jdbc.url=jdbc:h2:D:\\Projects\\signals-usage\\data\\test
jdbc.url=jdbc:h2:~/Projects/signals-usage/data/test
jdbc.username=sa
jdbc.password=sa

With the groundwork for integrating with the database laid out, the next step is to implement the programming piece.

Integrate with the database

In this particular case, there's only one table and one entity object to go with it. The first thing should therefore be to create an implementation of the RowMapper<T> interface that will map the values in a jdbc ResultSet returned by the select operation to an instance of the entity object.

entity object

@Data
@NoArgsConstructor
@AllArgsConstructor
public class TodoEntity {

    UUID id;
    String title;
    Boolean completed;

    public static TodoEntity from(Todo todo) {
        return new TodoEntity(todo.getId(), todo.getTitle().value(), todo.getCompleted().value());
    }
}

select operation mapper

public record TodoMapper (Connections pool, String query, String[] columns, Object... params) implements RowMapper<Todo> {

    @Override
    public void map(Todo entity, ResultSet rs) throws Throwable {
        entity.setId(rs.getObject("id", UUID.class));
        entity.getTitle().value(rs.getString("title"));
        entity.getCompleted().value(rs.getBoolean("completed"));
    }
}

For the insert operation, this would require creating an implementation of the RowInsert<T> interface to map query parameters to a PreparedStatement object.

public record TodoInsert(Connections pool, String query, RowResult result) implements RowInsert<TodoEntity> {

    @Override
    public void map(TodoEntity entity, PreparedStatement pst) throws Throwable {
        pst.setString(1, entity.getTitle());
    }
}

Similarly, for the delete operation, this would require creating an implementation of the RowDelete<T> interface to map query parameters to a PreparedStatement object.

public record TodoDelete (Connections pool, String query) implements RowDelete<TodoEntity> {

    @Override
    public void map(TodoEntity entity, PreparedStatement pst) throws Throwable {
        pst.setObject(1, entity.getId());
    }
}

Now since inserting, deleting and fetching concerns have been handled, updating the entity's title and toggling the completion status can now be handled too by creating the respective implementations of the RowUpdate<T> interface to map query parameters to a PreparedStatement object.

for updating title

public record TodoUpdate(Connections pool, String query, RowResult result) implements RowUpdate<TodoEntity> {

    @Override
    public void map(TodoEntity entity, PreparedStatement pst) throws Throwable {
        pst.setString(1, entity.getTitle());
        pst.setObject(2, entity.getId());
    }
}

for toggling completion status

public record TodoToggle(Connections pool, String query, RowResult result) implements RowUpdate<TodoEntity> {

    @Override
    public void map(TodoEntity entity, PreparedStatement pst) throws Throwable {
        pst.setBoolean(1, !entity.getCompleted());
        pst.setObject(2, entity.getId());
    }
}

The next sequence of actions will be applying database operations in the respective effects. At this point, it is worth mentioning that there is a real temptation to slap the database access code in the respective interface methods, in this case being the add, update, toggle and delete methods. Doing so would be falling back to the imperative approach where code execution proceeds strictly linearly. In an event-driven approach, code execution proceeds in response to events of interest happening in the program. In this case, those events of interest are happening in the Signals effects.

Fetch operations do not trigger Signals effects, and so these should be executed imperatively where it is deemed fit. The TodoMapper can be used when both selecting a single entity or a collection of entities.

single entity

public TodoEntity getById(String id) {
        Todo todo = new Todo();
        try {
            select(todo, new TodoMapper(Connections.pool(), "select * from todos where id = ?", new String[]{"id"}, id));
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
        return TodoEntity.from(todo);
    }

collection of entities

private void cacheIfEmpty() {
        try {
            if (this.todos.value().isEmpty()) {
                Collection<Todo> todos = selectList(Todo::new, new TodoMapper(Connections.pool(), "select * from todos", new String[]{}));
                this.todos.value().addAll(todos); //cache fetched values
            }
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

public Collection<TodoEntity> getAll() {
        cacheIfEmpty();
        return todos.value().stream().map(TodoEntity::from).collect(Collectors.toList());
    }

For the insert operation, the database has an autogenerated random_uuid() field for the primary key that is returned when the operation is completed. This value is extracted and set in the respective attribute of the Todo Observable.

insert (add effect) and delete (remove effect)

    public TodosList() {
        Signals.observe((event, entity) -> {
            switch (event) {
                case "add" -> {
                    System.out.printf("Event: %s, entity: %s\n", event, entity);
                    if (entity instanceof Todo todo) {
                        TodoEntity todoEntity = TodoEntity.from(todo);
                        try {
                            insert(todoEntity, new TodoInsert(Connections.pool(), "insert into todos (title) values (?)", (generated) -> {
                                generated.extract("id", UUID.class).accept(todo::setId);
                            }));
                        } catch (Throwable e) {
                            throw new RuntimeException(e);
                        }
                    }
                }
                case "remove" -> {
                    if (entity instanceof Todo todo) {
                        TodoEntity target = new TodoEntity(todo.getId(), null, null);
                        try {
                            DeleteOp.delete(target, new TodoDelete(Connections.pool(), "delete from todos where id = ?"));
                        } catch (Throwable e) {
                            throw new RuntimeException(e);
                        }
                    }
                }
                default -> System.out.printf("todos size: %d\n", todos.get().size());
            }
        });
    }

update title and toggle completion (update effects)

    public static Todo from(String title) {
        Todo todo = new Todo(title);
        // observe title attribute
        Signals.observe(() -> {
            System.out.printf("Task title changed to %s\n", todo.getTitle().get());
            TodoEntity entity = TodoEntity.from(todo);
            try {
                update(entity, new TodoUpdate(Connections.pool(), "update todos set title=? where id=?", null));
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        });
        // observe completed attribute
        Signals.observe(() -> {
            System.out.printf("Task completion changed to %s\n", todo.getCompleted().get());
            TodoEntity entity = TodoEntity.from(todo);
            try {
                update(entity, new TodoToggle(Connections.pool(), "update todos set completed=? where id=?", null));
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        });
        return todo;
    }

In the end, the database should reflect the final state of the operations, which is as expected. Open a terminal in the lib folder and the run commands as shown below.

DB_URL=~/Projects/signals-usage/data/test

java -cp h2-2.2.224.jar org.h2.tools.Shell \
-url "jdbc:h2:$DB_URL" \
-user "sa" \
-password "sa"

Welcome to H2 Shell 2.2.224 (2023-09-17)
Exit with Ctrl+C
...
...
sql> select * from todos;
ID                                   | TITLE          | COMPLETED
943820a9-f393-49bf-8ac9-9bf7f266d73d | wake up        | FALSE
74adb770-1468-40c0-85fe-002e92d08102 | make breakfast | FALSE
cdb29cc3-cd77-4fbe-a2bb-d24d85de6bbc | watch tv       | FALSE
...
...
sql> exit

Extending usage to UI tools

A great use case with UI tools is rendering a DOM element and then having itself or its children get re-rendered without explicitly searching for any elements in the DOM to make these updates. The parent and children elements are instrumented to re-render themselves whenever a corresponding update is made to the underlying data. The JSoup library is a popular tool in the Java ecosystem for creating and manipulating a DOM, and it is the perfect candidate to showcase this use case.

TodoItem without any observers

@Getter
@NoArgsConstructor
@AllArgsConstructor
public class TodoItem {

    private final Signal<String> title = Signals.signal("");
    private final Signal<Boolean> completed = Signals.signal(false);
    @Setter
    private UUID id;

    public TodoItem(String title) {
        this(UUID.randomUUID(), title, false);
    }

    public TodoItem(UUID id, String title, Boolean completed) {
        this();
        this.id = id;
        this.title.value(title);
        this.completed.value(completed);
    }

    @Override
    public String toString() {
        return "Todo{" +
                "title=" + title.value() +
                ", completed=" + completed.value() +
                ", id=" + id +
                '}';
    }
}

The first thing would be to create a TodosListUI class having:

a main method that creates and empty <ul> element and a Collection of TodoItem objects
static field with the corresponding markup for the unordered list element and the list item element.

TodosListUI app

public class TodosListUI {

    public static void main(String[] args) {
        Element ul = Objects.requireNonNull(new Element("template").html(listTemplate).firstElementChild());
        List<TodoItem> list = new ArrayList<>(List.of(new TodoItem("amka"), new TodoItem("lala")));
        // not yet implemented
    }

    // not yet implemented

    static final String itemTemplate = """
            <li style="padding: 5px" data-bind-key="id">
                <label style="display:flex;">
                    <input style="display:block;padding:5px;" type="checkbox" name="completed" data-bind-checked="completed" />
                    <span style="flex:1;color:#ff0000;" data-bind-text="title">%s</span>
                    <i class="delete" style="font-weight:bold" data-bind-event="click">x</i>
                </label>
            </li>
            """;

    static final String listTemplate = """
            <ul style="list-style-type:none;" data-bind-list>
            </ul>
            """;
}

The next thing is to understand that items can be added to or removed from an unordered list, and these operations happen at the Collection level. The observeElement function will instrument the TodoItems' Collection so that it will inform the <ul> element to either add or remove a child element in response to the add or remove operation happening in the underlying Collection object.

The helper methods, addListItem and removeListItem shown here will be implemented shortly.

static SignalCollection<TodoItem> observeElement(Element ul, List<TodoItem> list) {
    SignalCollection<TodoItem> observable = new SignalCollection<>(list);
    Signals.observe((event, entity) -> {
        switch (event) {
            case "add" -> {
                if (entity instanceof TodoItem todo) {
                    addListItem(ul, itemTemplate, todo);
                }
            }
            case "remove" -> {
                if (entity instanceof TodoItem todo) {
                    removeListItem(ul, todo);
                }
            }
            default -> System.out.println("List size: " + observable.get().size()); // trigger that will capture effects
        }
    });
    return observable;
}

The addListItem method will instrument two properties in a TodoItem entity. The first Signal will re-render the <span> text when the entity's title is updated. The second Signal will add or removed the checked attribute to the checkbox element when the entity's completed value is toggled.

static void addListItem(Element parent, String template, TodoItem entity) {
    Element li = Objects.requireNonNull(new Element("template").html(template).firstElementChild());
    Objects.requireNonNull(li.select("li").first()).attr("data-bind-key", entity.getId().toString());
    Signals.observe(() -> Objects.requireNonNull(li.select("span").first()).text(entity.getTitle().get()));
    Signals.observe(() -> Objects.requireNonNull(li.select("input[type=checkbox]").first()).attr("checked", entity.getCompleted().get()));
    parent.appendChild(li);
}

The removeListItem method is pretty straightforward also, and it will look for a child element inside the <ul> parent element, and remove it if its [data-bind-key] attribute value matches the entity's id value.

static void removeListItem(Element parent, TodoItem entity) {
    Element el = parent.select(String.format("[data-bind-key=%s]", entity.getId().toString())).first();
    Objects.requireNonNull(el).remove();
}

To render the <ul> element using the list of todo items, a simple helper can be used to achieve this.

static void renderList(Element parent, SignalCollection<TodoItem> observable) {
    for (TodoItem item : observable.value()) {
        addListItem(parent, itemTemplate, item);
    }
}

Finally, all this can be brought together in the main method for testing.

public static void main(String[] args) {
    Element ul = Objects.requireNonNull(new Element("template").html(listTemplate).firstElementChild());
    List<TodoItem> list = new ArrayList<>(List.of(new TodoItem("amka"), new TodoItem("lala")));
    SignalCollection<TodoItem> observable = observeElement(ul, list);

    renderList(ul, observable);

    // make changes to the underlying objects
    list.get(0).getCompleted().set(true);
    list.get(1).getTitle().set("something");
    observable.add(new TodoItem("zusha"));
    System.out.println(ul);
}

It's easy to spot that there is no explicit lookup happening in the <ul> element to find chile elements that need updating when the Collection of TodoItems is updated or when a TodoItem itself is updated. The results below show that the updated made have been applied correctly.

List size: 2 // initial size of list
<ul style="list-style-type:none;" data-bind-list>
 <li style="padding: 5px" data-bind-key="5a98acd0-ef31-4749-b377-18a0d8c931f4"><label style="display:flex;"> <input style="display:block;padding:5px;" type="checkbox" name="completed" data-bind-checked="completed" checked> <span style="flex:1;color:#ff0000;" data-bind-text="title">amka</span> <i class="delete" style="font-weight:bold" data-bind-event="click">x</i> </label></li>
 <li style="padding: 5px" data-bind-key="fced5b16-778e-4c8e-92ef-75af9695ac42"><label style="display:flex;"> <input style="display:block;padding:5px;" type="checkbox" name="completed" data-bind-checked="completed"> <span style="flex:1;color:#ff0000;" data-bind-text="title">something</span> <i class="delete" style="font-weight:bold" data-bind-event="click">x</i> </label></li>
 <li style="padding: 5px" data-bind-key="c35f3106-f2ab-4a2a-a3ff-28200de2de44"><label style="display:flex;"> <input style="display:block;padding:5px;" type="checkbox" name="completed" data-bind-checked="completed"> <span style="flex:1;color:#ff0000;" data-bind-text="title">zusha</span> <i class="delete" style="font-weight:bold" data-bind-event="click">x</i> </label></li>
</ul>

For the next example, I'm working on eliminating the need to explicitly implement the observeElement, addListItem, removeListItem and renderList methods and instead have then executed implicitly when the right configuration is provided.

Stay tuned!

Finito

The source code to this presentation can be found in this repo. A more detailed and non-trivial example is also presented in the same repo.