EventPiper

The service listens directly to PostgreSQL notification channel new_events and consume all the new events created in events table. The database schema could be found in postgresql/schema.sql file. The service uses only 1 connection to digest all the notifications from PostgreSQL, and it pipes the notifications to concerned live subscribers. So if we scale the services to n instances and share the load, they would consume only n connections in normal usage (when on-demand historical events replay is not requested).

Build and run tests

$ make build

To run it without Elixir or PostgreSQL (need to run make build first)

$ make run

Use the PUSH API

$ curl -v localhost:4000/events -H "X-Consumer-Id: toto"

You should not see anything but some heartbeat events.

Keep the previous call open. Now, let's create some events in the database:

$ docker exec -it postgresql bash

(in the container)
$ psql -d event_piper -U event_piper

(in psql)
# insert into events (type, timestamp, subscriber, payload) values ('insert', now(), 'toto', '{"foo": "bar"}');

You should see a new event in the previous terminal:

id: 1
event: insert
data: {"foo":"bar"}

You can create more and all the events should be pushed to the terminal directly.

Historical events replay

In the case of a broken connection or missing events, we could use an extra header Last-Event-ID to go back in history and fetch the missing events.

$ curl -v localhost:4000/events -H "X-Consumer-Id: toto" -H "Last-Event-ID: 5"

You should see immediately some responses streamed from the database:

id: 6
event: insert
data: {"foo":"bar"}


id: 7
event: insert
data: {"foo":"bar"}


id: 8
event: insert
data: {"foo":"bar"}

...

Java client example (with Reactor)

WebClient client = WebClient.create("http://localhost:4000");
ParameterizedTypeReference<ServerSentEvent<String>> type = new ParameterizedTypeReference<ServerSentEvent<String>>() {};

Flux<ServerSentEvent<String>> eventStream = client.get()
    .uri("/events")
    .header("X-Consumer-Id", "toto")
    .header("Last-Event-Id", "4")
    .retrieve()
    .bodyToFlux(type);

Now, if you have 2 sessions open for toto, and you insert another event into the database, you should see the pushed event in both.

TODO

• Database schema is very primitive. If we prefer an append-only approach, we should add more fields to allow event patching.
• Complete CI + proper config management
• The historical replay is using a PostgreSQL streaming approach, which could check out a connection for long time if the number of events is huge.