exposing client ExecutorService for multi-threading support

Author: jkyberneees

README.md

@@ -147,6 +147,31 @@ client.on("topic1.*", (Object... eventArgs) -> {
 });
 ```
 
+#### Concurrency Support 
+
+The underlying EventEmitter implementation uses a single thread for event handling. If you need to handle events concurrently, 
+consider using a thread pool or executor service to process events in parallel.
+
+You can access the client `ExecutorService` instance by calling the `getExecutorService()` method:
+
+```java
+client.on("secure/inbound.gettime", (Object... eventArgs) -> {
+    var replyFn = (ReplyFunction) eventArgs[1];
+    logger.info("Responding to gettime request on a separate thread...");
+
+    client.getExecutorService().submit(() -> {
+        try {
+            var response = Map.of("time", new Date());
+            // Send a reply and wait for acknowledgment
+            replyFn.reply(response, "ok", false).waitForAck().get();
+            logger.info("Response delivered!");
+        } catch (Exception e) {
+            logger.log(Level.SEVERE, "Failed to send reply", e);
+        }
+    });
+});
+```
+
 ### Publishing Messages
 
 Publish messages to a topic:
@@ -254,6 +279,12 @@ Creates a new `RealtimeClient` instance.
 
 #### Methods
 
+- **getExecutorService()**: Returns the `ExecutorService` instance used by the client.
+
+  ```java
+  public ExecutorService getExecutorService();
+  ```
+
 - **connect()**: Connects the client to the WebSocket Messaging Gateway.
 
   ```java

src/main/java/realtime/pubsub/RealtimeClient.java

@@ -115,9 +115,18 @@ public ReplyFunction createReplyFunction(IncomingMessage message) {
         };
     }
 
+    /**
+     * Gets the ExecutorService used by the RealtimeClient. A fixed thread pool (size = number of processors).
+     *
+     * @return The ExecutorService instance.
+     */
+    public ExecutorService getExecutorService() {
+        return executorService;
+    }
+
     private void onAck(Object[] args) {
         if (args.length == 0) return;
-        var message = (IncomingMessage)(args[0]);
+        var message = (IncomingMessage) (args[0]);
         if (message == null) return;
 
         Map<String, Object> data = safeCastToMap(message.getData());
@@ -132,7 +141,7 @@ private void onAck(Object[] args) {
 
     private void onResponse(Object[] args) {
         if (args.length == 0) return;
-        var message = (IncomingMessage)(args[0]);
+        var message = (IncomingMessage) (args[0]);
         if (message == null) return;
 
         String topic = (String) message.getTopic();
@@ -153,7 +162,7 @@ private void onResponse(Object[] args) {
 
     private void onWelcome(Object[] args) {
         if (args.length == 0) return;
-        var message = (IncomingMessage)(args[0]);
+        var message = (IncomingMessage) (args[0]);
         if (message == null) return;
 
         Map<String, Object> data = safeCastToMap(message.getData());
@@ -550,15 +559,7 @@ private String maskUrl(String url) {
         try {
             URI uri = new URI(url);
             String maskedQuery = (uri.getQuery() != null) ? "****" : null;
-            URI maskedUri = new URI(
-                    uri.getScheme(),
-                    uri.getUserInfo(),
-                    uri.getHost(),
-                    uri.getPort(),
-                    uri.getPath(),
-                    maskedQuery,
-                    uri.getFragment()
-            );
+            URI maskedUri = new URI(uri.getScheme(), uri.getUserInfo(), uri.getHost(), uri.getPort(), uri.getPath(), maskedQuery, uri.getFragment());
             return maskedUri.toString();
         } catch (URISyntaxException e) {
             // If masking fails, return a generic message

src/test/java/realtime/pubsub/RealtimeClientTest.java

@@ -57,6 +57,16 @@ protected WebSocketContainer createWebSocketContainer() {
         }).when(containerMock).connectToServer(any(Endpoint.class), any(ClientEndpointConfig.class), any(URI.class));
     }
 
+    @Test
+    public void testGetExecutorService() {
+        // Act
+        ExecutorService executorService = client.getExecutorService();
+
+        // Assert
+        assertNotNull(executorService);
+        assertInstanceOf(ThreadPoolExecutor.class, executorService);
+    }
+
     @Test
     public void testConnect() throws Exception {
         // Act