6251 lines
120 KiB
Markdown
6251 lines
120 KiB
Markdown
# 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](#appendix-a--complete-java-sources). 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`](supplemental-java/01-procedural/ProceduralRot13.java) |
|
|
| mediated queues in one JVM | [`02-static-in-jvm/StaticQueueRot13.java`](supplemental-java/02-static-in-jvm/StaticQueueRot13.java) |
|
|
| mediated filesystem | [`03-static-filesystem/`](supplemental-java/03-static-filesystem/) |
|
|
| dynamic queues in one JVM | [`04-dynamic-in-jvm/DynamicQueueRot13.java`](supplemental-java/04-dynamic-in-jvm/DynamicQueueRot13.java) |
|
|
| dynamic filesystem | [`05-dynamic-filesystem/`](supplemental-java/05-dynamic-filesystem/) |
|
|
| MQTT | [`06-mqtt/`](supplemental-java/06-mqtt/) |
|
|
|
|
Install or bookmark the tools before the corresponding exercise:
|
|
|
|
- [Java 26](https://jdk.java.net/26/);
|
|
- [Syncthing downloads](https://syncthing.net/downloads/);
|
|
- [Eclipse Mosquitto](https://mosquitto.org/download/), the MQTT broker used in the exercises;
|
|
- [MQTT Explorer](https://mqtt-explorer.com/), a graphical MQTT client for inspecting, publishing and subscribing to messages;
|
|
- [Eclipse Paho Java client](https://github.com/eclipse-paho/paho.mqtt.java), used by the Java MQTT examples.
|
|
|
|
The examples deliberately use Java 26 compact source files whenever possible:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
keyboard → ROT13 → console
|
|
```
|
|
|
|
You type:
|
|
|
|
```text
|
|
Hello World
|
|
```
|
|
|
|
and receive:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ConsoleIn
|
|
│
|
|
│ call
|
|
▼
|
|
ROT13
|
|
│
|
|
│ call
|
|
▼
|
|
ConsoleOut
|
|
```
|
|
|
|
The important part is tiny:
|
|
|
|
```java
|
|
String text = IO.readln();
|
|
|
|
String transformed = rot13(text);
|
|
|
|
IO.println(transformed);
|
|
```
|
|
|
|
The complete compact Java 26 program is
|
|
[`ProceduralRot13.java`](supplemental-java/01-procedural/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:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
rot13(...)
|
|
```
|
|
|
|
ROT13 should ideally not contain:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
BlockingQueue<Message>
|
|
```
|
|
|
|
We can maintain several of them:
|
|
|
|
```java
|
|
Map<String, BlockingQueue<Message>> mailboxes =
|
|
new ConcurrentHashMap<>();
|
|
```
|
|
|
|
with a convenience method:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
rot13(text);
|
|
```
|
|
|
|
ConsoleIn publishes into its own mailbox:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
Message input =
|
|
mailbox("rot13.inbox").take();
|
|
```
|
|
|
|
It performs its capability:
|
|
|
|
```java
|
|
Message output = new Message(
|
|
"TEXT",
|
|
rot13(input.value())
|
|
);
|
|
```
|
|
|
|
and writes its own output:
|
|
|
|
```java
|
|
mailbox("rot13.outbox").put(output);
|
|
```
|
|
|
|
Its world is now:
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
while (true) {
|
|
Message message =
|
|
mailbox("consoleOut.inbox").take();
|
|
|
|
IO.println(message.value());
|
|
}
|
|
```
|
|
|
|
ConsoleOut knows only:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ConsoleIn → ROT13 → ConsoleOut
|
|
```
|
|
|
|
So introduce the smallest possible mediator:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
Thread.ofVirtual().start(
|
|
() -> mediate(
|
|
"consoleIn.outbox",
|
|
"rot13.inbox"
|
|
)
|
|
);
|
|
```
|
|
|
|
and another:
|
|
|
|
```java
|
|
Thread.ofVirtual().start(
|
|
() -> mediate(
|
|
"rot13.outbox",
|
|
"consoleOut.inbox"
|
|
)
|
|
);
|
|
```
|
|
|
|
The system becomes:
|
|
|
|
```text
|
|
ConsoleIn
|
|
│
|
|
▼
|
|
consoleIn.outbox
|
|
│
|
|
▼
|
|
Mediator
|
|
│
|
|
▼
|
|
rot13.inbox
|
|
│
|
|
▼
|
|
ROT13
|
|
│
|
|
▼
|
|
rot13.outbox
|
|
│
|
|
▼
|
|
Mediator
|
|
│
|
|
▼
|
|
consoleOut.inbox
|
|
│
|
|
▼
|
|
ConsoleOut
|
|
```
|
|
|
|
Run it.
|
|
|
|
Functionally:
|
|
|
|
```text
|
|
Hello World
|
|
→
|
|
Uryyb Jbeyq
|
|
```
|
|
|
|
Again.
|
|
|
|
Architecturally, however, something important has happened.
|
|
|
|
The complete runnable version is
|
|
[`StaticQueueRot13.java`](supplemental-java/02-static-in-jvm/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?
|
|
|
|
```text
|
|
No.
|
|
```
|
|
|
|
Does ROT13 know ConsoleIn?
|
|
|
|
```text
|
|
No.
|
|
```
|
|
|
|
Does ROT13 know ConsoleOut?
|
|
|
|
```text
|
|
No.
|
|
```
|
|
|
|
Does ConsoleOut know ROT13?
|
|
|
|
```text
|
|
No.
|
|
```
|
|
|
|
Has the topology disappeared?
|
|
|
|
```text
|
|
No.
|
|
```
|
|
|
|
Who knows it?
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ROT13 transforms text
|
|
```
|
|
|
|
belongs to ROT13.
|
|
|
|
The topology:
|
|
|
|
```text
|
|
ConsoleIn → ROT13 → ConsoleOut
|
|
```
|
|
|
|
belongs to system composition.
|
|
|
|
---
|
|
|
|
# 9. Why Virtual Threads?
|
|
|
|
Several components now wait indefinitely:
|
|
|
|
```text
|
|
wait for keyboard input
|
|
|
|
wait for mailbox input
|
|
|
|
wait for another mailbox
|
|
```
|
|
|
|
Virtual threads make it convenient to run each such loop independently:
|
|
|
|
```java
|
|
Thread.ofVirtual().start(service::run);
|
|
```
|
|
|
|
This is merely a Java implementation convenience.
|
|
|
|
Do not confuse:
|
|
|
|
```text
|
|
virtual threads
|
|
```
|
|
|
|
with:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ConsoleIn
|
|
↓
|
|
Mediator
|
|
↓
|
|
ROT13
|
|
```
|
|
|
|
and think:
|
|
|
|
> Why doesn't ROT13 simply read ConsoleIn's outbox?
|
|
|
|
Because ROT13 is still egoistic.
|
|
|
|
ROT13 knows:
|
|
|
|
```text
|
|
I transform text.
|
|
```
|
|
|
|
It does not know:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
Process 1 Process 2 Process 3
|
|
|
|
ConsoleIn ROT13 ConsoleOut
|
|
```
|
|
|
|
They can no longer share:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
messageBoard/
|
|
```
|
|
|
|
and inside it:
|
|
|
|
```text
|
|
messageBoard/
|
|
├── consoleIn/
|
|
│ └── outbox/
|
|
│
|
|
├── rot13/
|
|
│ ├── inbox/
|
|
│ └── outbox/
|
|
│
|
|
└── consoleOut/
|
|
└── inbox/
|
|
```
|
|
|
|
For this version:
|
|
|
|
```text
|
|
directory = mailbox
|
|
|
|
file = message
|
|
```
|
|
|
|
Publishing means:
|
|
|
|
```text
|
|
create message file
|
|
```
|
|
|
|
Receiving means:
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
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
|
|
TEXT
|
|
Hello World
|
|
TIMESTAMP: 2026-09-22T10:42:12.123Z
|
|
```
|
|
|
|
A message written manually may omit it:
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
String serialize(Message message) {
|
|
var text = message.type()
|
|
+ "\n"
|
|
+ message.value();
|
|
|
|
if (message.timestamp() != null)
|
|
text += "\nTIMESTAMP: " + message.timestamp();
|
|
|
|
return text;
|
|
}
|
|
```
|
|
|
|
And reading:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
Files.writeString(
|
|
Path.of("message.msg"),
|
|
hugeMessage
|
|
);
|
|
```
|
|
|
|
Another process may discover:
|
|
|
|
```text
|
|
message.msg
|
|
```
|
|
|
|
before the writer has finished.
|
|
|
|
So define a publication protocol:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
WatchService watcher =
|
|
FileSystems
|
|
.getDefault()
|
|
.newWatchService();
|
|
```
|
|
|
|
Register:
|
|
|
|
```java
|
|
directory.register(
|
|
watcher,
|
|
StandardWatchEventKinds.ENTRY_CREATE
|
|
);
|
|
```
|
|
|
|
Wait:
|
|
|
|
```java
|
|
WatchKey key = watcher.take();
|
|
```
|
|
|
|
Then inspect the newly created `.msg` files.
|
|
|
|
The observer must ignore every other filename, including `.tmp`:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
producer
|
|
│
|
|
│ publish
|
|
▼
|
|
filesystem
|
|
│
|
|
│ observation
|
|
▼
|
|
consumer
|
|
```
|
|
|
|
The mechanics will later change.
|
|
|
|
The idea will not.
|
|
|
|
The complete shared implementation is
|
|
[`Message.java`](supplemental-java/03-static-filesystem/Message.java) and
|
|
[`FileMessageBoard.java`](supplemental-java/03-static-filesystem/FileMessageBoard.java).
|
|
|
|
---
|
|
|
|
# 16. The Three Services Become Separate Programs
|
|
|
|
We now have:
|
|
|
|
```text
|
|
SensorService.java
|
|
|
|
ProcessorService.java
|
|
|
|
ActuatorService.java
|
|
```
|
|
|
|
The sensor writes only:
|
|
|
|
```text
|
|
consoleIn/outbox
|
|
```
|
|
|
|
ROT13 reads only:
|
|
|
|
```text
|
|
rot13/inbox
|
|
```
|
|
|
|
and writes:
|
|
|
|
```text
|
|
rot13/outbox
|
|
```
|
|
|
|
ConsoleOut reads only:
|
|
|
|
```text
|
|
consoleOut/inbox
|
|
```
|
|
|
|
They are now separate programs.
|
|
|
|
Complete sources:
|
|
|
|
- [`SensorService.java`](supplemental-java/03-static-filesystem/SensorService.java)
|
|
- [`ProcessorService.java`](supplemental-java/03-static-filesystem/ProcessorService.java)
|
|
- [`ActuatorService.java`](supplemental-java/03-static-filesystem/ActuatorService.java)
|
|
|
|
But once again:
|
|
|
|
```text
|
|
nobody connects them
|
|
```
|
|
|
|
Good.
|
|
|
|
We already know how to solve that.
|
|
|
|
---
|
|
|
|
# 17. The Mediator Becomes a Process
|
|
|
|
Create:
|
|
|
|
```text
|
|
Mediator.java
|
|
```
|
|
|
|
Its job is still conceptually:
|
|
|
|
```java
|
|
mediate(from, to)
|
|
```
|
|
|
|
But now mediation means:
|
|
|
|
```text
|
|
watch source directory
|
|
|
|
for each new message:
|
|
|
|
publish a copy
|
|
into destination directory
|
|
```
|
|
|
|
Run:
|
|
|
|
```text
|
|
Mediator
|
|
consoleIn/outbox
|
|
→
|
|
rot13/inbox
|
|
```
|
|
|
|
and:
|
|
|
|
```text
|
|
Mediator
|
|
rot13/outbox
|
|
→
|
|
consoleOut/inbox
|
|
```
|
|
|
|
Our architecture is unchanged:
|
|
|
|
```text
|
|
ConsoleIn
|
|
│
|
|
▼
|
|
Mediator
|
|
│
|
|
▼
|
|
ROT13
|
|
│
|
|
▼
|
|
Mediator
|
|
│
|
|
▼
|
|
ConsoleOut
|
|
```
|
|
|
|
Only the implementation boundary changed.
|
|
|
|
The complete process is
|
|
[`Mediator.java`](supplemental-java/03-static-filesystem/Mediator.java).
|
|
|
|
Notice something slightly asymmetric. The three capability services receive
|
|
their work through mailboxes, but the mediator receives its route through
|
|
startup arguments:
|
|
|
|
```text
|
|
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:
|
|
|
|
```bash
|
|
rm -rf messageBoard
|
|
```
|
|
|
|
Open terminals.
|
|
|
|
### Terminal 1
|
|
|
|
```bash
|
|
java SensorService.java
|
|
```
|
|
|
|
### Terminal 2
|
|
|
|
```bash
|
|
java Mediator.java \
|
|
consoleIn/outbox \
|
|
rot13/inbox
|
|
```
|
|
|
|
### Terminal 3
|
|
|
|
```bash
|
|
java ProcessorService.java
|
|
```
|
|
|
|
### Terminal 4
|
|
|
|
```bash
|
|
java Mediator.java \
|
|
rot13/outbox \
|
|
consoleOut/inbox
|
|
```
|
|
|
|
### Terminal 5
|
|
|
|
```bash
|
|
java ActuatorService.java
|
|
```
|
|
|
|
Count them.
|
|
|
|
```text
|
|
1 sensor
|
|
1 mediator
|
|
1 processor
|
|
1 mediator
|
|
1 actuator
|
|
|
|
= 5 processes
|
|
```
|
|
|
|
All to perform:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
> **[Syncthing](https://syncthing.net/downloads/)**
|
|
|
|
---
|
|
|
|
# 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:
|
|
|
|
```text
|
|
Intent
|
|
|
|
Event
|
|
|
|
Status
|
|
|
|
ROT13
|
|
|
|
mediators
|
|
```
|
|
|
|
It merely gives us local directories whose contents are asynchronously replicated between participating machines.
|
|
|
|
Conceptually:
|
|
|
|
```text
|
|
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:
|
|
|
|
1. install and start Syncthing;
|
|
2. connect the participating devices;
|
|
3. share one folder between them;
|
|
4. choose a local directory for that folder;
|
|
5. use this directory as the root of our message board.
|
|
|
|
For example:
|
|
|
|
```text
|
|
Computer A:
|
|
|
|
/home/alice/course-message-board
|
|
```
|
|
|
|
```text
|
|
Computer B:
|
|
|
|
/home/bob/sin/course-message-board
|
|
```
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
Path MESSAGE_BOARD_ROOT =
|
|
Path.of(
|
|
System.getProperty(
|
|
"messageBoard.root",
|
|
"messageBoard"
|
|
)
|
|
);
|
|
```
|
|
|
|
Then:
|
|
|
|
```bash
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
.tmp
|
|
↓
|
|
atomic rename
|
|
↓
|
|
.msg
|
|
```
|
|
|
|
That gives a useful local publication boundary.
|
|
|
|
But Syncthing performs **asynchronous replication**.
|
|
|
|
Do not infer:
|
|
|
|
```text
|
|
atomic locally
|
|
=
|
|
atomic globally
|
|
```
|
|
|
|
Those are different guarantees.
|
|
|
|
Our consumers still ignore unfinished protocol files such as:
|
|
|
|
```text
|
|
*.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:
|
|
|
|
```text
|
|
Computer A
|
|
──────────
|
|
SensorService
|
|
|
|
Computer B
|
|
──────────
|
|
Mediator 1
|
|
ProcessorService
|
|
|
|
Computer C
|
|
──────────
|
|
Mediator 2
|
|
ActuatorService
|
|
```
|
|
|
|
Or, if you have enough machines:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
Hello
|
|
```
|
|
|
|
we get approximately:
|
|
|
|
```text
|
|
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 → TEXT
|
|
```
|
|
|
|
We also operate:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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
|
|
TEXT
|
|
...
|
|
Hello
|
|
```
|
|
|
|
and publishes:
|
|
|
|
```text
|
|
TEXT
|
|
...
|
|
Hello
|
|
```
|
|
|
|
Mediator 2 receives:
|
|
|
|
```text
|
|
TEXT
|
|
...
|
|
Uryyb
|
|
```
|
|
|
|
and publishes:
|
|
|
|
```text
|
|
TEXT
|
|
...
|
|
Uryyb
|
|
```
|
|
|
|
Neither adds information.
|
|
|
|
Neither changes semantics.
|
|
|
|
Neither adapts an incompatible format.
|
|
|
|
They perform approximately:
|
|
|
|
```text
|
|
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:
|
|
|
|
```bash
|
|
java Mediator.java \
|
|
consoleIn/outbox \
|
|
rot13/inbox
|
|
```
|
|
|
|
Its capability is already generic:
|
|
|
|
```text
|
|
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:
|
|
|
|
```bash
|
|
java ConfigurableMediator.java mediator1
|
|
```
|
|
|
|
From that identity it derives one permanent control mailbox:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ROUTE
|
|
consoleIn/outbox
|
|
rot13/inbox
|
|
```
|
|
|
|
The first line is the message type. The next two lines form its value:
|
|
|
|
```text
|
|
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:
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-configurable SensorService.java
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-configurable \
|
|
ConfigurableMediator.java mediator1
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-configurable ProcessorService.java
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-configurable \
|
|
ConfigurableMediator.java mediator2
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-configurable ActuatorService.java
|
|
```
|
|
|
|
Both mediators initially know no route. They wait on:
|
|
|
|
```text
|
|
mediators/mediator1/inbox
|
|
|
|
mediators/mediator2/inbox
|
|
```
|
|
|
|
After all five programs are waiting, run the short-lived orchestrator in a
|
|
sixth terminal:
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-configurable \
|
|
MediatorOrchestrator.java
|
|
```
|
|
|
|
It publishes:
|
|
|
|
```text
|
|
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:
|
|
|
|
- [`ConfigurableMediator.java`](supplemental-java/03-static-filesystem/ConfigurableMediator.java)
|
|
- [`MediatorOrchestrator.java`](supplemental-java/03-static-filesystem/MediatorOrchestrator.java)
|
|
|
|
## What Improved—and What Did Not?
|
|
|
|
The named mediator now behaves like a normal configurable service:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ConsoleIn
|
|
|
|
ROT13
|
|
|
|
ConsoleOut
|
|
```
|
|
|
|
inside one JVM.
|
|
|
|
Previously:
|
|
|
|
```text
|
|
ConsoleIn
|
|
↓
|
|
Mediator
|
|
↓
|
|
ROT13
|
|
```
|
|
|
|
What if ROT13 could listen directly to:
|
|
|
|
```text
|
|
consoleIn.outbox
|
|
```
|
|
|
|
?
|
|
|
|
Then:
|
|
|
|
```text
|
|
ConsoleIn
|
|
│
|
|
▼
|
|
consoleIn.outbox
|
|
│
|
|
▼
|
|
ROT13
|
|
```
|
|
|
|
No mediator.
|
|
|
|
Wonderful.
|
|
|
|
So we might write inside ROT13:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
I can transform text.
|
|
```
|
|
|
|
It may also know:
|
|
|
|
```text
|
|
I am capable of listening
|
|
for compatible text messages.
|
|
```
|
|
|
|
It should not necessarily know:
|
|
|
|
```text
|
|
my input always comes from
|
|
consoleIn.outbox
|
|
```
|
|
|
|
That is not part of the ROT13 capability.
|
|
|
|
That is deployment topology.
|
|
|
|
So instead of compiling:
|
|
|
|
```java
|
|
listen("consoleIn.outbox");
|
|
```
|
|
|
|
into ROT13, we send:
|
|
|
|
```text
|
|
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`:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
Map<String, Thread> subscriptions =
|
|
new HashMap<>();
|
|
```
|
|
|
|
and provide:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
subscribe(mailboxName);
|
|
```
|
|
|
|
where:
|
|
|
|
```text
|
|
mailboxName
|
|
```
|
|
|
|
arrived from outside the service.
|
|
|
|
Now configure:
|
|
|
|
```java
|
|
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`](supplemental-java/04-dynamic-in-jvm/DynamicQueueRot13.java).
|
|
|
|
---
|
|
|
|
# 31. Do the Same for ConsoleOut
|
|
|
|
Send:
|
|
|
|
```java
|
|
mailbox("consoleOut.inbox").put(
|
|
new Message(
|
|
"TOPIC",
|
|
"rot13.outbox"
|
|
)
|
|
);
|
|
```
|
|
|
|
ConsoleOut starts listening directly to ROT13's output.
|
|
|
|
Now:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ROUTE
|
|
consoleIn/outbox
|
|
rot13/inbox
|
|
```
|
|
|
|
That meant:
|
|
|
|
> **Please forward messages from here to there.**
|
|
|
|
Now ROT13 receives:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
"mediator1, forward from here to there."
|
|
```
|
|
|
|
But the mediator then continuously transported application data:
|
|
|
|
```text
|
|
event
|
|
↓
|
|
mediator
|
|
↓
|
|
consumer
|
|
```
|
|
|
|
Now the orchestrator sends configuration to the consumers themselves:
|
|
|
|
```text
|
|
"ROT13, listen there."
|
|
|
|
"ConsoleOut, listen there."
|
|
```
|
|
|
|
Afterwards:
|
|
|
|
```text
|
|
ConsoleIn ─────► ROT13 ─────► ConsoleOut
|
|
```
|
|
|
|
runs independently.
|
|
|
|
This is a useful distinction:
|
|
|
|
```text
|
|
control plane
|
|
```
|
|
|
|
versus:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
STATIC MEDIATED
|
|
|
|
Producer
|
|
│
|
|
▼
|
|
Mediator
|
|
│
|
|
▼
|
|
Consumer
|
|
```
|
|
|
|
with:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
rot13/inbox
|
|
```
|
|
|
|
A configuration message arrives:
|
|
|
|
```text
|
|
TOPIC consoleIn/outbox
|
|
```
|
|
|
|
ROT13 then begins watching:
|
|
|
|
```text
|
|
consoleIn/outbox
|
|
```
|
|
|
|
directly.
|
|
|
|
---
|
|
|
|
# 36. One `WatchService` Is No Longer Enough
|
|
|
|
Our first filesystem helper could do:
|
|
|
|
```java
|
|
watch(topic, receiver);
|
|
```
|
|
|
|
and block forever.
|
|
|
|
That was sufficient when one process always watched one fixed mailbox.
|
|
|
|
Now ROT13 needs to:
|
|
|
|
```text
|
|
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:
|
|
|
|
```java
|
|
subscriptions.subscribe(topic);
|
|
|
|
subscriptions.unsubscribe(topic);
|
|
|
|
Message message =
|
|
subscriptions.take();
|
|
```
|
|
|
|
Internally it manages several:
|
|
|
|
```java
|
|
WatchKey
|
|
```
|
|
|
|
registrations.
|
|
|
|
ROT13 can therefore do:
|
|
|
|
```java
|
|
subscriptions.subscribe(
|
|
"rot13/inbox"
|
|
);
|
|
```
|
|
|
|
permanently.
|
|
|
|
Then when it receives:
|
|
|
|
```text
|
|
TOPIC consoleIn/outbox
|
|
```
|
|
|
|
it executes:
|
|
|
|
```java
|
|
subscriptions.subscribe(
|
|
"consoleIn/outbox"
|
|
);
|
|
```
|
|
|
|
The service implementation still does not contain:
|
|
|
|
```text
|
|
consoleIn/outbox
|
|
```
|
|
|
|
as a compiled topology decision.
|
|
|
|
The complete helper is
|
|
[`FileSubscriptions.java`](supplemental-java/05-dynamic-filesystem/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:
|
|
|
|
```text
|
|
SensorService
|
|
|
|
DynamicProcessorService
|
|
|
|
DynamicActuatorService
|
|
```
|
|
|
|
Do **not** start the forwarding mediators.
|
|
|
|
Then send configuration:
|
|
|
|
```text
|
|
ROT13:
|
|
TOPIC consoleIn/outbox
|
|
```
|
|
|
|
and:
|
|
|
|
```text
|
|
ConsoleOut:
|
|
TOPIC rot13/outbox
|
|
```
|
|
|
|
Type:
|
|
|
|
```text
|
|
Hello
|
|
```
|
|
|
|
The path is now:
|
|
|
|
```text
|
|
consoleIn/outbox
|
|
│
|
|
│ observed directly
|
|
▼
|
|
ROT13
|
|
│
|
|
▼
|
|
rot13/outbox
|
|
│
|
|
│ observed directly
|
|
▼
|
|
ConsoleOut
|
|
```
|
|
|
|
Count the application publications.
|
|
|
|
Previously:
|
|
|
|
```text
|
|
input
|
|
mediator copy
|
|
ROT13 result
|
|
mediator copy
|
|
```
|
|
|
|
Now:
|
|
|
|
```text
|
|
input
|
|
ROT13 result
|
|
```
|
|
|
|
The wiring Intent is sent only when topology changes.
|
|
|
|
Now the architectural advantage is visible.
|
|
|
|
Complete sources:
|
|
|
|
- [`SensorService.java`](supplemental-java/05-dynamic-filesystem/SensorService.java)
|
|
- [`DynamicProcessorService.java`](supplemental-java/05-dynamic-filesystem/DynamicProcessorService.java)
|
|
- [`DynamicActuatorService.java`](supplemental-java/05-dynamic-filesystem/DynamicActuatorService.java)
|
|
- [`Configure.java`](supplemental-java/05-dynamic-filesystem/Configure.java)
|
|
|
|
`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:
|
|
|
|
```text
|
|
Computer A
|
|
──────────
|
|
Sensor
|
|
|
|
Computer B
|
|
──────────
|
|
ROT13
|
|
|
|
Computer C
|
|
──────────
|
|
ConsoleOut
|
|
```
|
|
|
|
No forwarding mediator processes.
|
|
|
|
Send configuration:
|
|
|
|
```text
|
|
ROT13:
|
|
listen to
|
|
consoleIn/outbox
|
|
```
|
|
|
|
and:
|
|
|
|
```text
|
|
ConsoleOut:
|
|
listen to
|
|
rot13/outbox
|
|
```
|
|
|
|
through their own control mailboxes.
|
|
|
|
Then type:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
Hello
|
|
```
|
|
|
|
we now publish:
|
|
|
|
```text
|
|
1. input Event
|
|
|
|
2. ROT13 output Event
|
|
```
|
|
|
|
instead of:
|
|
|
|
```text
|
|
1. input
|
|
|
|
2. mediator copy
|
|
|
|
3. ROT13 output
|
|
|
|
4. mediator copy
|
|
```
|
|
|
|
We also removed two continuously running processes.
|
|
|
|
More importantly:
|
|
|
|
```text
|
|
ROT13 capability
|
|
```
|
|
|
|
still does not contain:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
I = Intent
|
|
|
|
S = Status / State
|
|
|
|
E = Event
|
|
```
|
|
|
|
---
|
|
|
|
# 43. `E` — Event
|
|
|
|
An Event describes:
|
|
|
|
> **something that happened**
|
|
|
|
Examples:
|
|
|
|
```text
|
|
textEntered
|
|
|
|
buttonPressed
|
|
|
|
temperatureChanged
|
|
|
|
transformationCompleted
|
|
|
|
vehiclePassedBarcode
|
|
```
|
|
|
|
Past tense is often a useful naming hint.
|
|
|
|
Our input publication becomes conceptually:
|
|
|
|
```text
|
|
ConsoleIn/E/text
|
|
```
|
|
|
|
ROT13 output becomes:
|
|
|
|
```text
|
|
ROT13/E/crypted
|
|
```
|
|
|
|
---
|
|
|
|
# 44. `S` — Status / State
|
|
|
|
Status describes:
|
|
|
|
> **something currently true**
|
|
|
|
Examples:
|
|
|
|
```text
|
|
online = true
|
|
|
|
mode = automatic
|
|
|
|
currentSpeed = 250
|
|
|
|
subscription =
|
|
ConsoleIn/E/text
|
|
```
|
|
|
|
If someone asks:
|
|
|
|
> Why is ROT13 reacting to ConsoleIn?
|
|
|
|
the service can publish:
|
|
|
|
```text
|
|
ROT13/S/subscriptions
|
|
```
|
|
|
|
containing its current wiring.
|
|
|
|
State explains current behaviour.
|
|
|
|
---
|
|
|
|
# 45. `I` — Intent
|
|
|
|
Intent means:
|
|
|
|
> **what should happen or become true**
|
|
|
|
Examples:
|
|
|
|
```text
|
|
transform this text
|
|
|
|
subscribe to this Event
|
|
|
|
unsubscribe from this Event
|
|
|
|
change speed
|
|
|
|
switch lights on
|
|
```
|
|
|
|
Our primitive:
|
|
|
|
```text
|
|
TOPIC consoleIn/outbox
|
|
```
|
|
|
|
was really an early Intent.
|
|
|
|
A more explicit structure could eventually become:
|
|
|
|
```text
|
|
ROT13/I
|
|
```
|
|
|
|
with:
|
|
|
|
```text
|
|
SUBSCRIBE ConsoleIn/E/text
|
|
```
|
|
|
|
---
|
|
|
|
# 46. Event-Driven Does Not Mean "No Commands"
|
|
|
|
Suppose somebody sends:
|
|
|
|
```text
|
|
ROT13/I
|
|
|
|
ACTION Hello
|
|
```
|
|
|
|
That is command-like.
|
|
|
|
Fine.
|
|
|
|
ROT13 performs its capability and publishes:
|
|
|
|
```text
|
|
ROT13/E/crypted
|
|
|
|
Uryyb
|
|
```
|
|
|
|
Likewise:
|
|
|
|
```text
|
|
SUBSCRIBE ConsoleIn/E/text
|
|
```
|
|
|
|
is Intent.
|
|
|
|
And:
|
|
|
|
```text
|
|
ROT13/S/subscriptions
|
|
```
|
|
|
|
describes State.
|
|
|
|
A useful flow is:
|
|
|
|
```text
|
|
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.
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
ConsoleIn/E/...
|
|
```
|
|
|
|
is written by ConsoleIn.
|
|
|
|
```text
|
|
ROT13/E/...
|
|
```
|
|
|
|
is written by ROT13.
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
A → A
|
|
```
|
|
|
|
may be unnecessary.
|
|
|
|
Now consider another case.
|
|
|
|
Producer publishes:
|
|
|
|
```json
|
|
{
|
|
"value": "Hello"
|
|
}
|
|
```
|
|
|
|
Consumer expects:
|
|
|
|
```json
|
|
{
|
|
"display": {
|
|
"text": "Hello",
|
|
"durationMs": 5000
|
|
}
|
|
}
|
|
```
|
|
|
|
Can the consumer simply subscribe to the producer?
|
|
|
|
Transport-wise:
|
|
|
|
```text
|
|
yes
|
|
```
|
|
|
|
Semantically:
|
|
|
|
```text
|
|
no
|
|
```
|
|
|
|
The data contracts are incompatible.
|
|
|
|
Now mediation has a real job.
|
|
|
|
---
|
|
|
|
# 49. A Transformer Has Earned Its Existence
|
|
|
|
Introduce:
|
|
|
|
```text
|
|
TextToDisplayTransformer
|
|
```
|
|
|
|
Its architecture is:
|
|
|
|
```text
|
|
Producer
|
|
│
|
|
│ {"value":"Hello"}
|
|
▼
|
|
Transformer
|
|
│
|
|
│ {"display":{...}}
|
|
▼
|
|
Consumer
|
|
```
|
|
|
|
This mediator is not an expensive identity function.
|
|
|
|
It contributes:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
Transformer/I
|
|
|
|
Transformer/S/...
|
|
|
|
Transformer/E/...
|
|
```
|
|
|
|
It can itself receive:
|
|
|
|
```text
|
|
SUBSCRIBE SomeProducer/E/text
|
|
```
|
|
|
|
through Intent.
|
|
|
|
It transforms compatible input into its own Event contract.
|
|
|
|
Then the final consumer may dynamically subscribe to:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
Intent
|
|
│
|
|
▼
|
|
┌────────────┐
|
|
Events ────────►│ Service │──────► Status
|
|
│ │
|
|
└─────┬──────┘
|
|
│
|
|
▼
|
|
Events
|
|
```
|
|
|
|
A service:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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](https://mosquitto.org/download/). 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:
|
|
|
|
```bash
|
|
mosquitto -v
|
|
```
|
|
|
|
The default local address used below is:
|
|
|
|
```text
|
|
tcp://localhost:1883
|
|
```
|
|
|
|
Then use [MQTT Explorer](https://mqtt-explorer.com/), a graphical MQTT client
|
|
that lets you inspect the topic hierarchy and publish messages by hand.
|
|
|
|
Connect to the broker.
|
|
|
|
Publish:
|
|
|
|
```text
|
|
Tutorial/ConsoleIn/U/alice/E/text
|
|
```
|
|
|
|
with:
|
|
|
|
```json
|
|
{"value":"Hello"}
|
|
```
|
|
|
|
Observe it.
|
|
|
|
Try another client.
|
|
|
|
Subscribe to a topic hierarchy.
|
|
|
|
See what the broker does.
|
|
|
|
Learn:
|
|
|
|
```text
|
|
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](https://github.com/eclipse-paho/paho.mqtt.java).
|
|
|
|
For Maven:
|
|
|
|
```xml
|
|
<dependency>
|
|
<groupId>org.eclipse.paho</groupId>
|
|
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
|
|
<version>1.2.5</version>
|
|
</dependency>
|
|
```
|
|
|
|
Initially we need only:
|
|
|
|
```text
|
|
connect
|
|
|
|
publish
|
|
|
|
subscribe
|
|
|
|
callback
|
|
```
|
|
|
|
The complete Maven configuration—including Paho and the Jackson dependencies
|
|
introduced later—is
|
|
[`pom.xml`](supplemental-java/06-mqtt/pom.xml).
|
|
|
|
---
|
|
|
|
# 57. Connect
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
publish(
|
|
client,
|
|
"Tutorial/ConsoleIn/U/alice/E/text",
|
|
"{\"value\":\"Hello\"}",
|
|
1,
|
|
false
|
|
);
|
|
```
|
|
|
|
Compare that with:
|
|
|
|
```java
|
|
FileMessageBoard.publish(
|
|
"Tutorial/ConsoleIn/U/alice/E/text",
|
|
...
|
|
);
|
|
```
|
|
|
|
The conceptual operation is the same.
|
|
|
|
---
|
|
|
|
# 59. Subscribe
|
|
|
|
```java
|
|
client.subscribe(
|
|
"Tutorial/ConsoleIn/U/alice/E/text",
|
|
1
|
|
);
|
|
```
|
|
|
|
But receiving is asynchronous.
|
|
|
|
So we need a callback.
|
|
|
|
---
|
|
|
|
# 60. React to Publications
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
WatchService
|
|
```
|
|
|
|
has effectively become:
|
|
|
|
```text
|
|
MQTT callback
|
|
```
|
|
|
|
Again:
|
|
|
|
```text
|
|
transport changed
|
|
```
|
|
|
|
The architecture did not.
|
|
|
|
---
|
|
|
|
# 61. Rebuild Our Three Services
|
|
|
|
## ConsoleIn
|
|
|
|
Publishes:
|
|
|
|
```text
|
|
Tutorial/ConsoleIn/U/<unit>/E/text
|
|
```
|
|
|
|
## ROT13
|
|
|
|
Owns:
|
|
|
|
```text
|
|
Tutorial/ROT13/U/<unit>/I
|
|
|
|
Tutorial/ROT13/U/<unit>/S/...
|
|
|
|
Tutorial/ROT13/U/<unit>/E/crypted
|
|
```
|
|
|
|
## ConsoleOut
|
|
|
|
Owns:
|
|
|
|
```text
|
|
Tutorial/ConsoleOut/U/<unit>/I
|
|
|
|
Tutorial/ConsoleOut/U/<unit>/S/...
|
|
```
|
|
|
|
The service responsibilities are the same as before.
|
|
|
|
The runnable MQTT programs are:
|
|
|
|
- [`MqttConsoleIn.java`](supplemental-java/06-mqtt/src/main/java/MqttConsoleIn.java)
|
|
- [`MqttRot13.java`](supplemental-java/06-mqtt/src/main/java/MqttRot13.java)
|
|
- [`MqttConsoleOut.java`](supplemental-java/06-mqtt/src/main/java/MqttConsoleOut.java)
|
|
- [`MqttConfigure.java`](supplemental-java/06-mqtt/src/main/java/MqttConfigure.java)
|
|
|
|
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.java`](supplemental-java/06-mqtt/src/main/java/JsonTools.java)
|
|
configures Jackson;
|
|
- [`MqttTools.java`](supplemental-java/06-mqtt/src/main/java/MqttTools.java)
|
|
validates 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:
|
|
|
|
```bash
|
|
mvn package dependency:copy-dependencies
|
|
```
|
|
|
|
On Linux or macOS, use three terminals:
|
|
|
|
```bash
|
|
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:
|
|
|
|
```bash
|
|
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:
|
|
|
|
```java
|
|
client.subscribe(
|
|
"Tutorial/ConsoleIn/U/alice/E/text",
|
|
1
|
|
);
|
|
```
|
|
|
|
inside ROT13.
|
|
|
|
It works beautifully.
|
|
|
|
Then ask:
|
|
|
|
> **Who knows whom?**
|
|
|
|
ROT13 now contains:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
TOPIC some/path
|
|
```
|
|
|
|
was sufficient for learning the architecture.
|
|
|
|
Now we want richer Intent.
|
|
|
|
For example:
|
|
|
|
```json
|
|
{
|
|
"subscribe": {
|
|
"topic":
|
|
"Tutorial/ConsoleIn/U/alice/E/text"
|
|
}
|
|
}
|
|
```
|
|
|
|
or:
|
|
|
|
```json
|
|
{
|
|
"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:
|
|
|
|
```json
|
|
{"value":"Hello"}
|
|
```
|
|
|
|
to become:
|
|
|
|
```java
|
|
record Rot13Message(
|
|
String value
|
|
) {}
|
|
```
|
|
|
|
and:
|
|
|
|
```json
|
|
{
|
|
"subscribe": {
|
|
"topic": "..."
|
|
}
|
|
}
|
|
```
|
|
|
|
to become:
|
|
|
|
```java
|
|
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:
|
|
|
|
```xml
|
|
<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:
|
|
|
|
```java
|
|
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`](supplemental-java/06-mqtt/src/main/java/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
|
|
|
|
```java
|
|
Rot13Intent intent =
|
|
JsonTools
|
|
.deserializationObjectMapper
|
|
.readValue(
|
|
message.getPayload(),
|
|
Rot13Intent.class
|
|
);
|
|
```
|
|
|
|
Direct Intent:
|
|
|
|
```yaml
|
|
text:
|
|
value: Hello
|
|
```
|
|
|
|
can cause:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
Tutorial/ROT13/U/<unit>/I
|
|
```
|
|
|
|
Now publish:
|
|
|
|
```yaml
|
|
subscribe:
|
|
topic: Tutorial/ConsoleIn/U/alice/E/text
|
|
```
|
|
|
|
ROT13 executes:
|
|
|
|
```java
|
|
client.subscribe(
|
|
intent.subscribe().topic(),
|
|
1
|
|
);
|
|
```
|
|
|
|
Exactly the same architectural operation as:
|
|
|
|
```text
|
|
TOPIC consoleIn/outbox
|
|
```
|
|
|
|
in our file system.
|
|
|
|
Only the transport and serialization changed.
|
|
|
|
---
|
|
|
|
# 68. Rewire at Runtime
|
|
|
|
Publish later:
|
|
|
|
```yaml
|
|
subscribe:
|
|
topic: Tutorial/ConsoleIn/U/bob/E/text
|
|
```
|
|
|
|
If ROT13 accepts one input source at a time:
|
|
|
|
```java
|
|
if (dynamicTextTopic != null) {
|
|
client.unsubscribe(
|
|
dynamicTextTopic
|
|
);
|
|
}
|
|
```
|
|
|
|
Then:
|
|
|
|
```java
|
|
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:
|
|
|
|
```java
|
|
MqttTools.isValidTopic(...)
|
|
|
|
MqttTools.isValidFilter(...)
|
|
|
|
MqttTools.isTopicMatchingFilter(...)
|
|
```
|
|
|
|
For example:
|
|
|
|
```java
|
|
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`](supplemental-java/06-mqtt/src/main/java/MqttTools.java).
|
|
|
|
---
|
|
|
|
# 70. Status Should Explain Current Behaviour
|
|
|
|
After subscribing, publish:
|
|
|
|
```text
|
|
Tutorial/ROT13/U/<unit>/S/subscriptions/text
|
|
```
|
|
|
|
with:
|
|
|
|
```json
|
|
{
|
|
"value":
|
|
"Tutorial/ConsoleIn/U/alice/E/text"
|
|
}
|
|
```
|
|
|
|
Now the service can explain:
|
|
|
|
```text
|
|
why it currently reacts
|
|
to Alice's Events
|
|
```
|
|
|
|
That is Status.
|
|
|
|
Not Event.
|
|
|
|
---
|
|
|
|
# 71. Retained State, Non-Retained Events
|
|
|
|
An Event:
|
|
|
|
```text
|
|
buttonPressed
|
|
```
|
|
|
|
usually means:
|
|
|
|
```text
|
|
something happened at a particular time
|
|
```
|
|
|
|
A client joining later normally should not receive it as if it just occurred.
|
|
|
|
Therefore Events are typically:
|
|
|
|
```java
|
|
retained = false
|
|
```
|
|
|
|
Status such as:
|
|
|
|
```text
|
|
online = true
|
|
```
|
|
|
|
represents current condition.
|
|
|
|
A late subscriber may want that immediately.
|
|
|
|
Therefore Status may sensibly use:
|
|
|
|
```java
|
|
retained = true
|
|
```
|
|
|
|
MQTT now provides transport semantics that fit our architectural distinction.
|
|
|
|
---
|
|
|
|
# 72. Last Will
|
|
|
|
A service can normally publish:
|
|
|
|
```json
|
|
{"value":false}
|
|
```
|
|
|
|
when it shuts down.
|
|
|
|
But not if:
|
|
|
|
```text
|
|
the machine crashes
|
|
|
|
the process dies
|
|
|
|
the network disappears
|
|
```
|
|
|
|
MQTT provides Last Will.
|
|
|
|
Configure:
|
|
|
|
```java
|
|
var lastWillTopic =
|
|
statusTopic + "/online";
|
|
|
|
MqttMessage will =
|
|
new MqttMessage(
|
|
"{\"value\":false}".getBytes(),
|
|
1,
|
|
true,
|
|
new MqttProperties()
|
|
);
|
|
|
|
connectionOptions.setWill(
|
|
lastWillTopic,
|
|
will
|
|
);
|
|
```
|
|
|
|
After connecting:
|
|
|
|
```java
|
|
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:
|
|
|
|
```text
|
|
D — Description
|
|
|
|
U — Unit
|
|
|
|
I — Intent
|
|
|
|
S — Status
|
|
|
|
E — Event
|
|
```
|
|
|
|
Example:
|
|
|
|
```text
|
|
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:
|
|
|
|
```json
|
|
{
|
|
"value": "Hello"
|
|
}
|
|
```
|
|
|
|
and ROT13 understands exactly that structure.
|
|
|
|
Then:
|
|
|
|
```text
|
|
ConsoleIn/E/text
|
|
│
|
|
▼
|
|
ROT13
|
|
```
|
|
|
|
is sufficient.
|
|
|
|
The broker already handles transport-level delivery.
|
|
|
|
Likewise:
|
|
|
|
```text
|
|
ROT13/E/crypted
|
|
│
|
|
▼
|
|
ConsoleOut
|
|
```
|
|
|
|
if ConsoleOut understands that contract.
|
|
|
|
No explicit transformer is required.
|
|
|
|
---
|
|
|
|
# 75. Incompatible Services: Transformer Required
|
|
|
|
Suppose ROT13 produces:
|
|
|
|
```json
|
|
{
|
|
"value": "Uryyb"
|
|
}
|
|
```
|
|
|
|
but an actuator expects:
|
|
|
|
```json
|
|
{
|
|
"display": {
|
|
"text": "Uryyb",
|
|
"durationMs": 5000
|
|
}
|
|
}
|
|
```
|
|
|
|
Then introduce:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
1. Does it know another component's Java class?
|
|
2. Does it require another component in the same JVM?
|
|
3. Does it know another component's machine?
|
|
4. Does it know another component's programming language?
|
|
5. Does its capability contain a hard-coded producer identity?
|
|
6. Does it write State owned by somebody else?
|
|
7. Are multiple actors competing to own the same resource?
|
|
8. Which foreign message contracts must it genuinely understand?
|
|
9. Where is system topology defined?
|
|
10. Can topology change without recompiling the capability?
|
|
11. Could the transport be replaced without changing the core capability?
|
|
12. Does every mediator perform something that actually needs mediation?
|
|
|
|
The architecture does not necessarily answer:
|
|
|
|
```text
|
|
NO
|
|
```
|
|
|
|
to every question.
|
|
|
|
But the answer should always be intentional.
|
|
|
|
---
|
|
|
|
# 79. The Whole Journey
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
one source-like capability
|
|
|
|
one processing capability
|
|
|
|
one sink-like capability
|
|
```
|
|
|
|
You may use:
|
|
|
|
```text
|
|
filesystem
|
|
|
|
MQTT
|
|
|
|
or both
|
|
```
|
|
|
|
You must be able to explain:
|
|
|
|
```text
|
|
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:
|
|
|
|
```text
|
|
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:
|
|
|
|
```bash
|
|
java DynamicMediatorService.java mediator1
|
|
```
|
|
|
|
It permanently subscribes to:
|
|
|
|
```text
|
|
mediators/mediator1/inbox
|
|
```
|
|
|
|
Every valid `ROUTE` Intent contains one atomic pair:
|
|
|
|
```text
|
|
ROUTE
|
|
<from mailbox>
|
|
<to mailbox>
|
|
```
|
|
|
|
When another `ROUTE` arrives, the mediator:
|
|
|
|
1. validates both mailbox names;
|
|
2. subscribes to the new source if it changed;
|
|
3. unsubscribes from the old source;
|
|
4. replaces its current route;
|
|
5. 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`](supplemental-java/05-dynamic-filesystem/DynamicMediatorService.java).
|
|
|
|
## Send Rewiring Intent
|
|
|
|
The short-lived orchestrator accepts:
|
|
|
|
```text
|
|
mediator name
|
|
from mailbox
|
|
to mailbox
|
|
```
|
|
|
|
and publishes the corresponding `ROUTE` Intent:
|
|
|
|
```bash
|
|
java DynamicMediatorOrchestrator.java \
|
|
mediator1 \
|
|
consoleIn/outbox \
|
|
rot13/inbox
|
|
```
|
|
|
|
Complete source:
|
|
[`DynamicMediatorOrchestrator.java`](supplemental-java/05-dynamic-filesystem/DynamicMediatorOrchestrator.java).
|
|
|
|
## Run the Rewirable Version
|
|
|
|
Work from `supplemental-java/05-dynamic-filesystem`. Start five long-running
|
|
programs with a fresh message-board root:
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable SensorService.java
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable \
|
|
DynamicMediatorService.java mediator1
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable \
|
|
DynamicProcessorService.java
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable \
|
|
DynamicMediatorService.java mediator2
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable \
|
|
DynamicActuatorService.java
|
|
```
|
|
|
|
After they are waiting, establish the mediated ROT13 path with two short-lived
|
|
orchestrator calls:
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable \
|
|
DynamicMediatorOrchestrator.java \
|
|
mediator1 consoleIn/outbox rot13/inbox
|
|
```
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable \
|
|
DynamicMediatorOrchestrator.java \
|
|
mediator2 rot13/outbox consoleOut/inbox
|
|
```
|
|
|
|
Type:
|
|
|
|
```text
|
|
Hello
|
|
```
|
|
|
|
ConsoleOut should receive:
|
|
|
|
```text
|
|
Uryyb
|
|
```
|
|
|
|
Now rewire `mediator1` while every long-running process remains alive:
|
|
|
|
```bash
|
|
java -DmessageBoard.root=messageBoard-rewirable \
|
|
DynamicMediatorOrchestrator.java \
|
|
mediator1 consoleIn/outbox consoleOut/inbox
|
|
```
|
|
|
|
Type:
|
|
|
|
```text
|
|
Hello
|
|
```
|
|
|
|
This time ConsoleOut should receive the unmodified text:
|
|
|
|
```text
|
|
Hello
|
|
```
|
|
|
|
Restore the ROT13 route without restarting the mediator:
|
|
|
|
```bash
|
|
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:
|
|
|
|
```text
|
|
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`](supplemental-java/01-procedural/ProceduralRot13.java)
|
|
|
|
```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`](supplemental-java/02-static-in-jvm/StaticQueueRot13.java)
|
|
|
|
```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`](supplemental-java/03-static-filesystem/Message.java)
|
|
|
|
```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`](supplemental-java/03-static-filesystem/FileMessageBoard.java)
|
|
|
|
```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`](supplemental-java/03-static-filesystem/SensorService.java)
|
|
|
|
```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`](supplemental-java/03-static-filesystem/ProcessorService.java)
|
|
|
|
```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`](supplemental-java/03-static-filesystem/ActuatorService.java)
|
|
|
|
```java
|
|
void main() throws Exception {
|
|
FileMessageBoard.watch("consoleOut/inbox", message -> {
|
|
if (message.type().equals("TEXT"))
|
|
IO.println(message.value());
|
|
});
|
|
}
|
|
```
|
|
|
|
### [`Mediator.java`](supplemental-java/03-static-filesystem/Mediator.java)
|
|
|
|
```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`](supplemental-java/03-static-filesystem/ConfigurableMediator.java)
|
|
|
|
```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`](supplemental-java/03-static-filesystem/MediatorOrchestrator.java)
|
|
|
|
```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`](supplemental-java/04-dynamic-in-jvm/DynamicQueueRot13.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/Message.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/FileMessageBoard.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/FileSubscriptions.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/SensorService.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/DynamicProcessorService.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/DynamicActuatorService.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/Configure.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/DynamicMediatorService.java)
|
|
|
|
```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`](supplemental-java/05-dynamic-filesystem/DynamicMediatorOrchestrator.java)
|
|
|
|
```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`](supplemental-java/06-mqtt/pom.xml)
|
|
|
|
```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`](supplemental-java/06-mqtt/src/main/java/JsonTools.java)
|
|
|
|
```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`](supplemental-java/06-mqtt/src/main/java/MqttTools.java)
|
|
|
|
```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`](supplemental-java/06-mqtt/src/main/java/MqttConsoleIn.java)
|
|
|
|
```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`](supplemental-java/06-mqtt/src/main/java/MqttRot13.java)
|
|
|
|
```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`](supplemental-java/06-mqtt/src/main/java/MqttConsoleOut.java)
|
|
|
|
```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`](supplemental-java/06-mqtt/src/main/java/MqttConfigure.java)
|
|
|
|
```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");
|
|
}
|
|
```
|