From Calling to Eventing
Building Egoistic, Loosely Coupled and Eventually Distributed Services
A hands-on tutorial using Java 26, virtual threads, the filesystem, Syncthing, and finally MQTT.
Course Material and Software
Every Java program used in this tutorial appears in full in Appendix A. The identical files are also available separately so that you can run and modify them immediately:
| Stage | Supplemental source |
|---|---|
| procedural calls | 01-procedural/ProceduralRot13.java |
| mediated queues in one JVM | 02-static-in-jvm/StaticQueueRot13.java |
| mediated filesystem | 03-static-filesystem/ |
| dynamic queues in one JVM | 04-dynamic-in-jvm/DynamicQueueRot13.java |
| dynamic filesystem | 05-dynamic-filesystem/ |
| MQTT | 06-mqtt/ |
Install or bookmark the tools before the corresponding exercise:
- Java 26;
- Syncthing downloads;
- Eclipse Mosquitto, the MQTT broker used in the exercises;
- MQTT Explorer, a graphical MQTT client for inspecting, publishing and subscribing to messages;
- Eclipse Paho Java client, used by the Java MQTT examples.
The examples deliberately use Java 26 compact source files whenever possible:
void main() {
var text = IO.readln();
IO.println(text);
}
There is no ceremonial public class ... or public static void main(...)
unless a conventional class genuinely helps. The goal is fast experimentation,
not maximum Java ceremony.
0. What This Tutorial Is About
We are going to implement something spectacularly uninteresting:
keyboard → ROT13 → console
You type:
Hello World
and receive:
Uryyb Jbeyq
The functionality will remain almost unchanged throughout the tutorial.
That is deliberate.
Our subject is not ROT13.
Our subject is the architecture around it.
We will implement the same system repeatedly while asking three questions:
Who knows whom?
then:
Who owns what?
and eventually:
Who decides how the system is wired together?
The tutorial follows one more rule:
Meet each solution only after you have experienced the problem it solves.
That is why you will not begin with MQTT.
You will not begin with dynamic subscriptions either.
First you will deliberately build a system containing somewhat ridiculous mediators. After you have operated it, you will decide whether they deserve to remain.
Our journey will be:
direct method calls
↓
egoistic services
↓
mediators glue them together
↓
independent processes
↓
filesystem as transport
↓
Syncthing
↓
completely distributed mediated system
↓
"Why on earth do we need all these mediators?"
↓
REWIND
↓
let compatible services listen themselves
↓
make that wiring dynamic
↓
repeat across processes
↓
repeat across Syncthing
↓
discover when a mediator IS actually useful
↓
filesystem has now become sufficiently annoying
↓
MQTT
↓
same architecture, better transport
Later we will distinguish:
Intent = what should happen or become true.
Status/State = what is currently true.
Event = something that happened.
Each concept receives a name only after you have already used it.
Part I — Start With What You Already Know
1. Rock Bottom: Just Call It
The first implementation is procedural.
Conceptually:
ConsoleIn
│
│ call
▼
ROT13
│
│ call
▼
ConsoleOut
The important part is tiny:
String text = IO.readln();
String transformed = rot13(text);
IO.println(transformed);
The complete compact Java 26 program is
ProceduralRot13.java
and is reproduced in Appendix A.
There is absolutely nothing wrong with this program.
That is important.
We are not learning:
Method calls are bad.
We are learning to recognize what a method call implies.
Look at:
String transformed = rot13(text);
The caller knows:
- that ROT13 exists;
- how to invoke it;
- what argument it expects;
- what result it returns;
- that ROT13 is available right now;
- and that execution continues after ROT13 returns.
Likewise the next part of the program knows where the result must go.
So we have:
ConsoleIn ──knows──► ROT13 ──knows──► ConsoleOut
Our recurring question begins here:
Who knows whom?
2. Make Three Egoistic Capabilities
Now imagine that we want three independent capabilities:
ConsoleIn ROT13 ConsoleOut
"I read" "I rotate" "I print"
We will call these components egoistic.
An egoistic component says:
I know what I can do.
I know what data I accept.
I know what data I produce.
I would prefer not to know the complete business process surrounding me.
For example, ConsoleIn should ideally not contain:
rot13(...)
ROT13 should ideally not contain:
consoleOut(...)
But there is an immediate problem.
If nobody knows anybody...
Who makes this into a system?
Excellent.
We have created a problem worth solving.
3. Give Every Capability a Mailbox
Before creating a mailbox, define the small object that travels through it:
record Message(String type, String value) {}
A Message contains:
- a
type, which tells a receiver what kind of message it is; - a
value, which contains the actual payload.
For example:
new Message("TEXT", "Hello World")
Inside this first JVM, that is all a message needs. We will add an optional timestamp only when messages become files in step 13.
Inside one JVM, the smallest useful primitive is a queue:
BlockingQueue<Message>
We can maintain several of them:
Map<String, BlockingQueue<Message>> mailboxes =
new ConcurrentHashMap<>();
with a convenience method:
BlockingQueue<Message> mailbox(String name) {
return mailboxes.computeIfAbsent(
name,
_ -> new LinkedBlockingQueue<>()
);
}
For the moment, deliberately call these mailboxes, not MQTT topics.
A BlockingQueue is fundamentally queue-like: a message taken from it is consumed by one reader.
A queue is enough for this first experiment. It does not yet provide full Publish/Subscribe semantics.
This limitation is worth stating bluntly:
If two readers call
take()on the same mailbox, they compete. Each message goes to one of them; it is not copied to both.
That is perfectly adequate for our first command queue. Later, when we want
multiple independent subscribers, this limitation will help us understand what
Publish/Subscribe adds. For now we keep the Map and the queues because they
are wonderfully simple.
4. ConsoleIn Becomes Egoistic
Instead of calling ROT13:
rot13(text);
ConsoleIn publishes into its own mailbox:
void runConsoleIn() {
try {
while (true) {
var text = IO.readln();
if (text == null)
return;
mailbox("consoleIn.outbox")
.put(new Message("TEXT", text));
}
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
ConsoleIn knows only:
I read text.
I publish what I read.
There is no ROT13 dependency inside the service.
5. ROT13 Becomes Egoistic
ROT13 has its own inbox:
Message input =
mailbox("rot13.inbox").take();
It performs its capability:
Message output = new Message(
"TEXT",
rot13(input.value())
);
and writes its own output:
mailbox("rot13.outbox").put(output);
Its world is now:
rot13.inbox
│
▼
ROT13
│
▼
rot13.outbox
It does not know ConsoleIn.
It does not know ConsoleOut.
6. ConsoleOut Becomes Egoistic
Likewise, ConsoleOut is just another small function in our compact source file:
while (true) {
Message message =
mailbox("consoleOut.inbox").take();
IO.println(message.value());
}
ConsoleOut knows only:
I display what arrives in my inbox.
Wonderful.
We now have three nicely isolated capabilities.
And they do absolutely nothing together.
7. Enter the Mediator
Somebody has to establish:
ConsoleIn → ROT13 → ConsoleOut
So introduce the smallest possible mediator:
void mediate(String from, String to) {
try {
while (true) {
Message message =
mailbox(from).take();
mailbox(to).put(message);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
Start one:
Thread.ofVirtual().start(
() -> mediate(
"consoleIn.outbox",
"rot13.inbox"
)
);
and another:
Thread.ofVirtual().start(
() -> mediate(
"rot13.outbox",
"consoleOut.inbox"
)
);
The system becomes:
ConsoleIn
│
▼
consoleIn.outbox
│
▼
Mediator
│
▼
rot13.inbox
│
▼
ROT13
│
▼
rot13.outbox
│
▼
Mediator
│
▼
consoleOut.inbox
│
▼
ConsoleOut
Run it.
Functionally:
Hello World
→
Uryyb Jbeyq
Again.
Architecturally, however, something important has happened.
The complete runnable version is
StaticQueueRot13.java.
It keeps the Map<String, BlockingQueue<Message>>; it does not replace the
queue with an MVC-style listener abstraction.
8. Ask Again: Who Knows Whom?
Does ConsoleIn know ROT13?
No.
Does ROT13 know ConsoleIn?
No.
Does ROT13 know ConsoleOut?
No.
Does ConsoleOut know ROT13?
No.
Has the topology disappeared?
No.
Who knows it?
the mediators / orchestrator
This is our first major result:
Decoupling often does not make knowledge disappear.
It moves knowledge to where it belongs.
The capability:
ROT13 transforms text
belongs to ROT13.
The topology:
ConsoleIn → ROT13 → ConsoleOut
belongs to system composition.
9. Why Virtual Threads?
Several components now wait indefinitely:
wait for keyboard input
wait for mailbox input
wait for another mailbox
Virtual threads make it convenient to run each such loop independently:
Thread.ofVirtual().start(service::run);
This is merely a Java implementation convenience.
Do not confuse:
virtual threads
with:
Event-Driven Architecture
They are different concepts.
The same architecture can later run as separate processes.
In step 11, you will do exactly that.
10. Keep the Services Egoistic
At this point you might notice:
ConsoleIn
↓
Mediator
↓
ROT13
and think:
Why doesn't ROT13 simply read ConsoleIn's outbox?
Because ROT13 is still egoistic.
ROT13 knows:
I transform text.
It does not know:
ConsoleIn exists.
ConsoleIn publishes to consoleIn.outbox.
If ROT13 simply reads that outbox, the topology moves into ROT13. The program becomes shorter, but ROT13 is no longer an egoistic capability.
The same applies at the other end: ConsoleOut should not need to know that its text came from ROT13.
Therefore the mediators remain the components that know the current wiring:
egoistic services
topology in mediators
This may look indirect—even a little odd. That is an honest consequence of the architecture at this stage. Later you will find a different way to keep the services egoistic without forwarding every application message through a mediator.
Part II — Cross the Process Boundary
11. Separate the Components
We now want:
Process 1 Process 2 Process 3
ConsoleIn ROT13 ConsoleOut
They can no longer share:
BlockingQueue<Message>
because the queues live inside a JVM.
We need an external transport.
We could use TCP.
We could use MQTT.
For this experiment, deliberately use something primitive:
the filesystem
Why?
Because you can see it.
12. A Directory Becomes a Mailbox
Create:
messageBoard/
and inside it:
messageBoard/
├── consoleIn/
│ └── outbox/
│
├── rot13/
│ ├── inbox/
│ └── outbox/
│
└── consoleOut/
└── inbox/
For this version:
directory = mailbox
file = message
Publishing means:
create message file
Receiving means:
watch directory for message files
Still no framework.
Still no broker.
13. Keep the First File Format Ridiculously Simple
When a message becomes a file, extend the record with an optional timestamp:
record Message(
String type,
String value,
Instant timestamp
) {}
We could introduce Jackson already.
Do not add it yet.
Two lines are sufficient; an optional timestamp may be added as the final line:
TEXT
Hello World
TIMESTAMP: 2026-09-22T10:42:12.123Z
A message written manually may omit it:
TEXT
Hello World
The TIMESTAMP: marker keeps a plain ISO-looking payload from being mistaken
for metadata. In this deliberately tiny format, a final payload line beginning
with TIMESTAMP: is reserved for the timestamp.
Serialization can be:
String serialize(Message message) {
var text = message.type()
+ "\n"
+ message.value();
if (message.timestamp() != null)
text += "\nTIMESTAMP: " + message.timestamp();
return text;
}
And reading:
var lines = Files.readString(file).split("\\R", -1);
var endOfValue = lines.length;
Instant timestamp = null;
if (lines[lines.length - 1].startsWith("TIMESTAMP: ")) {
timestamp = Instant.parse(
lines[lines.length - 1]
.substring("TIMESTAMP: ".length())
);
endOfValue--;
}
var type = lines[0];
var value = String.join(
"\n",
Arrays.copyOfRange(lines, 1, endOfValue)
);
It is primitive.
That is useful.
There is almost nothing hiding the architecture.
14. The First Distributed-Systems Problem: Publication
Consider:
Files.writeString(
Path.of("message.msg"),
hugeMessage
);
Another process may discover:
message.msg
before the writer has finished.
So define a publication protocol:
1. create private temporary file
2. write complete message
3. close it
4. atomically rename it
5. consumers react only to final files
For example:
550e8400-e29b-41d4-a716-446655440000.tmp
│
│ writing
▼
550e8400-e29b-41d4-a716-446655440000.tmp
│
│ complete
▼
atomic rename
│
▼
550e8400-e29b-41d4-a716-446655440000.msg
The .tmp and .msg names are in the same mailbox directory. Keeping the
move within one directory avoids accidentally crossing filesystems and makes
the local atomic-move requirement explicit.
Implementation:
Path temporary =
directory.resolve(name + ".tmp");
Path published =
directory.resolve(name + ".msg");
Files.writeString(
temporary,
content
);
Files.move(
temporary,
published,
StandardCopyOption.ATOMIC_MOVE
);
The important concept is:
Writing and publishing are not the same operation.
15. Observe a Mailbox with WatchService
Java already gives us a primitive filesystem observer:
WatchService watcher =
FileSystems
.getDefault()
.newWatchService();
Register:
directory.register(
watcher,
StandardWatchEventKinds.ENTRY_CREATE
);
Wait:
WatchKey key = watcher.take();
Then inspect the newly created .msg files.
The observer must ignore every other filename, including .tmp:
var filename = (Path) event.context();
if (!filename.toString().endsWith(".msg"))
continue;
This also gives us a delightfully low-tech injection tool. Create a complete
file in a mailbox as, for example, manual.tmp, then rename it to
manual.msg. The watcher reacts only to the rename that creates the final
.msg name. Try both accepted formats from step 13—with and without the
trailing timestamp.
The architectural idea is:
producer
│
│ publish
▼
filesystem
│
│ observation
▼
consumer
The mechanics will later change.
The idea will not.
The complete shared implementation is
Message.java and
FileMessageBoard.java.
16. The Three Services Become Separate Programs
We now have:
SensorService.java
ProcessorService.java
ActuatorService.java
The sensor writes only:
consoleIn/outbox
ROT13 reads only:
rot13/inbox
and writes:
rot13/outbox
ConsoleOut reads only:
consoleOut/inbox
They are now separate programs.
Complete sources:
But once again:
nobody connects them
Good.
We already know how to solve that.
17. The Mediator Becomes a Process
Create:
Mediator.java
Its job is still conceptually:
mediate(from, to)
But now mediation means:
watch source directory
for each new message:
publish a copy
into destination directory
Run:
Mediator
consoleIn/outbox
→
rot13/inbox
and:
Mediator
rot13/outbox
→
consoleOut/inbox
Our architecture is unchanged:
ConsoleIn
│
▼
Mediator
│
▼
ROT13
│
▼
Mediator
│
▼
ConsoleOut
Only the implementation boundary changed.
The complete process is
Mediator.java.
Notice something slightly asymmetric. The three capability services receive their work through mailboxes, but the mediator receives its route through startup arguments:
Mediator.java <from> <to>
The route is still external configuration—it is not compiled into the mediator—so the mediator already has a generic capability. But it has no name and no control mailbox of its own. Keep that observation. After you have seen what the distributed mediators cost, you will make their control interface look like that of an ordinary service.
18. Run the Fully External Local Version
Use a fresh message board:
rm -rf messageBoard
Open terminals.
Terminal 1
java SensorService.java
Terminal 2
java Mediator.java \
consoleIn/outbox \
rot13/inbox
Terminal 3
java ProcessorService.java
Terminal 4
java Mediator.java \
rot13/outbox \
consoleOut/inbox
Terminal 5
java ActuatorService.java
Count them.
1 sensor
1 mediator
1 processor
1 mediator
1 actuator
= 5 processes
All to perform:
keyboard → ROT13 → console
Do not optimize this yet.
We are not done making it painful.
Part III — Make It Actually Distributed
19. One Filesystem Is Not Yet Very Distributed
Five processes on one computer prove that object references are gone.
But every process still sees the same local filesystem.
We now want something more interesting:
Computer A Computer B Computer C
Yet our programs communicate by files.
How can a file written on Computer A become visible on Computer B?
For this exercise, use:
20. What Syncthing Is Doing for Us
For this course, treat Syncthing as:
a tool that keeps a directory replicated between computers.
That is all we need from it.
Syncthing is not our message broker.
It knows nothing about:
Intent
Event
Status
ROT13
mediators
It merely gives us local directories whose contents are asynchronously replicated between participating machines.
Conceptually:
Computer A
messageBoard/
│
│
│ Syncthing
│
▼
Computer B
messageBoard/
│
│
│ Syncthing
│
▼
Computer C
messageBoard/
Each program continues using ordinary local files.
Syncthing makes those files appear elsewhere.
21. Prepare a Shared Course Folder
On every participating computer:
- install and start Syncthing;
- connect the participating devices;
- share one folder between them;
- choose a local directory for that folder;
- use this directory as the root of our message board.
For example:
Computer A:
/home/alice/course-message-board
Computer B:
/home/bob/sin/course-message-board
Computer C:
C:\SIN\course-message-board
The local paths do not need to be identical.
Our Java program only needs to know its own local root.
A useful implementation is:
Path MESSAGE_BOARD_ROOT =
Path.of(
System.getProperty(
"messageBoard.root",
"messageBoard"
)
);
Then:
java \
-DmessageBoard.root="/home/alice/course-message-board" \
SensorService.java
can run on one machine while another uses a completely different local path.
22. Verify Syncthing Before Starting the System
Do not debug Java and directory synchronization simultaneously.
First perform the world's least exciting distributed-systems test.
On Computer A create:
hello.txt
inside the synchronized folder.
Check that it appears on Computers B and C.
Then remove it.
Check that the change propagates.
Only after simple synchronization works do we run the distributed services.
This rule will save you considerable frustration:
Test one layer at a time.
23. Important: Syncthing Is Not a Shared Atomic Filesystem
Locally we have:
.tmp
↓
atomic rename
↓
.msg
That gives a useful local publication boundary.
But Syncthing performs asynchronous replication.
Do not infer:
atomic locally
=
atomic globally
Those are different guarantees.
Our consumers still ignore unfinished protocol files such as:
*.tmp
But synchronization latency and notification behaviour are transport characteristics we have not solved.
Good.
Leave these weaknesses visible and observe them. That is part of the exercise.
24. Distribute the Five Processes
One possible setup is:
Computer A
──────────
SensorService
Computer B
──────────
Mediator 1
ProcessorService
Computer C
──────────
Mediator 2
ActuatorService
Or, if you have enough machines:
A: Sensor
B: Mediator 1
C: ROT13
D: Mediator 2
E: ConsoleOut
The exact physical placement does not matter.
The important thing is that all processes use their local replica of the synchronized message board.
Now type:
Hello World
on Computer A.
Watch the files propagate.
Watch the mediator publish another file.
Watch that propagate.
Watch ROT13 publish another file.
Watch that propagate.
Watch the second mediator publish another file.
Watch that propagate.
Eventually:
Uryyb Jbeyq
appears somewhere else.
Congratulations.
You have built an absurdly elaborate ROT13 machine.
The absurdity is the lesson: the architecture now has a cost you can observe.
25. Count What One Keystroke Costs
For one input:
Hello
we get approximately:
1. ConsoleIn publishes input
2. Mediator 1 publishes copy
into ROT13 inbox
3. ROT13 publishes transformed output
4. Mediator 2 publishes copy
into ConsoleOut inbox
Four application-level publications for one logical input.
And each may be replicated through Syncthing.
Our two mediators provide no semantic transformation.
They simply copy:
TEXT → TEXT
We also operate:
two extra processes
with:
- startup;
- shutdown;
- error handling;
- directories;
- observation loops;
- possible failure;
- synchronization latency.
Now the mediator is no longer free.
Now the question becomes interesting.
26. Break a Mediator
Stop Mediator 1.
Type:
Hello
What happens?
ConsoleIn is alive.
ROT13 is alive.
ConsoleOut is alive.
The synchronized filesystem is alive.
But the business flow stops.
Now restart it.
Do the same with Mediator 2.
The mediators are part of our availability chain.
Again, this does not mean mediation is bad.
It means:
Mediation has a cost.
And now we are ready to ask whether every mediator deserves to exist.
27. What Does Our Mediator Actually Contribute?
Mediator 1 receives:
TEXT
...
Hello
and publishes:
TEXT
...
Hello
Mediator 2 receives:
TEXT
...
Uryyb
and publishes:
TEXT
...
Uryyb
Neither adds information.
Neither changes semantics.
Neither adapts an incompatible format.
They perform approximately:
identity(x) = x
but as separate distributed processes.
The Mediator Is a Service Too
Before removing the mediators, remove one unnecessary difference between them and the other services.
Our current mediator starts like this:
java Mediator.java \
consoleIn/outbox \
rot13/inbox
Its capability is already generic:
copy messages from one mailbox
to another mailbox
The mailbox names are not compiled into Mediator.java. Nevertheless, its
control interface is special: the shell supplies its route at startup, while
the other services receive messages through their own inboxes.
Give each mediator only a name at startup:
java ConfigurableMediator.java mediator1
From that identity it derives one permanent control mailbox:
mediators/mediator1/inbox
It knows its own name and capability. It still does not know ConsoleIn, ROT13 or ConsoleOut.
Now send a configuration message to that inbox:
ROUTE
consoleIn/outbox
rot13/inbox
The first line is the message type. The next two lines form its value:
from mailbox
to mailbox
Keep from and to in one message. Two independent messages could briefly
combine a new source with an old destination.
This ROUTE message describes desired behaviour. Later we will give that kind
of message a more precise name: Intent.
You can inject it by writing a .tmp file in the control mailbox and renaming
it to .msg, exactly as in step 15. For the complete example, let a small
program publish both routes so that the system topology is visible in one
place.
Let an Orchestrator Supply the Topology
Start the five long-running programs from
supplemental-java/03-static-filesystem. Use a separate terminal for each one:
java -DmessageBoard.root=messageBoard-configurable SensorService.java
java -DmessageBoard.root=messageBoard-configurable \
ConfigurableMediator.java mediator1
java -DmessageBoard.root=messageBoard-configurable ProcessorService.java
java -DmessageBoard.root=messageBoard-configurable \
ConfigurableMediator.java mediator2
java -DmessageBoard.root=messageBoard-configurable ActuatorService.java
Both mediators initially know no route. They wait on:
mediators/mediator1/inbox
mediators/mediator2/inbox
After all five programs are waiting, run the short-lived orchestrator in a sixth terminal:
java -DmessageBoard.root=messageBoard-configurable \
MediatorOrchestrator.java
It publishes:
mediator1:
consoleIn/outbox → rot13/inbox
mediator2:
rot13/outbox → consoleOut/inbox
Then it exits. Type text in the SensorService terminal and verify that the transformed text appears in the ActuatorService terminal.
Complete sources:
What Improved—and What Did Not?
The named mediator now behaves like a normal configurable service:
identity at startup
configuration through its own inbox
generic capability in its source code
topology supplied by an orchestrator
The orchestrator contains the system topology, but it is not in the application's data path. It sends configuration and exits.
This first configurable mediator deliberately accepts one valid ROUTE
message. It then stops observing its control inbox and watches the configured
source mailbox. It does not promise live rewiring.
Why not? Our current FileMessageBoard.watch(...) watches one directory and
blocks forever. Watching the control inbox while adding and removing other
mailboxes requires the multi-directory subscription helper introduced later.
That new requirement remains visible; it is the reason the later helper exists.
Most importantly, the data path did not improve:
event
↓
named mediator
↓
copied event
The mediator is now consistently controlled and still expensive. It remains a process in the availability chain and still copies every application message without adding semantics.
This is the moment to ask:
Why can't the service interested in the data simply listen to it itself?
Only now does the next architectural question have practical weight.
Part IV — Rewind
28. Go Back to the Simple JVM
Now rewind deliberately.
Forget Syncthing for a moment.
Forget processes.
Return to:
ConsoleIn
ROT13
ConsoleOut
inside one JVM.
Previously:
ConsoleIn
↓
Mediator
↓
ROT13
What if ROT13 could listen directly to:
consoleIn.outbox
?
Then:
ConsoleIn
│
▼
consoleIn.outbox
│
▼
ROT13
No mediator.
Wonderful.
So we might write inside ROT13:
listen("consoleIn.outbox");
But ask our old question:
Who knows whom?
Oops.
ROT13 now contains knowledge about ConsoleIn.
We removed the mediator...
and put topology back into the service.
29. Separate Capability from Topology
ROT13 should know:
I can transform text.
It may also know:
I am capable of listening
for compatible text messages.
It should not necessarily know:
my input always comes from
consoleIn.outbox
That is not part of the ROT13 capability.
That is deployment topology.
So instead of compiling:
listen("consoleIn.outbox");
into ROT13, we send:
TOPIC consoleIn.outbox
at runtime.
ROT13 learns its wiring from data.
30. Dynamic Subscription Inside One JVM
We do not replace our simple command queues here. listen(...) is merely a
name for a virtual thread that repeatedly calls take() on one of the same
queues stored in our original Map:
void listen(String mailboxName, Consumer<Message> receiver) {
try {
while (true)
receiver.accept(mailbox(mailboxName).take());
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
Consequently, the old limitation remains: two such listeners on the same mailbox compete for messages. This stage demonstrates runtime wiring, not yet true broadcast Publish/Subscribe. MQTT will supply those transport semantics later.
ROT13 can maintain:
Map<String, Thread> subscriptions =
new HashMap<>();
and provide:
void subscribe(String mailboxName) {
Thread listener =
Thread.ofVirtual().start(
() -> listen(
mailboxName,
this::process
)
);
subscriptions.put(
mailboxName,
listener
);
}
The important line is not the virtual thread.
It is:
subscribe(mailboxName);
where:
mailboxName
arrived from outside the service.
Now configure:
mailbox("rot13.inbox").put(
new Message(
"TOPIC",
"consoleIn.outbox"
)
);
ROT13 reacts to the special message and starts observing the supplied mailbox.
The complete runnable queue version is
DynamicQueueRot13.java.
31. Do the Same for ConsoleOut
Send:
mailbox("consoleOut.inbox").put(
new Message(
"TOPIC",
"rot13.outbox"
)
);
ConsoleOut starts listening directly to ROT13's output.
Now:
ConsoleIn
│
▼
ROT13
│
▼
ConsoleOut
Application data no longer passes through forwarding mediators.
But the services still did not compile their neighbours into their capabilities.
The wiring arrived separately.
32. Our ROUTE and TOPIC Messages Are Intent
The named mediator received:
ROUTE
consoleIn/outbox
rot13/inbox
That meant:
Please forward messages from here to there.
Now ROT13 receives:
TOPIC consoleIn.outbox
Does it describe something that happened?
No.
Does it describe current state?
No.
That means:
Please begin listening here.
That is desired behaviour.
In other words:
Intent
Both messages request desired behaviour. Both are Intent.
We reached the concept by operating the system before introducing a larger architectural vocabulary.
33. What Happened to the Mediator?
It did not entirely vanish.
With the named mediator, the orchestrator established a route:
"mediator1, forward from here to there."
But the mediator then continuously transported application data:
event
↓
mediator
↓
consumer
Now the orchestrator sends configuration to the consumers themselves:
"ROT13, listen there."
"ConsoleOut, listen there."
Afterwards:
ConsoleIn ─────► ROT13 ─────► ConsoleOut
runs independently.
This is a useful distinction:
control plane
versus:
data plane
The orchestrator participates in the control plane.
It does not have to remain in the application's data path.
34. Static Mediated vs Dynamic Direct
Compare:
STATIC MEDIATED
Producer
│
▼
Mediator
│
▼
Consumer
with:
DYNAMIC DIRECT
Intent
│
▼
Producer ─────► Consumer
The dynamic version is not automatically superior.
It makes sense only if the consumer actually understands the producer's message contract.
Keep that condition in mind. It becomes important in step 48.
Part V — Replay the Dynamic Idea Across Processes
35. Now Leave the JVM Again
We solved the mediator problem in the easiest environment.
Now repeat the architectural journey.
We again have separate programs:
SensorService
ProcessorService
ActuatorService
But instead of mediators continuously copying files, services should be able to watch external mailboxes themselves.
ROT13 always knows one control mailbox:
rot13/inbox
A configuration message arrives:
TOPIC consoleIn/outbox
ROT13 then begins watching:
consoleIn/outbox
directly.
36. One WatchService Is No Longer Enough
Our first filesystem helper could do:
watch(topic, receiver);
and block forever.
That was sufficient when one process always watched one fixed mailbox.
Now ROT13 needs to:
always watch its own control inbox
AND
dynamically add another watched directory
possibly remove it later
This new requirement justifies one additional helper.
37. Helper: FileSubscriptions
The helper conceptually provides:
subscriptions.subscribe(topic);
subscriptions.unsubscribe(topic);
Message message =
subscriptions.take();
Internally it manages several:
WatchKey
registrations.
ROT13 can therefore do:
subscriptions.subscribe(
"rot13/inbox"
);
permanently.
Then when it receives:
TOPIC consoleIn/outbox
it executes:
subscriptions.subscribe(
"consoleIn/outbox"
);
The service implementation still does not contain:
consoleIn/outbox
as a compiled topology decision.
The complete helper is
FileSubscriptions.java.
It registers several directories with one WatchService and places only
finished .msg publications onto its internal queue.
38. Run Dynamic Filesystem Wiring Locally
Start:
SensorService
DynamicProcessorService
DynamicActuatorService
Do not start the forwarding mediators.
Then send configuration:
ROT13:
TOPIC consoleIn/outbox
and:
ConsoleOut:
TOPIC rot13/outbox
Type:
Hello
The path is now:
consoleIn/outbox
│
│ observed directly
▼
ROT13
│
▼
rot13/outbox
│
│ observed directly
▼
ConsoleOut
Count the application publications.
Previously:
input
mediator copy
ROT13 result
mediator copy
Now:
input
ROT13 result
The wiring Intent is sent only when topology changes.
Now the architectural advantage is visible.
Complete sources:
Configure.java publishes the two TOPIC messages. At this point it is just
one tiny compact Java 26 program that writes configuration messages.
Part VI — Replay It Over Syncthing
39. Distribute the Dynamic Version
Return to our synchronized course folder.
Possible arrangement:
Computer A
──────────
Sensor
Computer B
──────────
ROT13
Computer C
──────────
ConsoleOut
No forwarding mediator processes.
Send configuration:
ROT13:
listen to
consoleIn/outbox
and:
ConsoleOut:
listen to
rot13/outbox
through their own control mailboxes.
Then type:
Hello
on Computer A.
The input message is replicated by Syncthing.
ROT13 observes its local synchronized copy.
ROT13 publishes its own output.
That publication is replicated.
ConsoleOut observes it.
Conceptually:
Computer A Computer B Computer C
ConsoleIn ROT13 ConsoleOut
│ ▲ ▲
│ │ │
└──── Syncthing ───┘ │
└──── Syncthing ───┘
No distributed identity-function MiMs remain.
40. Be Precise About "Direct"
The services are logically directly connected:
ConsoleIn Event
↓
ROT13
ROT13 Event
↓
ConsoleOut
That does not mean network packets necessarily travel only between those two machines.
Syncthing decides how it replicates the synchronized folder.
Our architecture controls:
which messages a service reacts to
not the internal networking strategy of Syncthing.
That distinction matters.
41. Why This Version Is Better
For one logical input:
Hello
we now publish:
1. input Event
2. ROT13 output Event
instead of:
1. input
2. mediator copy
3. ROT13 output
4. mediator copy
We also removed two continuously running processes.
More importantly:
ROT13 capability
still does not contain:
ConsoleIn identity
The topology came from runtime configuration.
This is the architectural payoff.
42. Now Introduce the Proper Vocabulary
We have already encountered three kinds of messages.
It is time to name them properly.
We use:
I = Intent
S = Status / State
E = Event
43. E — Event
An Event describes:
something that happened
Examples:
textEntered
buttonPressed
temperatureChanged
transformationCompleted
vehiclePassedBarcode
Past tense is often a useful naming hint.
Our input publication becomes conceptually:
ConsoleIn/E/text
ROT13 output becomes:
ROT13/E/crypted
44. S — Status / State
Status describes:
something currently true
Examples:
online = true
mode = automatic
currentSpeed = 250
subscription =
ConsoleIn/E/text
If someone asks:
Why is ROT13 reacting to ConsoleIn?
the service can publish:
ROT13/S/subscriptions
containing its current wiring.
State explains current behaviour.
45. I — Intent
Intent means:
what should happen or become true
Examples:
transform this text
subscribe to this Event
unsubscribe from this Event
change speed
switch lights on
Our primitive:
TOPIC consoleIn/outbox
was really an early Intent.
A more explicit structure could eventually become:
ROT13/I
with:
SUBSCRIBE ConsoleIn/E/text
46. Event-Driven Does Not Mean "No Commands"
Suppose somebody sends:
ROT13/I
ACTION Hello
That is command-like.
Fine.
ROT13 performs its capability and publishes:
ROT13/E/crypted
Uryyb
Likewise:
SUBSCRIBE ConsoleIn/E/text
is Intent.
And:
ROT13/S/subscriptions
describes State.
A useful flow is:
Intent
│
▼
Component
│
├────► State
│
└────► Events
There is no contradiction.
47. The One-Writer Rule
We now also introduce:
Every writable resource has exactly one writer authority.
Do not confuse this with the UUID in a message filename.
UUID
prevents two publications from choosing the same physical filename
writer authority
says which component owns a logical Event or Status resource
Two processes can generate perfectly different UUIDs and still contradict one another by both claiming to publish the current value of the same Status. For append-only Events, UUIDs are exactly what allow many distinct publications to coexist. The ownership question becomes important when those publications claim to describe one logical resource—especially current State.
We introduce the distinction here, after I/S/E, because only now do we have the vocabulary needed to make it useful.
For example:
ConsoleIn/E/...
is written by ConsoleIn.
ROT13/E/...
is written by ROT13.
ROT13/S/...
is written by ROT13.
For this small exercise, choose one designated authority for Intent directed at ROT13. A larger system may deliberately allow several command sources; if so, that is an explicit policy rather than a filename-collision problem.
Readers may be many.
Subscribers may be many.
But ownership of a writable resource is clear.
This avoids:
Component A ─┐
├──► shared-state
Component B ─┘
If two components both own the same fact, ask:
Why are there two authorities?
Do not immediately solve every architectural ownership problem with locks.
Part VII — Now Merge the Two Architectural Ideas
48. Did We Just Prove That Mediators Are Bad?
No.
Absolutely not.
We proved that a mediator that performs only:
A → A
may be unnecessary.
Now consider another case.
Producer publishes:
{
"value": "Hello"
}
Consumer expects:
{
"display": {
"text": "Hello",
"durationMs": 5000
}
}
Can the consumer simply subscribe to the producer?
Transport-wise:
yes
Semantically:
no
The data contracts are incompatible.
Now mediation has a real job.
49. A Transformer Has Earned Its Existence
Introduce:
TextToDisplayTransformer
Its architecture is:
Producer
│
│ {"value":"Hello"}
▼
Transformer
│
│ {"display":{...}}
▼
Consumer
This mediator is not an expensive identity function.
It contributes:
schema transformation
semantic adaptation
possibly unit conversion
possibly aggregation
That is useful.
50. The Transformer Should Also Be Egoistic
Do not make the transformer special.
It is just another service.
Conceptually:
Transformer/I
Transformer/S/...
Transformer/E/...
It can itself receive:
SUBSCRIBE SomeProducer/E/text
through Intent.
It transforms compatible input into its own Event contract.
Then the final consumer may dynamically subscribe to:
Transformer/E/display
So even explicit mediation can still participate in our dynamic architecture.
51. The Final Rule for Mediation
We now have a more precise principle:
If a consumer semantically understands a producer's publication, direct subscription is often sufficient.
If the contracts are incompatible, introduce a transformer/mediator that performs an actual adaptation.
Or more provocatively:
A mediator that merely forwards compatible data is suspicious.
A mediator that adds necessary semantics has earned its existence.
That is the architectural distinction we were looking for.
52. The Complete Filesystem Architecture
We have now reached:
Intent
│
▼
┌────────────┐
Events ────────►│ Service │──────► Status
│ │
└─────┬──────┘
│
▼
Events
A service:
owns its capability
owns its Status
owns its Events
receives Intent
may dynamically observe compatible
foreign Events or Status
does not need its neighbours
compiled into its capability
If data is incompatible:
Producer
│
▼
Transformer
│
▼
Consumer
And all of this currently runs over...
files.
Lots and lots of files.
Part VIII — The Filesystem Has Done Its Job
53. What Our Filesystem Transport Has Taught Us
We have experienced:
- publication boundaries;
- file visibility;
- directories as addresses;
- observation;
- independent processes;
- independent languages;
- synchronization latency;
- runtime wiring;
- State;
- Intent;
- Events;
- mediator costs;
- service ownership;
- transformer services.
Excellent.
The filesystem was a good teacher.
It is not a particularly elegant message broker.
We currently deal with:
directories
.tmp files
atomic moves
.msg files
WatchService
rescanning
shared paths
Syncthing
synchronization delay
The architecture is getting clean.
The transport is increasingly ridiculous.
For small coordination messages, that is.
The filesystem has an advantage MQTT cannot pretend to match: it routinely stores gigabytes or terabytes, while MQTT brokers and clients are designed for comparatively small messages. Large MQTT payloads encounter broker limits, memory pressure, retransmission cost and uncomfortable failure handling. Chunking a huge file over MQTT merely means that we have started inventing a file-transfer protocol inside a message protocol.
So the conclusion is not:
files bad; MQTT good.
It is:
Use messages for coordination and events. Use an appropriate file or object store for large data. A message can carry the location, identity and metadata of that data instead of carrying the terabytes themselves.
A replicated filesystem also gives every participating machine a local copy. That is a valuable backup-like property and may be exactly what an application needs. Strictly speaking, replication alone is not a complete backup strategy: deletion or corruption may also replicate, unless versioning or a separate backup policy preserves older copies. Nevertheless, distributed storage is a real strength of the filesystem version, not an embarrassment we should hide.
Now MQTT is allowed to enter.
Part IX — MQTT
54. MQTT Should Look Familiar
Map what we already know:
| Filesystem / Syncthing | MQTT |
|---|---|
| synchronized message-board root | broker / namespace |
| directory path | topic |
.msg file |
publication |
| file contents | payload |
WatchService |
subscription |
| dynamic directory | topic / filter |
| component control directory | I |
| state directory | S/... |
| event directory | E/... |
At this point, the mapping should look familiar:
Oh. We already know this architecture.
Exactly.
55. Touch MQTT Without Java
Before using a library, install and start Eclipse Mosquitto. Mosquitto is the MQTT broker: clients connect to it, and it routes publications to matching subscriptions. For a local experiment, start it in verbose mode:
mosquitto -v
The default local address used below is:
tcp://localhost:1883
Then use MQTT Explorer, a graphical MQTT client that lets you inspect the topic hierarchy and publish messages by hand.
Connect to the broker.
Publish:
Tutorial/ConsoleIn/U/alice/E/text
with:
{"value":"Hello"}
Observe it.
Try another client.
Subscribe to a topic hierarchy.
See what the broker does.
Learn:
broker
client
topic
publish
subscribe
before adding the MQTT Java API.
56. Only Now Do We Need an MQTT Library
Java does not provide a convenient MQTT client in its standard library.
Now a library solves an actual problem.
Use the Eclipse Paho Java MQTT 5 client.
For Maven:
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
<version>1.2.5</version>
</dependency>
Initially we need only:
connect
publish
subscribe
callback
The complete Maven configuration—including Paho and the Jackson dependencies
introduced later—is
pom.xml.
57. Connect
var broker =
"tcp://localhost:1883";
var clientId =
UUID.randomUUID().toString();
var client =
new MqttAsyncClient(
broker,
clientId,
new MemoryPersistence()
);
client.connect().waitForCompletion();
Nothing architectural happened here.
We merely attached our process to the transport.
58. Publish
A small local function now makes sense:
void publish(
IMqttAsyncClient client,
String topic,
String content,
int qos,
boolean retained)
throws Exception {
var message =
new MqttMessage(
content.getBytes(
StandardCharsets.UTF_8
)
);
message.setQos(qos);
message.setRetained(retained);
client.publish(topic, message)
.waitForCompletion();
}
Then:
publish(
client,
"Tutorial/ConsoleIn/U/alice/E/text",
"{\"value\":\"Hello\"}",
1,
false
);
Compare that with:
FileMessageBoard.publish(
"Tutorial/ConsoleIn/U/alice/E/text",
...
);
The conceptual operation is the same.
59. Subscribe
client.subscribe(
"Tutorial/ConsoleIn/U/alice/E/text",
1
);
But receiving is asynchronous.
So we need a callback.
60. React to Publications
client.setCallback(
new MqttCallback() {
@Override
public void messageArrived(
String topic,
MqttMessage message)
throws Exception {
IO.println(
topic
+ " -> "
+ new String(
message.getPayload()
)
);
}
// other callbacks omitted here
}
);
Our:
WatchService
has effectively become:
MQTT callback
Again:
transport changed
The architecture did not.
61. Rebuild Our Three Services
ConsoleIn
Publishes:
Tutorial/ConsoleIn/U/<unit>/E/text
ROT13
Owns:
Tutorial/ROT13/U/<unit>/I
Tutorial/ROT13/U/<unit>/S/...
Tutorial/ROT13/U/<unit>/E/crypted
ConsoleOut
Owns:
Tutorial/ConsoleOut/U/<unit>/I
Tutorial/ConsoleOut/U/<unit>/S/...
The service responsibilities are the same as before.
The runnable MQTT programs are:
They intentionally do not depend on a small home-made MQTT framework. Each runnable file contains its own connection, publication, callback and Last-Will code where it needs that code. This creates some duplication, but you can open one file and see how that complete program behaves.
Only two focused helpers remain shared:
JsonTools.javaconfigures Jackson;MqttTools.javavalidates and matches MQTT topic filters.
During your own project, repeated code may eventually become annoying enough that you choose to extract a library. Make that decision after you have felt the repetition and can name the problem the library should solve.
From supplemental-java/06-mqtt, compile once and copy the dependencies:
mvn package dependency:copy-dependencies
On Linux or macOS, use three terminals:
java -cp 'target/classes:target/dependency/*' MqttRot13 alice
java -cp 'target/classes:target/dependency/*' MqttConsoleOut alice
java -cp 'target/classes:target/dependency/*' MqttConsoleIn alice
Then configure the running services:
java -cp 'target/classes:target/dependency/*' MqttConfigure alice alice alice
On Windows, replace the classpath separator : with ;.
62. Deliberately Make the Old Mistake Once
Hard-code:
client.subscribe(
"Tutorial/ConsoleIn/U/alice/E/text",
1
);
inside ROT13.
It works beautifully.
Then ask:
Who knows whom?
ROT13 now contains:
alice
as topology.
We know how to fix this.
The filesystem tutorial already taught us the architecture.
Now we merely implement it using MQTT.
63. We Need Structured Intent
Our primitive filesystem message:
TOPIC some/path
was sufficient for learning the architecture.
Now we want richer Intent.
For example:
{
"subscribe": {
"topic":
"Tutorial/ConsoleIn/U/alice/E/text"
}
}
or:
{
"text": {
"value": "Hello"
}
}
or both.
At this point manual parsing becomes distracting.
Manual parsing would now distract from the architecture, so a serialization library has a clear job.
64. Enter Jackson
We want:
{"value":"Hello"}
to become:
record Rot13Message(
String value
) {}
and:
{
"subscribe": {
"topic": "..."
}
}
to become:
record Rot13Subscription(
String topic
) {}
record Rot13Intent(
Rot13Subscription subscribe,
Rot13Message text
) {}
This is exactly the problem Jackson solves.
Keep these tiny records in the runnable service file that uses them. There is no separate message-model layer to understand before you can read the service.
Add both Jackson modules used by the example. jackson-databind performs object
mapping; jackson-dataformat-yaml lets the same mapper accept YAML Intent:
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.22.2</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-yaml</artifactId>
<version>2.22.2</version>
</dependency>
Keep the two Jackson versions aligned.
65. Minimal JsonTools
A helper can centralize the configuration:
final class JsonTools {
public static final ObjectMapper
deserializationObjectMapper =
new ObjectMapper(
new YAMLFactory()
);
public static final ObjectMapper
serializationObjectMapper =
new ObjectMapper();
static {
deserializationObjectMapper
.configure(
DeserializationFeature
.FAIL_ON_UNKNOWN_PROPERTIES,
false
);
serializationObjectMapper
.configure(
SerializationFeature
.INDENT_OUTPUT,
true
);
}
}
Complete source:
JsonTools.java.
We can now accept JSON and human-friendly YAML Intent.
The helper is useful because we already understand what serialization problem it hides.
66. Deserialize Intent
Rot13Intent intent =
JsonTools
.deserializationObjectMapper
.readValue(
message.getPayload(),
Rot13Intent.class
);
Direct Intent:
text:
value: Hello
can cause:
cryptAndPublish(
intent.text().value()
);
Again, command-like Intent is perfectly valid inside an Event-Driven system.
67. Dynamic MQTT Wiring
ROT13 always subscribes to its own Intent:
Tutorial/ROT13/U/<unit>/I
Now publish:
subscribe:
topic: Tutorial/ConsoleIn/U/alice/E/text
ROT13 executes:
client.subscribe(
intent.subscribe().topic(),
1
);
Exactly the same architectural operation as:
TOPIC consoleIn/outbox
in our file system.
Only the transport and serialization changed.
68. Rewire at Runtime
Publish later:
subscribe:
topic: Tutorial/ConsoleIn/U/bob/E/text
If ROT13 accepts one input source at a time:
if (dynamicTextTopic != null) {
client.unsubscribe(
dynamicTextTopic
);
}
Then:
dynamicTextTopic =
intent.subscribe().topic();
client.subscribe(
dynamicTextTopic,
1
);
No restart.
No recompilation.
No forwarding mediator.
69. Validate Dynamic Topic Filters
We now accept externally supplied topic filters.
That string affects transport behaviour.
Therefore validate it.
Use a small validation helper for this boundary:
MqttTools.isValidTopic(...)
MqttTools.isValidFilter(...)
MqttTools.isTopicMatchingFilter(...)
For example:
if (MqttTools.isValidFilter(
intent.subscribe().topic())) {
// accept subscription
}
The helper is small, and its purpose is now visible: untrusted configuration must not become an unchecked broker subscription.
Complete source:
MqttTools.java.
70. Status Should Explain Current Behaviour
After subscribing, publish:
Tutorial/ROT13/U/<unit>/S/subscriptions/text
with:
{
"value":
"Tutorial/ConsoleIn/U/alice/E/text"
}
Now the service can explain:
why it currently reacts
to Alice's Events
That is Status.
Not Event.
71. Retained State, Non-Retained Events
An Event:
buttonPressed
usually means:
something happened at a particular time
A client joining later normally should not receive it as if it just occurred.
Therefore Events are typically:
retained = false
Status such as:
online = true
represents current condition.
A late subscriber may want that immediately.
Therefore Status may sensibly use:
retained = true
MQTT now provides transport semantics that fit our architectural distinction.
72. Last Will
A service can normally publish:
{"value":false}
when it shuts down.
But not if:
the machine crashes
the process dies
the network disappears
MQTT provides Last Will.
Configure:
var lastWillTopic =
statusTopic + "/online";
MqttMessage will =
new MqttMessage(
"{\"value\":false}".getBytes(),
1,
true,
new MqttProperties()
);
connectionOptions.setWill(
lastWillTopic,
will
);
After connecting:
publish(
client,
lastWillTopic,
"{\"value\":true}",
1,
true
);
Again ask:
Why is this Status rather than Event?
The runnable MqttRot13.java and MqttConsoleOut.java files each contain this
Last-Will setup themselves. The duplication stays visible: each program can be
read on its own, and you can decide later whether the repetition has become
painful enough to extract.
73. At Last: DUISE
Only now do we give the complete structure a compact name:
D — Description
U — Unit
I — Intent
S — Status
E — Event
Example:
Tutorial/
└── ROT13/
└── U/
└── alice/
├── I
│
├── S/
│ ├── online
│ └── subscriptions/
│ └── text
│
└── E/
└── crypted
D — Description
What data contracts does this service understand and publish?
U — Unit
Which concrete actor/service instance?
I — Intent
What behaviour is requested?
S — Status
What is currently true?
E — Event
What happened?
Hopefully none of these concepts now feels arbitrary.
We needed every one of them before we gave it a letter.
74. Compatible Services: No Explicit Mediator
Suppose ConsoleIn publishes:
{
"value": "Hello"
}
and ROT13 understands exactly that structure.
Then:
ConsoleIn/E/text
│
▼
ROT13
is sufficient.
The broker already handles transport-level delivery.
Likewise:
ROT13/E/crypted
│
▼
ConsoleOut
if ConsoleOut understands that contract.
No explicit transformer is required.
75. Incompatible Services: Transformer Required
Suppose ROT13 produces:
{
"value": "Uryyb"
}
but an actuator expects:
{
"display": {
"text": "Uryyb",
"durationMs": 5000
}
}
Then introduce:
ROT13
│
▼
DisplayTransformer
│
▼
Display
The transformer:
- receives compatible input;
- transforms it;
- publishes its own Event contract.
It too can be dynamically wired.
Our mediation rule survives the move to MQTT unchanged.
76. What Changed From Files to MQTT?
Compare:
FILESYSTEM + SYNCTHING MQTT
directory → topic
.msg → publication
file contents → payload
WatchService → subscription
Syncthing → broker routing
dynamic directory → topic/filter
I → I
S → S
E → E
egoistic service → egoistic service
runtime topology → runtime topology
transformer if needed → transformer if needed
The lower half is the important part.
The architecture survived the transport replacement.
77. What Coupling Still Remains?
Loose coupling does not mean:
nothing knows anything
Services still need agreement about:
- message syntax;
- message semantics;
- topic contracts;
- Intent;
- ownership;
- meaning of State;
- meaning of Events.
The goal is not:
eliminate all coupling.
The goal is:
eliminate unnecessary knowledge.
78. Final Coupling Test
For any component, ask:
- Does it know another component's Java class?
- Does it require another component in the same JVM?
- Does it know another component's machine?
- Does it know another component's programming language?
- Does its capability contain a hard-coded producer identity?
- Does it write State owned by somebody else?
- Are multiple actors competing to own the same resource?
- Which foreign message contracts must it genuinely understand?
- Where is system topology defined?
- Can topology change without recompiling the capability?
- Could the transport be replaced without changing the core capability?
- Does every mediator perform something that actually needs mediation?
The architecture does not necessarily answer:
NO
to every question.
But the answer should always be intentional.
79. The Whole Journey
1. PROCEDURAL
────────────────────────────────
ConsoleIn
│ call
▼
ROT13
│ call
▼
ConsoleOut
2. EGOISTIC SERVICES
────────────────────────────────
ConsoleIn ROT13 ConsoleOut
Each knows only its own capability.
Problem:
Who composes the system?
3. STATIC MEDIATION — ONE JVM
────────────────────────────────
ConsoleIn
│
▼
Mediator
│
▼
ROT13
│
▼
Mediator
│
▼
ConsoleOut
Capabilities are egoistic.
Mediators know topology.
4. STATIC MEDIATION — FILESYSTEM
────────────────────────────────
Sensor process
│
▼
Mediator process
│
▼
ROT13 process
│
▼
Mediator process
│
▼
ConsoleOut process
5. STATIC MEDIATION — SYNCTHING
────────────────────────────────
Computer A
Sensor
│
│ synchronized files
▼
Computer B
Mediator
│
▼
ROT13
│
│ synchronized files
▼
Computer C
Mediator
│
▼
ConsoleOut
Now the forwarding MiMs
have become visibly expensive.
6. NAMED, CONFIGURABLE MEDIATORS
────────────────────────────────
Orchestrator
│ ROUTE
▼
ConsoleIn ─────► Mediator ─────► ROT13
The mediator is an ordinary named service.
Topology arrives through its control inbox.
Application data still passes through it.
7. REWIND — DYNAMIC IN-JVM
────────────────────────────────
Intent
│
▼
ConsoleIn ─────► ROT13 ─────► ConsoleOut
Services themselves listen.
Topology arrives as data.
8. DYNAMIC FILESYSTEM
────────────────────────────────
Intent
│
▼
ConsoleIn files ───► ROT13 ───► ConsoleOut files
No forwarding mediator.
9. DYNAMIC SYNCTHING
────────────────────────────────
Computer A Computer B Computer C
ConsoleIn ──sync──────► ROT13 ──sync──────► ConsoleOut
▲
│
Intent
Capabilities remain egoistic.
Topology is external.
10. CONTRACTS MERGE THE TWO IDEAS
────────────────────────────────
Compatible?
YES:
Producer ─────────────► Consumer
NO:
Producer
│
▼
Transformer
│
▼
Consumer
11. MQTT
────────────────────────────────
Intent
│
▼
ConsoleIn/E ──MQTT─► ROT13 ──MQTT─► ROT13/E
│
MQTT
│
▼
ConsoleOut
Same architecture.
Better transport.
80. Your Turn
Build your own system containing at least:
one source-like capability
one processing capability
one sink-like capability
You may use:
filesystem
MQTT
or both
You must be able to explain:
What is my capability?
What Intent do I understand?
What Status do I publish?
What Events do I publish?
Who owns each writable resource?
Which external publications do I observe?
How did I learn where to observe?
Where is the topology?
Why is each mediator present?
Could the system be rewired
without recompiling its capabilities?
And be prepared for:
Is your architecture Event-Driven because you used MQTT?
The answer is:
No.
MQTT is a transport.
The architecture comes from how your services are designed and composed.
81. The Sentences to Remember
Event-driven describes how information and control flow.
Loose coupling describes how much participants need to know about one another.
Publish what happened.
Declare what you want.
Publish your own State.
Keep capabilities egoistic.
Keep topology out of capabilities.
Do not pay for a mediator unless it actually mediates something.
And whenever architecture becomes confusing:
Who knows whom?
Optional Epilogue — Complete the Dynamic Mediator
At the end of Part III, the named ConfigurableMediator accepted one route and
then stopped observing its control inbox. That limitation was real:
FileMessageBoard.watch(...)
watches one mailbox
and blocks
Now you have already used FileSubscriptions, dynamic wiring, Intent and the
control-plane/data-plane distinction. We can finally return to the mediator and
finish the implementation without introducing unexplained machinery.
This is optional. The main architectural journey is already complete.
Keep the Control Inbox Permanently Subscribed
Start the mediator with only its identity:
java DynamicMediatorService.java mediator1
It permanently subscribes to:
mediators/mediator1/inbox
Every valid ROUTE Intent contains one atomic pair:
ROUTE
<from mailbox>
<to mailbox>
When another ROUTE arrives, the mediator:
- validates both mailbox names;
- subscribes to the new source if it changed;
- unsubscribes from the old source;
- replaces its current route;
- keeps running.
The mediator rejects a route whose source and destination are equal. Otherwise it would observe its own copied message, publish another copy into the same mailbox, observe that copy, and continue forever.
It also keeps its control mailbox out of the data route. Configuration and application data remain separate.
Complete source:
DynamicMediatorService.java.
Send Rewiring Intent
The short-lived orchestrator accepts:
mediator name
from mailbox
to mailbox
and publishes the corresponding ROUTE Intent:
java DynamicMediatorOrchestrator.java \
mediator1 \
consoleIn/outbox \
rot13/inbox
Complete source:
DynamicMediatorOrchestrator.java.
Run the Rewirable Version
Work from supplemental-java/05-dynamic-filesystem. Start five long-running
programs with a fresh message-board root:
java -DmessageBoard.root=messageBoard-rewirable SensorService.java
java -DmessageBoard.root=messageBoard-rewirable \
DynamicMediatorService.java mediator1
java -DmessageBoard.root=messageBoard-rewirable \
DynamicProcessorService.java
java -DmessageBoard.root=messageBoard-rewirable \
DynamicMediatorService.java mediator2
java -DmessageBoard.root=messageBoard-rewirable \
DynamicActuatorService.java
After they are waiting, establish the mediated ROT13 path with two short-lived orchestrator calls:
java -DmessageBoard.root=messageBoard-rewirable \
DynamicMediatorOrchestrator.java \
mediator1 consoleIn/outbox rot13/inbox
java -DmessageBoard.root=messageBoard-rewirable \
DynamicMediatorOrchestrator.java \
mediator2 rot13/outbox consoleOut/inbox
Type:
Hello
ConsoleOut should receive:
Uryyb
Now rewire mediator1 while every long-running process remains alive:
java -DmessageBoard.root=messageBoard-rewirable \
DynamicMediatorOrchestrator.java \
mediator1 consoleIn/outbox consoleOut/inbox
Type:
Hello
This time ConsoleOut should receive the unmodified text:
Hello
Restore the ROT13 route without restarting the mediator:
java -DmessageBoard.root=messageBoard-rewirable \
DynamicMediatorOrchestrator.java \
mediator1 consoleIn/outbox rot13/inbox
The next input passes through ROT13 again.
Be Honest About the Switching Boundary
The ROUTE message keeps from and to together as one configuration
request. That does not make the distributed route change globally atomic.
During a switch:
- a file may already be waiting in the observer's internal queue;
- Syncthing may expose files on different computers at different times;
- an application message may therefore follow the old or the new route near the boundary;
- this small example provides no acknowledgements, draining protocol or exactly-once guarantee.
The implementation ignores queued messages from a source that is no longer the current route. This gives the example a clear local rule, not a distributed transaction.
The Final Lesson
The mediator is now:
named
egoistic
configured through Intent
live-rewirable
still in the data path
So a mediator can be designed as a perfectly respectable service and still be unnecessary for a particular connection.
Dynamic configurability answers:
Can this mediator receive a new route without restarting?
It does not answer:
Does this mediator contribute semantics worth paying for?
Those remain two different architectural questions.
Appendix A — Complete Java Sources
These are the same files linked throughout the tutorial. The compact source-file style is deliberate: start them quickly, change them, break them, and observe what the architecture does.
A.1 — Procedural
ProceduralRot13.java
String rot13(String text) {
var result = new StringBuilder();
for (var character : text.toCharArray()) {
if (character >= 'a' && character <= 'z')
character = (char) ('a' + (character - 'a' + 13) % 26);
else if (character >= 'A' && character <= 'Z')
character = (char) ('A' + (character - 'A' + 13) % 26);
result.append(character);
}
return result.toString();
}
void main() {
while (true) {
var text = IO.readln();
if (text == null)
return;
IO.println(rot13(text));
}
}
A.2 — Static mediation inside one JVM
StaticQueueRot13.java
record Message(String type, String value) {}
Map<String, BlockingQueue<Message>> mailboxes =
new ConcurrentHashMap<>();
BlockingQueue<Message> mailbox(String name) {
return mailboxes.computeIfAbsent(
name,
_ -> new LinkedBlockingQueue<>()
);
}
String rot13(String text) {
var result = new StringBuilder();
for (var character : text.toCharArray()) {
if (character >= 'a' && character <= 'z')
character = (char) ('a' + (character - 'a' + 13) % 26);
else if (character >= 'A' && character <= 'Z')
character = (char) ('A' + (character - 'A' + 13) % 26);
result.append(character);
}
return result.toString();
}
void mediate(String from, String to) {
try {
while (true)
mailbox(to).put(mailbox(from).take());
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
void runConsoleIn() {
try {
while (true) {
var text = IO.readln();
if (text == null)
return;
mailbox("consoleIn.outbox")
.put(new Message("TEXT", text));
}
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
void runRot13() {
try {
while (true) {
var input = mailbox("rot13.inbox").take();
var output = new Message("TEXT", rot13(input.value()));
mailbox("rot13.outbox").put(output);
}
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
void runConsoleOut() {
try {
while (true)
IO.println(mailbox("consoleOut.inbox").take().value());
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
void main() throws InterruptedException {
Thread.ofVirtual().start(this::runConsoleIn);
Thread.ofVirtual().start(this::runRot13);
Thread.ofVirtual().start(this::runConsoleOut);
Thread.ofVirtual().start(
() -> mediate("consoleIn.outbox", "rot13.inbox")
);
Thread.ofVirtual().start(
() -> mediate("rot13.outbox", "consoleOut.inbox")
);
Thread.currentThread().join();
}
A.3 — Static and named filesystem mediation
Message.java
import java.time.Instant;
record Message(String type, String value, Instant timestamp) {
Message(String type, String value) {
this(type, value, Instant.now());
}
}
FileMessageBoard.java
import java.io.IOException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardWatchEventKinds;
import java.time.Instant;
import java.util.Arrays;
import java.util.UUID;
import java.util.function.Consumer;
final class FileMessageBoard {
static final Path ROOT = Path.of(
System.getProperty("messageBoard.root", "messageBoard")
).toAbsolutePath().normalize();
private FileMessageBoard() {}
static String serialize(Message message) {
var text = message.type() + "\n" + message.value();
if (message.timestamp() != null)
text += "\nTIMESTAMP: " + message.timestamp();
return text;
}
static Message read(Path file) throws IOException {
var lines = Files.readString(file).split("\\R", -1);
if (lines.length < 2)
throw new IOException("A message needs a type and a value: " + file);
var endOfValue = lines.length;
Instant timestamp = null;
var lastLine = lines[lines.length - 1];
if (lastLine.startsWith("TIMESTAMP: ")) {
timestamp = Instant.parse(lastLine.substring("TIMESTAMP: ".length()));
endOfValue--;
}
var value = String.join(
"\n",
Arrays.copyOfRange(lines, 1, endOfValue)
);
return new Message(lines[0], value, timestamp);
}
static void publish(String mailbox, Message message) throws IOException {
publishText(mailbox, serialize(message));
}
static void publishText(String mailbox, String content) throws IOException {
var directory = directory(mailbox);
Files.createDirectories(directory);
var id = UUID.randomUUID().toString();
var temporary = directory.resolve(id + ".tmp");
var published = directory.resolve(id + ".msg");
Files.writeString(temporary, content);
Files.move(temporary, published, StandardCopyOption.ATOMIC_MOVE);
}
static void watch(String mailbox, Consumer<Message> receiver)
throws IOException, InterruptedException {
var directory = directory(mailbox);
Files.createDirectories(directory);
try (var watcher = FileSystems.getDefault().newWatchService()) {
directory.register(watcher, StandardWatchEventKinds.ENTRY_CREATE);
while (true) {
var key = watcher.take();
for (var event : key.pollEvents()) {
if (event.kind() == StandardWatchEventKinds.OVERFLOW)
continue;
var filename = (Path) event.context();
if (!filename.toString().endsWith(".msg"))
continue;
var file = directory.resolve(filename);
if (Files.isRegularFile(file))
receiver.accept(read(file));
}
if (!key.reset())
return;
}
}
}
static Path directory(String name) {
var directory = ROOT.resolve(name).normalize();
if (!directory.startsWith(ROOT))
throw new IllegalArgumentException("Mailbox escapes the message board");
return directory;
}
}
SensorService.java
void main() throws Exception {
while (true) {
var text = IO.readln();
if (text == null)
return;
FileMessageBoard.publish(
"consoleIn/outbox",
new Message("TEXT", text)
);
}
}
ProcessorService.java
String rot13(String text) {
var result = new StringBuilder();
for (var character : text.toCharArray()) {
if (character >= 'a' && character <= 'z')
character = (char) ('a' + (character - 'a' + 13) % 26);
else if (character >= 'A' && character <= 'Z')
character = (char) ('A' + (character - 'A' + 13) % 26);
result.append(character);
}
return result.toString();
}
void main() throws Exception {
FileMessageBoard.watch("rot13/inbox", message -> {
if (!message.type().equals("TEXT"))
return;
try {
FileMessageBoard.publish(
"rot13/outbox",
new Message("TEXT", rot13(message.value()))
);
} catch (IOException exception) {
throw new UncheckedIOException(exception);
}
});
}
ActuatorService.java
void main() throws Exception {
FileMessageBoard.watch("consoleOut/inbox", message -> {
if (message.type().equals("TEXT"))
IO.println(message.value());
});
}
Mediator.java
void main(String[] arguments) throws Exception {
if (arguments.length != 2) {
IO.println("Usage: java Mediator.java <from-mailbox> <to-mailbox>");
return;
}
var from = arguments[0];
var to = arguments[1];
IO.println("Mediating: " + from + " -> " + to);
FileMessageBoard.watch(from, message -> {
try {
FileMessageBoard.publish(to, message);
} catch (IOException exception) {
throw new UncheckedIOException(exception);
}
});
}
ConfigurableMediator.java
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.concurrent.LinkedBlockingQueue;
record Route(String from, String to) {}
Route waitForRoute(String controlMailbox) throws InterruptedException {
var messages = new LinkedBlockingQueue<Message>();
var controlListener = Thread.ofVirtual().start(() -> {
try {
FileMessageBoard.watch(controlMailbox, messages::add);
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
} catch (IOException exception) {
throw new UncheckedIOException(exception);
}
});
try {
while (true) {
var message = messages.take();
if (!message.type().equals("ROUTE")) {
IO.println("Ignored " + message.type() + "; expected ROUTE");
continue;
}
var mailboxes = message.value().lines().toList();
if (mailboxes.size() != 2
|| mailboxes.get(0).isBlank()
|| mailboxes.get(1).isBlank()) {
IO.println("ROUTE needs exactly two lines: <from> and <to>");
continue;
}
return new Route(mailboxes.get(0), mailboxes.get(1));
}
} finally {
controlListener.interrupt();
controlListener.join();
}
}
void main(String[] arguments) throws Exception {
if (arguments.length != 1
|| !arguments[0].matches("[A-Za-z0-9_-]+")) {
IO.println("Usage: java ConfigurableMediator.java <mediator-name>");
return;
}
var name = arguments[0];
var controlMailbox = "mediators/" + name + "/inbox";
IO.println(name + " waits for ROUTE on " + controlMailbox);
var route = waitForRoute(controlMailbox);
IO.println(name + " mediates: " + route.from() + " -> " + route.to());
FileMessageBoard.watch(route.from(), message -> {
try {
FileMessageBoard.publish(route.to(), message);
} catch (IOException exception) {
throw new UncheckedIOException(exception);
}
});
}
MediatorOrchestrator.java
void configure(String mediator, String from, String to) throws Exception {
FileMessageBoard.publish(
"mediators/" + mediator + "/inbox",
new Message("ROUTE", from + "\n" + to)
);
}
void main() throws Exception {
configure(
"mediator1",
"consoleIn/outbox",
"rot13/inbox"
);
configure(
"mediator2",
"rot13/outbox",
"consoleOut/inbox"
);
IO.println("Configured mediator1 and mediator2");
}
A.4 — Dynamic queues inside one JVM
DynamicQueueRot13.java
record Message(String type, String value) {}
Map<String, BlockingQueue<Message>> mailboxes =
new ConcurrentHashMap<>();
BlockingQueue<Message> mailbox(String name) {
return mailboxes.computeIfAbsent(
name,
_ -> new LinkedBlockingQueue<>()
);
}
void listen(String mailboxName, Consumer<Message> receiver) {
try {
while (true)
receiver.accept(mailbox(mailboxName).take());
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
String rot13(String text) {
var result = new StringBuilder();
for (var character : text.toCharArray()) {
if (character >= 'a' && character <= 'z')
character = (char) ('a' + (character - 'a' + 13) % 26);
else if (character >= 'A' && character <= 'Z')
character = (char) ('A' + (character - 'A' + 13) % 26);
result.append(character);
}
return result.toString();
}
void runConsoleIn() {
try {
while (true) {
var text = IO.readln();
if (text == null)
return;
mailbox("consoleIn.outbox")
.put(new Message("TEXT", text));
}
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
Map<String, Thread> rot13Subscriptions = new HashMap<>();
Map<String, Thread> consoleOutSubscriptions = new HashMap<>();
void subscribe(
Map<String, Thread> subscriptions,
String mailboxName,
Consumer<Message> receiver) {
subscriptions.computeIfAbsent(
mailboxName,
name -> Thread.ofVirtual().start(() -> listen(name, receiver))
);
}
void processWithRot13(Message message) {
if (!message.type().equals("TEXT"))
return;
try {
mailbox("rot13.outbox").put(
new Message("TEXT", rot13(message.value()))
);
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
void runRot13() {
try {
while (true) {
var message = mailbox("rot13.inbox").take();
if (message.type().equals("TOPIC"))
subscribe(
rot13Subscriptions,
message.value(),
this::processWithRot13
);
else
processWithRot13(message);
}
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
void print(Message message) {
if (message.type().equals("TEXT"))
IO.println(message.value());
}
void runConsoleOut() {
try {
while (true) {
var message = mailbox("consoleOut.inbox").take();
if (message.type().equals("TOPIC"))
subscribe(
consoleOutSubscriptions,
message.value(),
this::print
);
else
print(message);
}
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
}
}
void main() throws Exception {
Thread.ofVirtual().start(this::runConsoleIn);
Thread.ofVirtual().start(this::runRot13);
Thread.ofVirtual().start(this::runConsoleOut);
mailbox("rot13.inbox").put(
new Message("TOPIC", "consoleIn.outbox")
);
mailbox("consoleOut.inbox").put(
new Message("TOPIC", "rot13.outbox")
);
Thread.currentThread().join();
}
A.5 — Dynamic filesystem subscriptions
Message.java
import java.time.Instant;
record Message(String type, String value, Instant timestamp) {
Message(String type, String value) {
this(type, value, Instant.now());
}
}
FileMessageBoard.java
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.time.Instant;
import java.util.Arrays;
import java.util.UUID;
final class FileMessageBoard {
static final Path ROOT = Path.of(
System.getProperty("messageBoard.root", "messageBoard")
).toAbsolutePath().normalize();
private FileMessageBoard() {}
static String serialize(Message message) {
var text = message.type() + "\n" + message.value();
if (message.timestamp() != null)
text += "\nTIMESTAMP: " + message.timestamp();
return text;
}
static Message read(Path file) throws IOException {
var lines = Files.readString(file).split("\\R", -1);
if (lines.length < 2)
throw new IOException("A message needs a type and a value: " + file);
var endOfValue = lines.length;
Instant timestamp = null;
var lastLine = lines[lines.length - 1];
if (lastLine.startsWith("TIMESTAMP: ")) {
timestamp = Instant.parse(lastLine.substring("TIMESTAMP: ".length()));
endOfValue--;
}
var value = String.join(
"\n",
Arrays.copyOfRange(lines, 1, endOfValue)
);
return new Message(lines[0], value, timestamp);
}
static void publish(String mailbox, Message message) throws IOException {
var directory = directory(mailbox);
Files.createDirectories(directory);
var id = UUID.randomUUID().toString();
var temporary = directory.resolve(id + ".tmp");
var published = directory.resolve(id + ".msg");
Files.writeString(temporary, serialize(message));
Files.move(temporary, published, StandardCopyOption.ATOMIC_MOVE);
}
static Path directory(String name) {
var directory = ROOT.resolve(name).normalize();
if (!directory.startsWith(ROOT))
throw new IllegalArgumentException("Mailbox escapes the message board");
return directory;
}
}
FileSubscriptions.java
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.ClosedWatchServiceException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardWatchEventKinds;
import java.nio.file.WatchKey;
import java.nio.file.WatchService;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
record IncomingMessage(String mailbox, Message message) {}
final class FileSubscriptions implements AutoCloseable {
private final WatchService watcher;
private final Map<String, WatchKey> subscriptions = new ConcurrentHashMap<>();
private final Map<WatchKey, String> mailboxes = new ConcurrentHashMap<>();
private final BlockingQueue<IncomingMessage> messages =
new LinkedBlockingQueue<>();
private final Thread observer;
FileSubscriptions() throws IOException {
watcher = FileSystems.getDefault().newWatchService();
observer = Thread.ofVirtual().start(this::observe);
}
void subscribe(String mailbox) throws IOException {
if (subscriptions.containsKey(mailbox))
return;
var directory = FileMessageBoard.directory(mailbox);
Files.createDirectories(directory);
var key = directory.register(
watcher,
StandardWatchEventKinds.ENTRY_CREATE
);
subscriptions.put(mailbox, key);
mailboxes.put(key, mailbox);
}
void unsubscribe(String mailbox) {
var key = subscriptions.remove(mailbox);
if (key != null) {
mailboxes.remove(key);
key.cancel();
}
}
IncomingMessage take() throws InterruptedException {
return messages.take();
}
private void observe() {
try {
while (true) {
var key = watcher.take();
var mailbox = mailboxes.get(key);
if (mailbox == null) {
key.reset();
continue;
}
var directory = FileMessageBoard.directory(mailbox);
for (var event : key.pollEvents()) {
if (event.kind() == StandardWatchEventKinds.OVERFLOW)
continue;
var filename = (Path) event.context();
if (!filename.toString().endsWith(".msg"))
continue;
var file = directory.resolve(filename);
if (Files.isRegularFile(file))
messages.put(
new IncomingMessage(mailbox, FileMessageBoard.read(file))
);
}
if (!key.reset()) {
subscriptions.remove(mailbox);
mailboxes.remove(key);
}
}
} catch (ClosedWatchServiceException _) {
// close() ends the observer.
} catch (InterruptedException _) {
Thread.currentThread().interrupt();
} catch (IOException exception) {
throw new UncheckedIOException(exception);
}
}
public void close() throws IOException {
watcher.close();
observer.interrupt();
}
}
SensorService.java
void main() throws Exception {
while (true) {
var text = IO.readln();
if (text == null)
return;
FileMessageBoard.publish(
"consoleIn/outbox",
new Message("TEXT", text)
);
}
}
DynamicProcessorService.java
String rot13(String text) {
var result = new StringBuilder();
for (var character : text.toCharArray()) {
if (character >= 'a' && character <= 'z')
character = (char) ('a' + (character - 'a' + 13) % 26);
else if (character >= 'A' && character <= 'Z')
character = (char) ('A' + (character - 'A' + 13) % 26);
result.append(character);
}
return result.toString();
}
void main() throws Exception {
try (var subscriptions = new FileSubscriptions()) {
var controlMailbox = "rot13/inbox";
String textMailbox = null;
subscriptions.subscribe(controlMailbox);
while (true) {
var incoming = subscriptions.take();
var message = incoming.message();
if (incoming.mailbox().equals(controlMailbox)
&& message.type().equals("TOPIC")) {
if (textMailbox != null)
subscriptions.unsubscribe(textMailbox);
textMailbox = message.value();
subscriptions.subscribe(textMailbox);
IO.println("ROT13 now listens to " + textMailbox);
} else if (message.type().equals("TEXT")) {
FileMessageBoard.publish(
"rot13/outbox",
new Message("TEXT", rot13(message.value()))
);
}
}
}
}
DynamicActuatorService.java
void main() throws Exception {
try (var subscriptions = new FileSubscriptions()) {
var controlMailbox = "consoleOut/inbox";
String textMailbox = null;
subscriptions.subscribe(controlMailbox);
while (true) {
var incoming = subscriptions.take();
var message = incoming.message();
if (incoming.mailbox().equals(controlMailbox)
&& message.type().equals("TOPIC")) {
if (textMailbox != null)
subscriptions.unsubscribe(textMailbox);
textMailbox = message.value();
subscriptions.subscribe(textMailbox);
IO.println("ConsoleOut now listens to " + textMailbox);
} else if (message.type().equals("TEXT")) {
IO.println(message.value());
}
}
}
}
Configure.java
void main() throws Exception {
FileMessageBoard.publish(
"rot13/inbox",
new Message("TOPIC", "consoleIn/outbox")
);
FileMessageBoard.publish(
"consoleOut/inbox",
new Message("TOPIC", "rot13/outbox")
);
IO.println("Wired: consoleIn -> rot13 -> consoleOut");
}
DynamicMediatorService.java
record DynamicRoute(String from, String to) {}
DynamicRoute readRoute(Message message, String controlMailbox) {
var mailboxes = message.value().lines().toList();
if (mailboxes.size() != 2
|| mailboxes.get(0).isBlank()
|| mailboxes.get(1).isBlank()) {
IO.println("ROUTE needs exactly two lines: <from> and <to>");
return null;
}
var route = new DynamicRoute(mailboxes.get(0), mailboxes.get(1));
try {
FileMessageBoard.directory(route.from());
FileMessageBoard.directory(route.to());
} catch (IllegalArgumentException exception) {
IO.println("Rejected ROUTE: " + exception.getMessage());
return null;
}
if (route.from().equals(route.to())) {
IO.println("Rejected ROUTE: from and to must differ");
return null;
}
if (route.from().equals(controlMailbox)
|| route.to().equals(controlMailbox)) {
IO.println("Rejected ROUTE: the control mailbox is not a data mailbox");
return null;
}
return route;
}
void main(String[] arguments) throws Exception {
if (arguments.length != 1
|| !arguments[0].matches("[A-Za-z0-9_-]+")) {
IO.println("Usage: java DynamicMediatorService.java <mediator-name>");
return;
}
var name = arguments[0];
var controlMailbox = "mediators/" + name + "/inbox";
try (var subscriptions = new FileSubscriptions()) {
DynamicRoute route = null;
subscriptions.subscribe(controlMailbox);
IO.println(name + " waits for ROUTE on " + controlMailbox);
while (true) {
var incoming = subscriptions.take();
var message = incoming.message();
if (incoming.mailbox().equals(controlMailbox)) {
if (!message.type().equals("ROUTE")) {
IO.println("Ignored " + message.type() + "; expected ROUTE");
continue;
}
var requestedRoute = readRoute(message, controlMailbox);
if (requestedRoute == null)
continue;
if (route == null
|| !route.from().equals(requestedRoute.from())) {
subscriptions.subscribe(requestedRoute.from());
if (route != null)
subscriptions.unsubscribe(route.from());
}
route = requestedRoute;
IO.println(
name + " now mediates: "
+ route.from() + " -> " + route.to()
);
} else if (route != null
&& incoming.mailbox().equals(route.from())) {
FileMessageBoard.publish(route.to(), message);
}
}
}
}
DynamicMediatorOrchestrator.java
void main(String[] arguments) throws Exception {
if (arguments.length != 3
|| !arguments[0].matches("[A-Za-z0-9_-]+")) {
IO.println(
"Usage: java DynamicMediatorOrchestrator.java "
+ "<mediator-name> <from-mailbox> <to-mailbox>"
);
return;
}
var mediator = arguments[0];
var from = arguments[1];
var to = arguments[2];
FileMessageBoard.publish(
"mediators/" + mediator + "/inbox",
new Message("ROUTE", from + "\n" + to)
);
IO.println(
"Requested: " + mediator + " " + from + " -> " + to
);
}
A.6 — MQTT
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>at.ac.fhstp.sin</groupId>
<artifactId>calling-to-eventing</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.release>26</maven.compiler.release>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
<version>1.2.5</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.22.2</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-yaml</artifactId>
<version>2.22.2</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.14.1</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<version>3.8.1</version>
</plugin>
</plugins>
</build>
</project>
JsonTools.java
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
final class JsonTools {
static final ObjectMapper deserializationObjectMapper =
new ObjectMapper(new YAMLFactory())
.configure(
DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false
);
static final ObjectMapper serializationObjectMapper =
new ObjectMapper()
.configure(SerializationFeature.INDENT_OUTPUT, true);
private JsonTools() {}
}
MqttTools.java
import java.nio.charset.StandardCharsets;
final class MqttTools {
private MqttTools() {}
static boolean isValidTopic(String topic) {
return hasValidLength(topic)
&& !topic.contains("+")
&& !topic.contains("#");
}
static boolean isValidFilter(String filter) {
if (!hasValidLength(filter))
return false;
var levels = filter.split("/", -1);
for (var index = 0; index < levels.length; index++) {
var level = levels[index];
if (level.contains("#")
&& !(level.equals("#") && index == levels.length - 1))
return false;
if (level.contains("+") && !level.equals("+"))
return false;
}
return true;
}
static boolean isTopicMatchingFilter(String topic, String filter) {
if (!isValidTopic(topic) || !isValidFilter(filter))
return false;
if (topic.startsWith("$") && !filter.startsWith("$"))
return false;
var topicLevels = topic.split("/", -1);
var filterLevels = filter.split("/", -1);
var topicIndex = 0;
for (var filterLevel : filterLevels) {
if (filterLevel.equals("#"))
return true;
if (topicIndex == topicLevels.length)
return false;
if (!filterLevel.equals("+")
&& !filterLevel.equals(topicLevels[topicIndex]))
return false;
topicIndex++;
}
return topicIndex == topicLevels.length;
}
private static boolean hasValidLength(String value) {
if (value == null || value.isEmpty() || value.contains("\u0000"))
return false;
return value.getBytes(StandardCharsets.UTF_8).length <= 65_535;
}
}
MqttConsoleIn.java
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import org.eclipse.paho.mqttv5.client.IMqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttMessage;
record ConsoleInMessage(String value) {}
void publish(
IMqttAsyncClient client,
String topic,
String content,
int qos,
boolean retained)
throws Exception {
var message = new MqttMessage(
content.getBytes(StandardCharsets.UTF_8)
);
message.setQos(qos);
message.setRetained(retained);
client.publish(topic, message).waitForCompletion();
}
void main(String[] arguments) throws Exception {
var unit = arguments.length == 0 ? "alice" : arguments[0];
var broker = System.getProperty("mqtt.broker", "tcp://localhost:1883");
var topic = "Tutorial/ConsoleIn/U/" + unit + "/E/text";
var client = new MqttAsyncClient(
broker,
"console-in-" + unit + "-" + UUID.randomUUID(),
new MemoryPersistence()
);
client.connect().waitForCompletion();
try {
while (true) {
var text = IO.readln();
if (text == null)
return;
var json = JsonTools.serializationObjectMapper
.writeValueAsString(new ConsoleInMessage(text));
publish(client, topic, json, 1, false);
}
} finally {
client.disconnect().waitForCompletion();
client.close();
}
}
MqttRot13.java
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import org.eclipse.paho.mqttv5.client.IMqttAsyncClient;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
record Rot13Message(String value) {}
record Rot13Subscription(String topic) {}
record Rot13Intent(
Rot13Subscription subscribe,
Rot13Message text
) {}
MqttAsyncClient client;
String intentTopic;
String eventTopic;
String subscriptionStatusTopic;
String dynamicTextTopic;
void publish(
IMqttAsyncClient client,
String topic,
String content,
int qos,
boolean retained)
throws Exception {
var message = new MqttMessage(
content.getBytes(StandardCharsets.UTF_8)
);
message.setQos(qos);
message.setRetained(retained);
client.publish(topic, message).waitForCompletion();
}
MqttConnectionOptions connectionOptions(String lastWillTopic) {
var options = new MqttConnectionOptions();
options.setAutomaticReconnect(true);
var will = new MqttMessage(
"{\"value\":false}".getBytes(StandardCharsets.UTF_8),
1,
true,
new MqttProperties()
);
options.setWill(lastWillTopic, will);
return options;
}
String rot13(String text) {
var result = new StringBuilder();
for (var character : text.toCharArray()) {
if (character >= 'a' && character <= 'z')
character = (char) ('a' + (character - 'a' + 13) % 26);
else if (character >= 'A' && character <= 'Z')
character = (char) ('A' + (character - 'A' + 13) % 26);
result.append(character);
}
return result.toString();
}
void cryptAndPublish(String text) throws Exception {
var json = JsonTools.serializationObjectMapper
.writeValueAsString(new Rot13Message(rot13(text)));
publish(client, eventTopic, json, 1, false);
}
void handleIntent(MqttMessage message) throws Exception {
var intent = JsonTools.deserializationObjectMapper.readValue(
message.getPayload(),
Rot13Intent.class
);
if (intent.subscribe() != null) {
var requestedTopic = intent.subscribe().topic();
if (!MqttTools.isValidFilter(requestedTopic)) {
IO.println("Rejected invalid topic filter: " + requestedTopic);
return;
}
if (dynamicTextTopic != null)
client.unsubscribe(dynamicTextTopic).waitForCompletion();
dynamicTextTopic = requestedTopic;
client.subscribe(dynamicTextTopic, 1).waitForCompletion();
var status = JsonTools.serializationObjectMapper
.writeValueAsString(new Rot13Message(dynamicTextTopic));
publish(
client,
subscriptionStatusTopic,
status,
1,
true
);
IO.println("ROT13 now listens to " + dynamicTextTopic);
}
if (intent.text() != null)
cryptAndPublish(intent.text().value());
}
void handlePublication(String topic, MqttMessage message) throws Exception {
if (topic.equals(intentTopic)) {
handleIntent(message);
} else if (dynamicTextTopic != null
&& MqttTools.isTopicMatchingFilter(topic, dynamicTextTopic)) {
var input = JsonTools.deserializationObjectMapper.readValue(
message.getPayload(),
Rot13Message.class
);
cryptAndPublish(input.value());
}
}
void main(String[] arguments) throws Exception {
var unit = arguments.length == 0 ? "alice" : arguments[0];
var broker = System.getProperty("mqtt.broker", "tcp://localhost:1883");
var root = "Tutorial/ROT13/U/" + unit;
intentTopic = root + "/I";
eventTopic = root + "/E/crypted";
subscriptionStatusTopic = root + "/S/subscriptions/text";
var onlineTopic = root + "/S/online";
client = new MqttAsyncClient(
broker,
"rot13-" + unit + "-" + UUID.randomUUID(),
new MemoryPersistence()
);
client.setCallback(new MqttCallback() {
public void messageArrived(String topic, MqttMessage message)
throws Exception {
handlePublication(topic, message);
}
public void disconnected(MqttDisconnectResponse response) {}
public void mqttErrorOccurred(MqttException exception) {
exception.printStackTrace();
}
public void deliveryComplete(IMqttToken token) {}
public void connectComplete(boolean reconnect, String serverURI) {}
public void authPacketArrived(
int reasonCode,
MqttProperties properties) {}
});
client.connect(connectionOptions(onlineTopic)).waitForCompletion();
client.subscribe(intentTopic, 1).waitForCompletion();
publish(client, onlineTopic, "{\"value\":true}", 1, true);
IO.println("ROT13 waits for Intent on " + intentTopic);
Thread.currentThread().join();
}
MqttConsoleOut.java
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import org.eclipse.paho.mqttv5.client.IMqttAsyncClient;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
record ConsoleOutMessage(String value) {}
record ConsoleOutSubscription(String topic) {}
record ConsoleOutIntent(ConsoleOutSubscription subscribe) {}
MqttAsyncClient client;
String intentTopic;
String subscriptionStatusTopic;
String dynamicTextTopic;
void publish(
IMqttAsyncClient client,
String topic,
String content,
int qos,
boolean retained)
throws Exception {
var message = new MqttMessage(
content.getBytes(StandardCharsets.UTF_8)
);
message.setQos(qos);
message.setRetained(retained);
client.publish(topic, message).waitForCompletion();
}
MqttConnectionOptions connectionOptions(String lastWillTopic) {
var options = new MqttConnectionOptions();
options.setAutomaticReconnect(true);
var will = new MqttMessage(
"{\"value\":false}".getBytes(StandardCharsets.UTF_8),
1,
true,
new MqttProperties()
);
options.setWill(lastWillTopic, will);
return options;
}
void handleIntent(MqttMessage message) throws Exception {
var intent = JsonTools.deserializationObjectMapper.readValue(
message.getPayload(),
ConsoleOutIntent.class
);
if (intent.subscribe() == null)
return;
var requestedTopic = intent.subscribe().topic();
if (!MqttTools.isValidFilter(requestedTopic)) {
IO.println("Rejected invalid topic filter: " + requestedTopic);
return;
}
if (dynamicTextTopic != null)
client.unsubscribe(dynamicTextTopic).waitForCompletion();
dynamicTextTopic = requestedTopic;
client.subscribe(dynamicTextTopic, 1).waitForCompletion();
var status = JsonTools.serializationObjectMapper
.writeValueAsString(new ConsoleOutMessage(dynamicTextTopic));
publish(client, subscriptionStatusTopic, status, 1, true);
IO.println("ConsoleOut now listens to " + dynamicTextTopic);
}
void handlePublication(String topic, MqttMessage message) throws Exception {
if (topic.equals(intentTopic)) {
handleIntent(message);
} else if (dynamicTextTopic != null
&& MqttTools.isTopicMatchingFilter(topic, dynamicTextTopic)) {
var text = JsonTools.deserializationObjectMapper.readValue(
message.getPayload(),
ConsoleOutMessage.class
);
IO.println(text.value());
}
}
void main(String[] arguments) throws Exception {
var unit = arguments.length == 0 ? "alice" : arguments[0];
var broker = System.getProperty("mqtt.broker", "tcp://localhost:1883");
var root = "Tutorial/ConsoleOut/U/" + unit;
intentTopic = root + "/I";
subscriptionStatusTopic = root + "/S/subscriptions/text";
var onlineTopic = root + "/S/online";
client = new MqttAsyncClient(
broker,
"console-out-" + unit + "-" + UUID.randomUUID(),
new MemoryPersistence()
);
client.setCallback(new MqttCallback() {
public void messageArrived(String topic, MqttMessage message)
throws Exception {
handlePublication(topic, message);
}
public void disconnected(MqttDisconnectResponse response) {}
public void mqttErrorOccurred(MqttException exception) {
exception.printStackTrace();
}
public void deliveryComplete(IMqttToken token) {}
public void connectComplete(boolean reconnect, String serverURI) {}
public void authPacketArrived(
int reasonCode,
MqttProperties properties) {}
});
client.connect(connectionOptions(onlineTopic)).waitForCompletion();
client.subscribe(intentTopic, 1).waitForCompletion();
publish(client, onlineTopic, "{\"value\":true}", 1, true);
IO.println("ConsoleOut waits for Intent on " + intentTopic);
Thread.currentThread().join();
}
MqttConfigure.java
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import org.eclipse.paho.mqttv5.client.IMqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttMessage;
void publish(
IMqttAsyncClient client,
String topic,
String content,
int qos,
boolean retained)
throws Exception {
var message = new MqttMessage(
content.getBytes(StandardCharsets.UTF_8)
);
message.setQos(qos);
message.setRetained(retained);
client.publish(topic, message).waitForCompletion();
}
void main(String[] arguments) throws Exception {
var consoleInUnit = arguments.length > 0 ? arguments[0] : "alice";
var rot13Unit = arguments.length > 1 ? arguments[1] : "alice";
var consoleOutUnit = arguments.length > 2 ? arguments[2] : "alice";
var broker = System.getProperty("mqtt.broker", "tcp://localhost:1883");
var client = new MqttAsyncClient(
broker,
"configure-" + UUID.randomUUID(),
new MemoryPersistence()
);
client.connect().waitForCompletion();
var rot13Intent = """
subscribe:
topic: Tutorial/ConsoleIn/U/%s/E/text
""".formatted(consoleInUnit);
var consoleOutIntent = """
subscribe:
topic: Tutorial/ROT13/U/%s/E/crypted
""".formatted(rot13Unit);
publish(
client,
"Tutorial/ROT13/U/" + rot13Unit + "/I",
rot13Intent,
1,
false
);
publish(
client,
"Tutorial/ConsoleOut/U/" + consoleOutUnit + "/I",
consoleOutIntent,
1,
false
);
client.disconnect().waitForCompletion();
client.close();
IO.println("Wired: ConsoleIn -> ROT13 -> ConsoleOut");
}