-
Notifications
You must be signed in to change notification settings - Fork 172
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Eric Deandrea <edeandrea@redhat.com>
- Loading branch information
Showing
6 changed files
with
199 additions
and
279 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
105 changes: 19 additions & 86 deletions
105
.../src/main/java/io/quarkus/sample/superheroes/statistics/endpoint/EventStatsWebSocket.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,117 +1,50 @@ | ||
| package io.quarkus.sample.superheroes.statistics.endpoint; | ||
|
|
||
| import java.io.IOException; | ||
| import java.nio.ByteBuffer; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.ConcurrentMap; | ||
|
|
||
| import jakarta.annotation.PostConstruct; | ||
| import jakarta.annotation.PreDestroy; | ||
| import jakarta.inject.Inject; | ||
| import jakarta.websocket.CloseReason; | ||
| import jakarta.websocket.CloseReason.CloseCodes; | ||
| import jakarta.websocket.OnClose; | ||
| import jakarta.websocket.OnMessage; | ||
| import jakarta.websocket.OnOpen; | ||
| import jakarta.websocket.PongMessage; | ||
| import jakarta.websocket.Session; | ||
|
|
||
| import org.jboss.logging.Logger; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import io.quarkus.websockets.next.OnClose; | ||
| import io.quarkus.websockets.next.OnOpen; | ||
| import io.quarkus.websockets.next.OnPongMessage; | ||
| import io.quarkus.websockets.next.WebSocketConnection; | ||
|
|
||
| import io.smallrye.mutiny.Multi; | ||
| import io.smallrye.mutiny.subscription.Cancellable; | ||
| import io.smallrye.mutiny.unchecked.Unchecked; | ||
| import io.vertx.core.buffer.Buffer; | ||
|
|
||
| /** | ||
| * Base WebSocket endpoint which handles caching the stream and replaying | ||
| * the last event from a stream upon subscription by new socket clients | ||
| * @param <V> The object type inside each event | ||
| */ | ||
| public abstract class EventStatsWebSocket<V> { | ||
| private final ConcurrentMap<Session, Cancellable> sessions = new ConcurrentHashMap<>(); | ||
| private Multi<String> cachedStream; | ||
| private Logger logger; | ||
|
|
||
| @Inject | ||
| ObjectMapper mapper; | ||
|
|
||
| protected abstract Multi<V> getStream(); | ||
|
|
||
| @OnOpen | ||
| public void onOpen(Session session) { | ||
| this.logger.debugf("Opening session with id %s", session.getId()); | ||
| this.sessions.put(session, createSubscription(session)); | ||
| public Multi<V> onOpen(WebSocketConnection connection) { | ||
| this.logger.debugf("Opening connection with id %s", connection.id()); | ||
|
|
||
| return Multi.createBy() | ||
| .replaying() | ||
| .upTo(1) | ||
| .ofMulti(getStream()) | ||
| .invoke(v -> this.logger.infof("[Connection %s] - Writing message %s", connection.id(), v)); | ||
| } | ||
|
|
||
| @OnMessage | ||
| public void onPongMessage(Session session, PongMessage pongMessage) { | ||
| this.logger.debugf("Got pong message (%s) from session %s", new String(pongMessage.getApplicationData().array(), StandardCharsets.UTF_8), session.getId()); | ||
| @OnPongMessage | ||
| public void onPongMessage(WebSocketConnection connection, Buffer pongMessage) { | ||
| this.logger.debugf("Got pong message (%s) on %s from connection %s", pongMessage.toString(), connection.handshakeRequest().path(), connection.id()); | ||
| } | ||
|
|
||
| @OnClose | ||
| public void onClose(Session session) { | ||
| this.logger.debugf("Closing session with id %s", session.getId()); | ||
| Optional.ofNullable(this.sessions.remove(session)) | ||
| .ifPresent(Cancellable::cancel); | ||
| public void onClose(WebSocketConnection connection) { | ||
| this.logger.debugf("Closing connection with id %s", connection.id()); | ||
| } | ||
|
|
||
| @PostConstruct | ||
| public void initialize() { | ||
| this.logger = Logger.getLogger(getClass()); | ||
| this.cachedStream = Multi.createBy().replaying().upTo(1).ofMulti(getStream()) | ||
| .map(Unchecked.function(this.mapper::writeValueAsString)); | ||
| } | ||
|
|
||
| @PreDestroy | ||
| public void cleanup() { | ||
| this.sessions.forEach((session, subscription) -> { | ||
| subscription.cancel(); | ||
|
|
||
| if (session.isOpen()) { | ||
| try { | ||
| session.close(new CloseReason(CloseCodes.GOING_AWAY, "Server shutting down")); | ||
| } | ||
| catch (IOException ex) { | ||
| this.logger.errorf(ex, "Got exception (%s) while closing session", ex.getClass().getName()); | ||
| } | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| void sendPings() { | ||
| this.sessions.keySet() | ||
| .stream() | ||
| .filter(Session::isOpen) | ||
| .forEach(this::sendPing); | ||
| } | ||
|
|
||
| private void sendPing(Session session) { | ||
| this.logger.debugf("Sending ping to session %s", session.getId()); | ||
|
|
||
| try { | ||
| session.getAsyncRemote().sendPing(ByteBuffer.wrap("PING".getBytes(StandardCharsets.UTF_8))); | ||
| } | ||
| catch (IOException e) { | ||
| this.logger.errorf(e, "Got error sending ping: %s", e.getMessage()); | ||
| } | ||
| } | ||
|
|
||
| private Cancellable createSubscription(Session session) { | ||
| return Optional.ofNullable(this.cachedStream) | ||
| .orElseThrow(() -> new IllegalArgumentException("Cached stream (Multi<String>) has not been created. Please initialize it inside an @PostConstruct method.")) | ||
| .subscribe().with(serialized -> write(session, serialized)); | ||
| } | ||
|
|
||
| private void write(Session session, String text) { | ||
| this.logger.infof("[Session %s] - Writing message %s", session.getId(), text); | ||
|
|
||
| session.getAsyncRemote().sendText(text, result -> { | ||
| if (result.getException() != null) { | ||
| this.logger.error("Unable to write message to web socket", result.getException()); | ||
| } | ||
| }); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.