Skip to content
This repository was archived by the owner on Mar 27, 2025. It is now read-only.

Commit 00c0211

Browse files
njlbennnjlbenn
authored
Add a part to be implemented in ListenNotifyResource (#20)
1 parent 9c02efa commit 00c0211

File tree

5 files changed

+103
-57
lines changed

5 files changed

+103
-57
lines changed

‎code/exercise_013_Listen_and_Notify/README.md‎

Lines changed: 55 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,61 @@ In this exercise, we will play a bit with Postgres’ `LISTEN / NOTIFY` feature,
44

55
The `LISTEN / NOTIFY` feature of Postgres allows you to setup a connection to Postgres, and listen for evens that pass by on a channel, as well as notifying such channels. With the reactive sql clients, we can connect to these channels as a `Multi` in Quarkus.
66

7-
* Get the code for `ListenNotifyResource` by executing this command from the command line: `cmtc pull-template src/main/java/com/lunatech/training/quarkus/ListenNotifyResource.java <root folder of exercise repo>`.
7+
* Create the class `ListenNotifyResource` from the following template:
8+
9+
```java
10+
import io.smallrye.mutiny.Multi;
11+
import io.smallrye.mutiny.Uni;
12+
import io.vertx.core.json.JsonObject;
13+
import io.vertx.mutiny.pgclient.PgConnection;
14+
import io.vertx.mutiny.pgclient.PgPool;
15+
import io.vertx.pgclient.PgNotification;
16+
import org.jboss.resteasy.reactive.RestStreamElementType;
17+
18+
import jakarta.inject.Inject;
19+
import jakarta.ws.rs.*;
20+
import jakarta.ws.rs.core.MediaType;
21+
22+
@Path("/channel")
23+
public class ListenNotifyResource {
24+
25+
@Inject
26+
PgPool client;
27+
28+
@Path("{channel}")
29+
@GET
30+
@Produces(MediaType.SERVER_SENT_EVENTS)
31+
@RestStreamElementType(MediaType.APPLICATION_JSON)
32+
public Multi<JsonObject> listen(@PathParam("channel") String channel) {
33+
return client
34+
.getConnection()
35+
.map(PgConnection::cast)
36+
.toMulti()
37+
.flatMap(connection ->
38+
connection.query("LISTEN " + channel)
39+
.execute()
40+
.toMulti()
41+
.flatMap(__ -> streamNotifications(connection))
42+
)
43+
.map(PgNotification::toJson);
44+
}
45+
46+
@Path("{channel}")
47+
@POST
48+
@Produces(MediaType.TEXT_PLAIN)
49+
@Consumes(MediaType.WILDCARD)
50+
public Uni<String> notify(@PathParam("channel") String channel, String stuff) {
51+
return client.preparedQuery("NOTIFY " + channel + ", $$" + stuff + "$$")
52+
.execute()
53+
.map(rs -> "Posted to " + channel + " channel");
54+
}
55+
56+
// Use PgConnection::notificationHandler to register a handler that emits PgNotification values on a Multi stream
57+
private Multi<PgNotification> streamNotifications(PgConnection connection) {
58+
// To be implemented
59+
}
60+
}
61+
```
862

963
* Connect to the channel `milkshakes` using the following cURL command.
1064

‎code/exercise_013_Listen_and_Notify/src/main/java/com/lunatech/training/quarkus/ListenNotifyResource.java‎

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import io.vertx.core.json.JsonObject;
66
import io.vertx.mutiny.pgclient.PgConnection;
77
import io.vertx.mutiny.pgclient.PgPool;
8-
import io.vertx.mutiny.sqlclient.SqlConnection;
98
import io.vertx.pgclient.PgNotification;
109
import org.jboss.resteasy.reactive.RestStreamElementType;
1110

@@ -26,31 +25,30 @@ public class ListenNotifyResource {
2625
public Multi<JsonObject> listen(@PathParam("channel") String channel) {
2726
return client
2827
.getConnection()
28+
.map(PgConnection::cast)
2929
.toMulti()
30-
.flatMap(connection -> {
31-
Multi<PgNotification> notifications = Multi.createFrom().
32-
emitter(c -> toPgConnection(connection).notificationHandler(c::emit));
33-
34-
return connection.query("LISTEN " + channel)
35-
.execute()
36-
.toMulti()
37-
.flatMap(__ -> notifications);
38-
})
30+
.flatMap(connection ->
31+
connection.query("LISTEN " + channel)
32+
.execute()
33+
.toMulti()
34+
.flatMap(__ -> streamNotifications(connection))
35+
)
3936
.map(PgNotification::toJson);
4037
}
4138

4239
@Path("{channel}")
4340
@POST
4441
@Produces(MediaType.TEXT_PLAIN)
4542
@Consumes(MediaType.WILDCARD)
46-
public Uni<String> notif(@PathParam("channel") String channel, String stuff) {
43+
public Uni<String> notify(@PathParam("channel") String channel, String stuff) {
4744
return client.preparedQuery("NOTIFY " + channel + ", $$" + stuff + "$$")
4845
.execute()
4946
.map(rs -> "Posted to " + channel + " channel");
5047
}
5148

52-
// We have to do some type juggling here. Solved in the mutiny client v2.
53-
PgConnection toPgConnection(SqlConnection sqlConnection) {
54-
return new PgConnection((io.vertx.pgclient.PgConnection) sqlConnection.getDelegate());
49+
// Use PgConnection::notificationHandler to register a handler that emits PgNotification values on a Multi stream
50+
private Multi<PgNotification> streamNotifications(PgConnection connection) {
51+
return Multi.createFrom()
52+
.emitter(multiEmitter -> connection.notificationHandler(multiEmitter::emit));
5553
}
5654
}

‎code/exercise_014_Internal_Channels/src/main/java/com/lunatech/training/quarkus/ListenNotifyResource.java‎

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import io.vertx.core.json.JsonObject;
66
import io.vertx.mutiny.pgclient.PgConnection;
77
import io.vertx.mutiny.pgclient.PgPool;
8-
import io.vertx.mutiny.sqlclient.SqlConnection;
98
import io.vertx.pgclient.PgNotification;
109
import org.jboss.resteasy.reactive.RestStreamElementType;
1110

@@ -26,31 +25,30 @@ public class ListenNotifyResource {
2625
public Multi<JsonObject> listen(@PathParam("channel") String channel) {
2726
return client
2827
.getConnection()
28+
.map(PgConnection::cast)
2929
.toMulti()
30-
.flatMap(connection -> {
31-
Multi<PgNotification> notifications = Multi.createFrom().
32-
emitter(c -> toPgConnection(connection).notificationHandler(c::emit));
33-
34-
return connection.query("LISTEN " + channel)
35-
.execute()
36-
.toMulti()
37-
.flatMap(__ -> notifications);
38-
})
30+
.flatMap(connection ->
31+
connection.query("LISTEN " + channel)
32+
.execute()
33+
.toMulti()
34+
.flatMap(__ -> streamNotifications(connection))
35+
)
3936
.map(PgNotification::toJson);
4037
}
4138

4239
@Path("{channel}")
4340
@POST
4441
@Produces(MediaType.TEXT_PLAIN)
4542
@Consumes(MediaType.WILDCARD)
46-
public Uni<String> notif(@PathParam("channel") String channel, String stuff) {
43+
public Uni<String> notify(@PathParam("channel") String channel, String stuff) {
4744
return client.preparedQuery("NOTIFY " + channel + ", $$" + stuff + "$$")
4845
.execute()
4946
.map(rs -> "Posted to " + channel + " channel");
5047
}
5148

52-
// We have to do some type juggling here. Solved in the mutiny client v2.
53-
PgConnection toPgConnection(SqlConnection sqlConnection) {
54-
return new PgConnection((io.vertx.pgclient.PgConnection) sqlConnection.getDelegate());
49+
// Use PgConnection::notificationHandler to register a handler that emits PgNotification values on a Multi stream
50+
private Multi<PgNotification> streamNotifications(PgConnection connection) {
51+
return Multi.createFrom()
52+
.emitter(multiEmitter -> connection.notificationHandler(multiEmitter::emit));
5553
}
5654
}

‎code/exercise_015_Connecting_to_Kafka/src/main/java/com/lunatech/training/quarkus/ListenNotifyResource.java‎

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import io.vertx.core.json.JsonObject;
66
import io.vertx.mutiny.pgclient.PgConnection;
77
import io.vertx.mutiny.pgclient.PgPool;
8-
import io.vertx.mutiny.sqlclient.SqlConnection;
98
import io.vertx.pgclient.PgNotification;
109
import org.jboss.resteasy.reactive.RestStreamElementType;
1110

@@ -26,31 +25,30 @@ public class ListenNotifyResource {
2625
public Multi<JsonObject> listen(@PathParam("channel") String channel) {
2726
return client
2827
.getConnection()
28+
.map(PgConnection::cast)
2929
.toMulti()
30-
.flatMap(connection -> {
31-
Multi<PgNotification> notifications = Multi.createFrom().
32-
emitter(c -> toPgConnection(connection).notificationHandler(c::emit));
33-
34-
return connection.query("LISTEN " + channel)
35-
.execute()
36-
.toMulti()
37-
.flatMap(__ -> notifications);
38-
})
30+
.flatMap(connection ->
31+
connection.query("LISTEN " + channel)
32+
.execute()
33+
.toMulti()
34+
.flatMap(__ -> streamNotifications(connection))
35+
)
3936
.map(PgNotification::toJson);
4037
}
4138

4239
@Path("{channel}")
4340
@POST
4441
@Produces(MediaType.TEXT_PLAIN)
4542
@Consumes(MediaType.WILDCARD)
46-
public Uni<String> notif(@PathParam("channel") String channel, String stuff) {
43+
public Uni<String> notify(@PathParam("channel") String channel, String stuff) {
4744
return client.preparedQuery("NOTIFY " + channel + ", $$" + stuff + "$$")
4845
.execute()
4946
.map(rs -> "Posted to " + channel + " channel");
5047
}
5148

52-
// We have to do some type juggling here. Solved in the mutiny client v2.
53-
PgConnection toPgConnection(SqlConnection sqlConnection) {
54-
return new PgConnection((io.vertx.pgclient.PgConnection) sqlConnection.getDelegate());
49+
// Use PgConnection::notificationHandler to register a handler that emits PgNotification values on a Multi stream
50+
private Multi<PgNotification> streamNotifications(PgConnection connection) {
51+
return Multi.createFrom()
52+
.emitter(multiEmitter -> connection.notificationHandler(multiEmitter::emit));
5553
}
5654
}

‎code/exercise_016_Dead_Letter_Queue_and_Stream_filtering/src/main/java/com/lunatech/training/quarkus/ListenNotifyResource.java‎

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import io.vertx.core.json.JsonObject;
66
import io.vertx.mutiny.pgclient.PgConnection;
77
import io.vertx.mutiny.pgclient.PgPool;
8-
import io.vertx.mutiny.sqlclient.SqlConnection;
98
import io.vertx.pgclient.PgNotification;
109
import org.jboss.resteasy.reactive.RestStreamElementType;
1110

@@ -26,31 +25,30 @@ public class ListenNotifyResource {
2625
public Multi<JsonObject> listen(@PathParam("channel") String channel) {
2726
return client
2827
.getConnection()
28+
.map(PgConnection::cast)
2929
.toMulti()
30-
.flatMap(connection -> {
31-
Multi<PgNotification> notifications = Multi.createFrom().
32-
emitter(c -> toPgConnection(connection).notificationHandler(c::emit));
33-
34-
return connection.query("LISTEN " + channel)
35-
.execute()
36-
.toMulti()
37-
.flatMap(__ -> notifications);
38-
})
30+
.flatMap(connection ->
31+
connection.query("LISTEN " + channel)
32+
.execute()
33+
.toMulti()
34+
.flatMap(__ -> streamNotifications(connection))
35+
)
3936
.map(PgNotification::toJson);
4037
}
4138

4239
@Path("{channel}")
4340
@POST
4441
@Produces(MediaType.TEXT_PLAIN)
4542
@Consumes(MediaType.WILDCARD)
46-
public Uni<String> notif(@PathParam("channel") String channel, String stuff) {
43+
public Uni<String> notify(@PathParam("channel") String channel, String stuff) {
4744
return client.preparedQuery("NOTIFY " + channel + ", $$" + stuff + "$$")
4845
.execute()
4946
.map(rs -> "Posted to " + channel + " channel");
5047
}
5148

52-
// We have to do some type juggling here. Solved in the mutiny client v2.
53-
PgConnection toPgConnection(SqlConnection sqlConnection) {
54-
return new PgConnection((io.vertx.pgclient.PgConnection) sqlConnection.getDelegate());
49+
// Use PgConnection::notificationHandler to register a handler that emits PgNotification values on a Multi stream
50+
private Multi<PgNotification> streamNotifications(PgConnection connection) {
51+
return Multi.createFrom()
52+
.emitter(multiEmitter -> connection.notificationHandler(multiEmitter::emit));
5553
}
5654
}

0 commit comments

Comments
 (0)