simple chat

pom.xml

@@ -6,17 +6,21 @@
     <groupId>dev.struchkov.example</groupId>
     <artifactId>quarkus-websocket</artifactId>
     <version>1.0-SNAPSHOT</version>
+
     <properties>
         <compiler-plugin.version>3.11.0</compiler-plugin.version>
         <maven.compiler.release>17</maven.compiler.release>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+
         <quarkus.platform.artifact-id>quarkus-bom</quarkus.platform.artifact-id>
         <quarkus.platform.group-id>io.quarkus.platform</quarkus.platform.group-id>
         <quarkus.platform.version>3.1.0.Final</quarkus.platform.version>
+
         <skipITs>true</skipITs>
         <surefire-plugin.version>3.0.0</surefire-plugin.version>
     </properties>
+
     <dependencyManagement>
         <dependencies>
             <dependency>
@@ -28,6 +32,7 @@
             </dependency>
         </dependencies>
     </dependencyManagement>
+
     <dependencies>
         <dependency>
             <groupId>io.quarkus</groupId>
@@ -41,16 +46,26 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-smallrye-graphql</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-reactive-routes</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-arc</artifactId>
         </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
-            <artifactId>quarkus-junit5</artifactId>
+            <artifactId>quarkus-jackson</artifactId>
-            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.26</version>
         </dependency>
     </dependencies>
+
     <build>
         <plugins>
             <plugin>
@@ -109,6 +124,7 @@
             </plugin>
         </plugins>
     </build>
+
     <profiles>
         <profile>
             <id>native</id>
@@ -123,4 +139,14 @@
             </properties>
         </profile>
     </profiles>
+
+    <developers>
+        <developer>
+            <id>uPagge</id>
+            <name>Struchkov Mark</name>
+            <email>mark@struchkov.dev</email>
+            <url>https://mark.struchkov.dev</url>
+        </developer>
+    </developers>
+
 </project>

src/main/java/dev/struchkov/example/ChatInputMessage.java

@@ -0,0 +1,14 @@
+package dev.struchkov.example;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+@Getter
+@Setter
+@ToString
+public class ChatInputMessage {
+
+    private String text;
+
+}

src/main/java/dev/struchkov/example/ChatMessageDecoder.java

@@ -0,0 +1,34 @@
+package dev.struchkov.example;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import jakarta.websocket.DecodeException;
+import jakarta.websocket.Decoder;
+import lombok.SneakyThrows;
+
+public class ChatMessageDecoder implements Decoder.Text<ChatInputMessage> {
+
+    private final ObjectMapper jackson = ChatMessageDecoder.getJackson();
+
+    @Override
+    @SneakyThrows
+    public ChatInputMessage decode(String s) throws DecodeException {
+        return jackson.readValue(s, ChatInputMessage.class);
+    }
+
+    @Override
+    public boolean willDecode(String s) {
+        return s != null;
+    }
+
+    public static ObjectMapper getJackson() {
+        ObjectMapper om = new ObjectMapper();
+        om.configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true);
+        om.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
+        om.registerModule(new JavaTimeModule());
+        return om;
+    }
+
+}

src/main/java/dev/struchkov/example/ChatMessageEncoder.java

@@ -0,0 +1,18 @@
+package dev.struchkov.example;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.websocket.EncodeException;
+import jakarta.websocket.Encoder;
+import lombok.SneakyThrows;
+
+public class ChatMessageEncoder implements Encoder.Text<ChatOutputMessage> {
+
+    private final ObjectMapper jackson = ChatMessageDecoder.getJackson();
+
+    @Override
+    @SneakyThrows
+    public String encode(ChatOutputMessage chatOutputMessage) throws EncodeException {
+        return jackson.writeValueAsString(chatOutputMessage);
+    }
+
+}

src/main/java/dev/struchkov/example/ChatOutputMessage.java

@@ -0,0 +1,19 @@
+package dev.struchkov.example;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+import java.util.UUID;
+
+@Getter
+@Setter
+@ToString
+@AllArgsConstructor
+public class ChatOutputMessage {
+
+    private UUID fromUserId;
+    private String text;
+
+}

src/main/java/dev/struchkov/example/StartWebSocket.java

@@ -1,7 +1,6 @@
 package dev.struchkov.example;
 
 import jakarta.enterprise.context.ApplicationScoped;
-import jakarta.websocket.EncodeException;
 import jakarta.websocket.OnClose;
 import jakarta.websocket.OnError;
 import jakarta.websocket.OnMessage;
@@ -9,33 +8,73 @@ import jakarta.websocket.OnOpen;
 import jakarta.websocket.Session;
 import jakarta.websocket.server.PathParam;
 import jakarta.websocket.server.ServerEndpoint;
+import lombok.RequiredArgsConstructor;
 
-import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 
-import static java.util.Objects.requireNonNull;
-
-@ServerEndpoint("/start-websocket/{name}")
 @ApplicationScoped
+@ServerEndpoint(
+        value = "/chat/{chatId}",
+        decoders = ChatMessageDecoder.class,
+        encoders = ChatMessageEncoder.class
+)
+@RequiredArgsConstructor
 public class StartWebSocket {
 
+    public static final ThreadLocal<UUID> CURRENT_USER = new ThreadLocal<>();
+    private final Map<String, List<Session>> sessions = new ConcurrentHashMap<>();
+
     @OnOpen
-    public void onOpen(Session session, @PathParam("name") String name) {
+    public void onOpen(Session session, @PathParam("chatId") String chatId) {
-        System.out.println("onOpen> " + name);
+        System.out.println("onOpen> " + chatId);
+        sessions.computeIfAbsent(chatId, key -> new ArrayList<>()).add(session);
     }
 
     @OnClose
-    public void onClose(Session session, @PathParam("name") String name) {
+    public void onClose(Session session, @PathParam("chatId") String chatId) {
-        System.out.println("onClose> " + name);
+        System.out.println("onClose> " + chatId);
+        closeSession(session, chatId);
     }
 
     @OnError
-    public void onError(Session session, @PathParam("name") String name, Throwable throwable) {
+    public void onError(Session session, @PathParam("chatId") String chatId, Throwable throwable) {
-        System.out.println("onError> " + name + ": " + throwable);
+        System.out.println("onError> " + chatId + ": " + throwable);
     }
 
     @OnMessage
-    public void onMessage(String message, @PathParam("name") String name) {
+    public void onMessage(Session session, @PathParam("chatId") String chatId, ChatInputMessage message) {
-        System.out.println("onMessage> " + name + ": " + message);
+        System.out.println("onMessage> " + chatId + ": " + message);
+        sendMessage(session, chatId, message);
+    }
+
+    private void sendMessage(Session session, String chatId, ChatInputMessage message) {
+        final List<Session> chatSessions = sessions.get(chatId);
+        for (Session chatSession : chatSessions) {
+            if (session.getId().equals(chatSession.getId())) {
+                continue;
+            }
+            final UUID fromUserId = CURRENT_USER.get();
+            final ChatOutputMessage outputMessage = new ChatOutputMessage(fromUserId, message.getText());
+            chatSession.getAsyncRemote().sendObject(outputMessage);
+            CURRENT_USER.remove();
+        }
+    }
+
+    private void closeSession(Session session, String chatId) {
+        final List<Session> chatSessions = sessions.get(chatId);
+        final Iterator<Session> sessionIterator = chatSessions.iterator();
+        while (sessionIterator.hasNext()) {
+            final Session chatSession = sessionIterator.next();
+            if (session.getId().equals(chatSession.getId())) {
+                sessionIterator.remove();
+                break;
+            }
+        }
     }
 
 }

src/main/java/dev/struchkov/example/WebsocketAuthFilter.java

@@ -0,0 +1,52 @@
+package dev.struchkov.example;
+
+import io.quarkus.vertx.web.RouteFilter;
+import io.vertx.core.http.Cookie;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.ext.web.RoutingContext;
+import io.vertx.ext.web.handler.HttpException;
+import lombok.RequiredArgsConstructor;
+
+import java.util.UUID;
+
+@RequiredArgsConstructor
+public class WebsocketAuthFilter {
+
+    @RouteFilter(100)
+    void authFilter(RoutingContext rc) {
+        final HttpServerRequest currentRequest = rc.request();
+
+        if (isWebsocketRequest(currentRequest)) {
+            final Cookie authCookie = currentRequest.getCookie("sessionId");
+            if (authCookie == null) {
+                throw new HttpException(401, "Не передан параметр авторизации.");
+            }
+
+            final String authValue = authCookie.getValue();
+            if (!authLogic(authValue)) {
+                throw new HttpException(403, "Пользователь не авторизован.");
+            }
+        }
+
+        rc.next();
+    }
+
+    private static boolean isWebsocketRequest(HttpServerRequest currentRequest) {
+        return currentRequest.headers().contains("Upgrade")
+               && "websocket".equals(currentRequest.getHeader("Upgrade"));
+    }
+
+    private boolean authLogic(String sessionId) {
+        // your auth logic here
+        if (sessionId.equals("user1")) {
+            StartWebSocket.CURRENT_USER.set(UUID.fromString("09e429de-a302-40b6-9d10-6b113ab9e89d"));
+            return true;
+        } else if (sessionId.equals("user2")) {
+            StartWebSocket.CURRENT_USER.set(UUID.fromString("f84dbae1-f9a9-4c37-8922-4eb207103676"));
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+}