diff --git a/bolt-jakarta-socket-mode/src/main/java/com/slack/api/bolt/jakarta_socket_mode/SocketModeApp.java b/bolt-jakarta-socket-mode/src/main/java/com/slack/api/bolt/jakarta_socket_mode/SocketModeApp.java index a1e25ef93..baa2615ab 100644 --- a/bolt-jakarta-socket-mode/src/main/java/com/slack/api/bolt/jakarta_socket_mode/SocketModeApp.java +++ b/bolt-jakarta-socket-mode/src/main/java/com/slack/api/bolt/jakarta_socket_mode/SocketModeApp.java @@ -9,6 +9,8 @@ import com.slack.api.bolt.response.Response; import com.slack.api.jakarta_socket_mode.JakartaSocketModeClientFactory; import com.slack.api.socket_mode.SocketModeClient; +import com.slack.api.socket_mode.SocketModeMessage; +import com.slack.api.socket_mode.listener.WebSocketMessageListener; import com.slack.api.socket_mode.response.AckResponse; import com.slack.api.util.json.GsonFactory; import com.slack.api.util.thread.DaemonThreadExecutorServiceFactory; @@ -80,14 +82,21 @@ private static Supplier buildSocketModeClientFactory( final SocketModeClient client = JakartaSocketModeClientFactory.create(app.slack(), appToken); final SocketModeRequestParser requestParser = new SocketModeRequestParser(app.config()); final Gson gson = GsonFactory.createSnakeCase(app.slack().getConfig()); - client.addWebSocketMessageListener(message -> { - if (executorService != null) { - // asynchronous - executorService.execute(() -> runBoltApp( - message, app, client, requestParser, errorHandler, gson)); - } else { - // synchronous - runBoltApp(message, app, client, requestParser, errorHandler, gson); + client.addWebSocketMessageListener(new WebSocketMessageListener() { + @Override + public void handle(String message) { + handleWithMetadata(new SocketModeMessage(message, null)); + } + + @Override + public void handleWithMetadata(SocketModeMessage message) { + if (executorService != null) { + // Retain the receive time while this task waits for a worker. + executorService.execute(() -> runBoltApp( + message, app, client, requestParser, errorHandler, gson)); + } else { + runBoltApp(message, app, client, requestParser, errorHandler, gson); + } } }); return client; @@ -99,7 +108,7 @@ private static Supplier buildSocketModeClientFactory( } private static void runBoltApp( - String message, + SocketModeMessage message, App app, SocketModeClient client, SocketModeRequestParser requestParser, @@ -107,10 +116,17 @@ private static void runBoltApp( Gson gson ) { long startMillis = System.currentTimeMillis(); - SocketModeRequest req = requestParser.parse(message); + SocketModeRequest req = requestParser.parse(message.getPayload()); if (req != null) { + Request boltRequest = req.getBoltRequest(); + if (boltRequest != null) { + boltRequest.setSocketModeEnvelopeId(req.getEnvelope().getEnvelopeId()); + Long receivedAtNanos = message.getReceivedAtNanos(); + boltRequest.setSocketModeDispatchDelayNanos( + receivedAtNanos == null ? null : System.nanoTime() - receivedAtNanos); + } try { - Response boltResponse = app.run(req.getBoltRequest()); + Response boltResponse = app.run(boltRequest); if (boltResponse.getStatusCode() != 200) { log.warn("Unsuccessful Bolt app execution (status: {}, body: {})", boltResponse.getStatusCode(), boltResponse.getBody()); diff --git a/bolt-jakarta-socket-mode/src/test/java/test_locally/SocketModeAppTest.java b/bolt-jakarta-socket-mode/src/test/java/test_locally/SocketModeAppTest.java index d54383704..0696ee903 100644 --- a/bolt-jakarta-socket-mode/src/test/java/test_locally/SocketModeAppTest.java +++ b/bolt-jakarta-socket-mode/src/test/java/test_locally/SocketModeAppTest.java @@ -16,6 +16,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; public class SocketModeAppTest { @@ -54,7 +55,9 @@ public void payloadHandling() throws Exception { AtomicBoolean commandCalled = new AtomicBoolean(false); AtomicBoolean actionCalled = new AtomicBoolean(false); AtomicBoolean eventCalled = new AtomicBoolean(false); + AtomicBoolean requestObserved = new AtomicBoolean(false); app.command("/hi-socket-mode", (req, ctx) -> { + assertTrue(requestObserved.get()); commandCalled.set(true); return ctx.ack("Hello!"); }); @@ -67,6 +70,14 @@ public void payloadHandling() throws Exception { return ctx.ack(); }); SocketModeApp socketModeApp = new SocketModeApp(VALID_APP_TOKEN, app); + app.use((request, response, chain) -> { + assertTrue(request.isSocketMode()); + assertNotNull(request.getSocketModeEnvelopeId()); + assertNotNull(request.getSocketModeDispatchDelayNanos()); + assertTrue(request.getSocketModeDispatchDelayNanos() >= 0); + requestObserved.set(true); + return chain.next(request); + }); socketModeApp.startAsync(); try { Thread.sleep(3_000L); diff --git a/bolt-socket-mode/src/main/java/com/slack/api/bolt/socket_mode/SocketModeApp.java b/bolt-socket-mode/src/main/java/com/slack/api/bolt/socket_mode/SocketModeApp.java index 422227d98..00303cd43 100644 --- a/bolt-socket-mode/src/main/java/com/slack/api/bolt/socket_mode/SocketModeApp.java +++ b/bolt-socket-mode/src/main/java/com/slack/api/bolt/socket_mode/SocketModeApp.java @@ -8,6 +8,8 @@ import com.slack.api.bolt.socket_mode.request.SocketModeRequest; import com.slack.api.bolt.socket_mode.request.SocketModeRequestParser; import com.slack.api.socket_mode.SocketModeClient; +import com.slack.api.socket_mode.SocketModeMessage; +import com.slack.api.socket_mode.listener.WebSocketMessageListener; import com.slack.api.socket_mode.response.AckResponse; import com.slack.api.util.json.GsonFactory; import com.slack.api.util.thread.DaemonThreadExecutorServiceFactory; @@ -80,14 +82,21 @@ private static Supplier buildSocketModeClientFactory( final SocketModeClient client = app.slack().socketMode(appToken, backend); final SocketModeRequestParser requestParser = new SocketModeRequestParser(app.config()); final Gson gson = GsonFactory.createSnakeCase(app.slack().getConfig()); - client.addWebSocketMessageListener(message -> { - if (executorService != null) { - // asynchronous - executorService.execute(() -> runBoltApp( - message, app, client, requestParser, errorHandler, gson)); - } else { - // synchronous - runBoltApp(message, app, client, requestParser, errorHandler, gson); + client.addWebSocketMessageListener(new WebSocketMessageListener() { + @Override + public void handle(String message) { + handleWithMetadata(new SocketModeMessage(message, null)); + } + + @Override + public void handleWithMetadata(SocketModeMessage message) { + if (executorService != null) { + // Retain the receive time while this task waits for a worker. + executorService.execute(() -> runBoltApp( + message, app, client, requestParser, errorHandler, gson)); + } else { + runBoltApp(message, app, client, requestParser, errorHandler, gson); + } } }); return client; @@ -99,7 +108,7 @@ private static Supplier buildSocketModeClientFactory( } private static void runBoltApp( - String message, + SocketModeMessage message, App app, SocketModeClient client, SocketModeRequestParser requestParser, @@ -107,10 +116,17 @@ private static void runBoltApp( Gson gson ) { long startMillis = System.currentTimeMillis(); - SocketModeRequest req = requestParser.parse(message); + SocketModeRequest req = requestParser.parse(message.getPayload()); if (req != null) { + Request boltRequest = req.getBoltRequest(); + if (boltRequest != null) { + boltRequest.setSocketModeEnvelopeId(req.getEnvelope().getEnvelopeId()); + Long receivedAtNanos = message.getReceivedAtNanos(); + boltRequest.setSocketModeDispatchDelayNanos( + receivedAtNanos == null ? null : System.nanoTime() - receivedAtNanos); + } try { - Response boltResponse = app.run(req.getBoltRequest()); + Response boltResponse = app.run(boltRequest); if (boltResponse.getStatusCode() != 200) { log.warn("Unsuccessful Bolt app execution (status: {}, body: {})", boltResponse.getStatusCode(), boltResponse.getBody()); diff --git a/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java b/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java index d763de1c5..38ed2a36b 100644 --- a/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java +++ b/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java @@ -54,7 +54,9 @@ public void payloadHandling() throws Exception { AtomicBoolean commandCalled = new AtomicBoolean(false); AtomicBoolean actionCalled = new AtomicBoolean(false); AtomicBoolean eventCalled = new AtomicBoolean(false); + AtomicBoolean requestObserved = new AtomicBoolean(false); app.command("/hi-socket-mode", (req, ctx) -> { + assertTrue(requestObserved.get()); commandCalled.set(true); return ctx.ack("Hello!"); }); @@ -67,6 +69,14 @@ public void payloadHandling() throws Exception { return ctx.ack(); }); SocketModeApp socketModeApp = new SocketModeApp(VALID_APP_TOKEN, app); + app.use((request, response, chain) -> { + assertTrue(request.isSocketMode()); + assertNotNull(request.getSocketModeEnvelopeId()); + assertNotNull(request.getSocketModeDispatchDelayNanos()); + assertTrue(request.getSocketModeDispatchDelayNanos() >= 0); + requestObserved.set(true); + return chain.next(request); + }); socketModeApp.startAsync(); try { Thread.sleep(3_000L); @@ -110,7 +120,9 @@ public void payloadHandling_JavaWebSocket() throws Exception { .build()); AtomicBoolean commandCalled = new AtomicBoolean(false); AtomicBoolean actionCalled = new AtomicBoolean(false); + AtomicBoolean requestObserved = new AtomicBoolean(false); app.command("/hi-socket-mode", (req, ctx) -> { + assertTrue(requestObserved.get()); commandCalled.set(true); return ctx.ack("Hello!"); }); @@ -123,6 +135,14 @@ public void payloadHandling_JavaWebSocket() throws Exception { SocketModeClient.Backend.JavaWebSocket, app ); + app.use((request, response, chain) -> { + assertTrue(request.isSocketMode()); + assertNotNull(request.getSocketModeEnvelopeId()); + assertNotNull(request.getSocketModeDispatchDelayNanos()); + assertTrue(request.getSocketModeDispatchDelayNanos() >= 0); + requestObserved.set(true); + return chain.next(request); + }); socketModeApp.startAsync(); try { Thread.sleep(3_000L); diff --git a/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java b/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java new file mode 100644 index 000000000..763739b1d --- /dev/null +++ b/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java @@ -0,0 +1,213 @@ +package test_locally; + +import com.google.gson.Gson; +import com.slack.api.Slack; +import com.slack.api.SlackConfig; +import com.slack.api.bolt.App; +import com.slack.api.bolt.AppConfig; +import com.slack.api.bolt.request.Request; +import com.slack.api.bolt.response.Response; +import com.slack.api.bolt.socket_mode.SocketModeApp; +import com.slack.api.socket_mode.SocketModeClient; +import com.slack.api.socket_mode.impl.SocketModeClientJavaWSImpl; +import com.slack.api.socket_mode.queue.SocketModeMessageQueue; +import com.slack.api.socket_mode.queue.impl.ConcurrentLinkedMessageQueue; +import com.slack.api.socket_mode.queue.impl.TimedMessageQueue; +import com.slack.api.socket_mode.request.SlashCommandsEnvelope; +import com.slack.api.socket_mode.response.AckResponse; +import com.slack.api.util.json.GsonFactory; +import org.junit.Test; +import util.socket_mode.MockSocketMode; + +import java.util.ArrayList; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; + +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +public class SocketModeTimingTest { + private static final Gson GSON = GsonFactory.createSnakeCase(); + + @Test + public void receiveTimeSurvivesBothQueuesAndTheSdkStillSendsAcknowledgements() throws Exception { + try (Fixture fixture = new Fixture(new TimedMessageQueue())) { + CountDownLatch receiverHeld = new CountDownLatch(1); + CountDownLatch releaseReceiver = new CountDownLatch(1); + CountDownLatch handlerHeld = new CountDownLatch(1); + CountDownLatch releaseHandler = new CountDownLatch(1); + CountDownLatch secondDispatched = new CountDownLatch(1); + CountDownLatch secondHandled = new CountDownLatch(1); + AtomicReference> secondRequest = new AtomicReference<>(); + AtomicReference> observedRequest = new AtomicReference<>(); + AtomicInteger handled = new AtomicInteger(); + AtomicLong secondStarted = new AtomicLong(); + fixture.socket.addWebSocketMessageListener(message -> { + if (message.contains("first-delivery")) { + receiverHeld.countDown(); + await(releaseReceiver); + } else if (message.contains("queued-delivery")) { + secondDispatched.countDown(); + } + }); + fixture.app.use((request, response, chain) -> { + if ("queued-delivery".equals(request.getSocketModeEnvelopeId())) { + observedRequest.set(request); + } + return chain.next(request); + }); + fixture.app.command("/hi-socket-mode", (request, context) -> { + if (handled.getAndIncrement() == 0) { + handlerHeld.countDown(); + await(releaseHandler); + } else { + secondStarted.set(System.nanoTime()); + secondRequest.set(request); + secondHandled.countDown(); + } + return context.ack(); + }); + try { + fixture.socketApp.startAsync(); + fixture.enqueueCommand("first-delivery"); + assertTrue(receiverHeld.await(5, TimeUnit.SECONDS)); + long beforeEnqueue = System.nanoTime(); + fixture.enqueueCommand("queued-delivery"); + long afterEnqueue = System.nanoTime(); + releaseReceiver.countDown(); + assertTrue(handlerHeld.await(5, TimeUnit.SECONDS)); + assertTrue(secondDispatched.await(5, TimeUnit.SECONDS)); + assertEquals(1, secondHandled.getCount()); + assertNull(observedRequest.get()); + assertNull(fixture.acknowledgements.poll()); + long handlerReleased = System.nanoTime(); + releaseHandler.countDown(); + assertTrue(secondHandled.await(5, TimeUnit.SECONDS)); + + Request request = observedRequest.get(); + assertEquals("queued-delivery", request.getSocketModeEnvelopeId()); + assertSame(secondRequest.get(), request); + assertNotNull(request.getSocketModeDispatchDelayNanos()); + assertTrue(request.getSocketModeDispatchDelayNanos() >= handlerReleased - afterEnqueue); + assertTrue(request.getSocketModeDispatchDelayNanos() <= secondStarted.get() - beforeEnqueue); + assertEquals(new AckResponse("first-delivery"), fixture.acknowledgements.poll(5, TimeUnit.SECONDS)); + assertEquals(new AckResponse("queued-delivery"), fixture.acknowledgements.poll(5, TimeUnit.SECONDS)); + } finally { + releaseReceiver.countDown(); + releaseHandler.countDown(); + } + } + } + + @Test + public void unknownTimingRemainsAvailableToMiddlewareAndTheHandlerStillAcknowledges() throws Exception { + try (Fixture fixture = new Fixture(new ConcurrentLinkedMessageQueue())) { + BlockingQueue> observed = new LinkedBlockingQueue<>(); + AtomicReference> handledRequest = new AtomicReference<>(); + fixture.app.use((request, response, chain) -> { + observed.add(request); + return chain.next(request); + }); + fixture.app.command("/hi-socket-mode", (request, context) -> { + handledRequest.set(request); + return context.ack(); + }); + fixture.socketApp.startAsync(); + fixture.enqueueCommand("legacy-delivery"); + Request request = observed.poll(5, TimeUnit.SECONDS); + assertNotNull(request); + assertEquals("legacy-delivery", request.getSocketModeEnvelopeId()); + assertNull(request.getSocketModeDispatchDelayNanos()); + assertEquals(new AckResponse("legacy-delivery"), fixture.acknowledgements.poll(5, TimeUnit.SECONDS)); + assertSame(request, handledRequest.get()); + } + } + + @Test + public void middlewareFailureUsesTheConfiguredErrorHandler() throws Exception { + AtomicReference error = new AtomicReference<>(); + try (Fixture fixture = new Fixture(new TimedMessageQueue(), context -> { + error.set(context); + return Response.ok(); + })) { + AtomicReference> observed = new AtomicReference<>(); + AtomicBoolean handled = new AtomicBoolean(); + fixture.app.use((request, response, chain) -> { + observed.set(request); + throw new IllegalStateException("Middleware failed"); + }); + fixture.app.command("/hi-socket-mode", (request, context) -> { + handled.set(true); + return context.ack(); + }); + fixture.socketApp.startAsync(); + fixture.enqueueCommand("failed-delivery"); + + assertEquals(new AckResponse("failed-delivery"), fixture.acknowledgements.poll(5, TimeUnit.SECONDS)); + assertEquals("failed-delivery", observed.get().getSocketModeEnvelopeId()); + assertNotNull(observed.get().getSocketModeDispatchDelayNanos()); + assertSame(observed.get(), error.get().getRequest()); + assertEquals("Middleware failed", error.get().getException().getMessage()); + assertFalse(handled.get()); + } + } + + private static class Fixture implements AutoCloseable { + // Real receive queue, dispatcher, Bolt parser, middleware and handler executor. + // Only network connection/send are stubbed. + private final Slack slack = spy(Slack.getInstance(new SlackConfig())); + private final SocketModeClient socket; + private final App app = new App(AppConfig.builder().slack(slack).build(), new ArrayList<>()); + private final SocketModeApp socketApp; + private final BlockingQueue acknowledgements = new LinkedBlockingQueue<>(); + + Fixture(SocketModeMessageQueue queue) throws Exception { + this(queue, null); + } + + Fixture(SocketModeMessageQueue queue, Function errorHandler) throws Exception { + socket = spy(new SocketModeClientJavaWSImpl( + slack, "xapp-test", "ws://127.0.0.1:1", 1, queue, false, false, 5000)); + doNothing().when(socket).connectToNewEndpoint(); + doReturn(socket).when(slack).socketMode(anyString(), any(SocketModeClient.Backend.class)); + doAnswer(invocation -> { + String response = invocation.getArgument(0); + acknowledgements.add(GSON.fromJson(response, AckResponse.class)); + return null; + }).when(socket).sendSocketModeResponse(anyString()); + socketApp = errorHandler == null + ? new SocketModeApp("xapp-test", app, SocketModeClient.Backend.JavaWebSocket, 1) + : new SocketModeApp("xapp-test", SocketModeClient.Backend.JavaWebSocket, errorHandler, app, 1); + } + + void enqueueCommand(String envelopeId) { + SlashCommandsEnvelope envelope = GSON.fromJson(MockSocketMode.COMMAND_ENVELOPE, SlashCommandsEnvelope.class); + envelope.setEnvelopeId(envelopeId); + socket.enqueueMessage(GSON.toJson(envelope)); + } + + @Override + public void close() throws Exception { + socketApp.close(); + socket.close(); + slack.close(); + } + } + + private static void await(CountDownLatch latch) { + try { + assertTrue(latch.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } +} diff --git a/bolt-socket-mode/src/test/java/util/socket_mode/MockSocketMode.java b/bolt-socket-mode/src/test/java/util/socket_mode/MockSocketMode.java index 78f049477..6b09e3ce1 100644 --- a/bolt-socket-mode/src/test/java/util/socket_mode/MockSocketMode.java +++ b/bolt-socket-mode/src/test/java/util/socket_mode/MockSocketMode.java @@ -221,7 +221,7 @@ public void onWebSocketError(Throwable cause) { " \"retry_reason\": \"timeout\"\n" + "}\n"; - String commandEnvelope = "{\n" + + public static final String COMMAND_ENVELOPE = "{\n" + " \"envelope_id\": \"xxx-11-222-yyy-zzz\",\n" + " \"payload\": {\n" + " \"token\": \"fixed-value\",\n" + @@ -244,7 +244,7 @@ public void onWebSocketError(Throwable cause) { List envelopes = Arrays.asList( interactiveEnvelope, eventsEnvelope, - commandEnvelope + COMMAND_ENVELOPE ); private SecureRandom random = new SecureRandom(); @@ -252,4 +252,4 @@ public void onWebSocketError(Throwable cause) { private String getRandomEnvelope() { return envelopes.get(random.nextInt(envelopes.size())); } -} \ No newline at end of file +} diff --git a/bolt/src/main/java/com/slack/api/bolt/request/Request.java b/bolt/src/main/java/com/slack/api/bolt/request/Request.java index cb3223c36..b1832039d 100644 --- a/bolt/src/main/java/com/slack/api/bolt/request/Request.java +++ b/bolt/src/main/java/com/slack/api/bolt/request/Request.java @@ -105,4 +105,29 @@ public void setSocketMode(boolean socketMode) { this.socketMode = socketMode; } + private String socketModeEnvelopeId; + private Long socketModeDispatchDelayNanos; + + /** The delivery's envelope ID, or null for requests received outside the Socket Mode adapter. */ + public String getSocketModeEnvelopeId() { + return socketModeEnvelopeId; + } + + public void setSocketModeEnvelopeId(String socketModeEnvelopeId) { + this.socketModeEnvelopeId = socketModeEnvelopeId; + } + + /** + * Monotonic elapsed time from entering the Socket Mode receive queue to dispatching this request, + * in nanoseconds. Includes both SDK queues and parsing, but excludes Bolt middleware, handlers + * and acknowledgement sending. Null means receive timing was unavailable, including for HTTP requests. + */ + public Long getSocketModeDispatchDelayNanos() { + return socketModeDispatchDelayNanos; + } + + public void setSocketModeDispatchDelayNanos(Long socketModeDispatchDelayNanos) { + this.socketModeDispatchDelayNanos = socketModeDispatchDelayNanos; + } + } diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeClient.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeClient.java index 140d5d962..b90d817cc 100644 --- a/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeClient.java +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeClient.java @@ -161,9 +161,9 @@ default void initializeMessageProcessorExecutor(int concurrency) { int num = i; messageProcessorExecutor.scheduleAtFixedRate(() -> { try { - String message = getMessageQueue().poll(); + SocketModeMessage message = getMessageQueue().pollMessage(); if (message != null) { - processMessage(message); + SocketModeMessageContext.process(this, message); } } catch (Exception e) { getLogger().error("Failed to poll a message or run processMessage (error: {})", e.getMessage(), e); @@ -363,6 +363,7 @@ default Gson getGson() { } default void processMessage(String message) throws IOException { + SocketModeMessage receivedMessage = SocketModeMessageContext.take(message); if (!message.startsWith("{")) { return; } @@ -383,7 +384,7 @@ default void processMessage(String message) throws IOException { } for (WebSocketMessageListener listener : getWebSocketMessageListeners()) { - listener.handle(message); + listener.handleWithMetadata(receivedMessage); } if (type.equals(EventsApiEnvelope.TYPE)) { for (EnvelopeListener listener : getEventsApiEnvelopeListeners()) { diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessage.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessage.java new file mode 100644 index 000000000..495e0c1a2 --- /dev/null +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessage.java @@ -0,0 +1,17 @@ +package com.slack.api.socket_mode; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** A received message and, when known, its local monotonic receive time. */ +@Getter +@RequiredArgsConstructor +public class SocketModeMessage { + private final String payload; + + /** + * System.nanoTime() when the message entered the client's queue, or null for + * custom queues and direct dispatch without timing information. This is not a wall-clock timestamp. + */ + private final Long receivedAtNanos; +} diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java new file mode 100644 index 000000000..297c110ad --- /dev/null +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java @@ -0,0 +1,33 @@ +package com.slack.api.socket_mode; + +import java.io.IOException; + +/** Carries queue metadata through the existing, overridable processMessage(String) method. */ +final class SocketModeMessageContext { + private static final ThreadLocal CURRENT = new ThreadLocal<>(); + + private SocketModeMessageContext() {} + + static void process(SocketModeClient client, SocketModeMessage message) throws IOException { + SocketModeMessage previous = CURRENT.get(); + CURRENT.set(message); + try { + client.processMessage(message.getPayload()); + } finally { + if (previous == null) { + CURRENT.remove(); + } else { + CURRENT.set(previous); + } + } + } + + static SocketModeMessage take(String payload) { + SocketModeMessage message = CURRENT.get(); + // Consume before invoking listeners so a nested direct dispatch cannot inherit this timing. + CURRENT.remove(); + return message != null && message.getPayload().equals(payload) + ? message + : new SocketModeMessage(payload, null); + } +} diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientJavaWSImpl.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientJavaWSImpl.java index 694e6ed3c..19167dd7a 100644 --- a/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientJavaWSImpl.java +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientJavaWSImpl.java @@ -10,7 +10,7 @@ import com.slack.api.socket_mode.listener.WebSocketErrorListener; import com.slack.api.socket_mode.listener.WebSocketMessageListener; import com.slack.api.socket_mode.queue.SocketModeMessageQueue; -import com.slack.api.socket_mode.queue.impl.ConcurrentLinkedMessageQueue; +import com.slack.api.socket_mode.queue.impl.TimedMessageQueue; import com.slack.api.socket_mode.request.EventsApiEnvelope; import com.slack.api.socket_mode.request.InteractiveEnvelope; import com.slack.api.socket_mode.request.SlashCommandsEnvelope; @@ -84,7 +84,7 @@ public SocketModeClientJavaWSImpl( appToken, wssUrl, concurrency, - new ConcurrentLinkedMessageQueue(), + new TimedMessageQueue(), true, true, DEFAULT_SESSION_MONITOR_INTERVAL_MILLISECONDS diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientTyrusImpl.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientTyrusImpl.java index 680d3ad2d..e900d8f79 100644 --- a/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientTyrusImpl.java +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/impl/SocketModeClientTyrusImpl.java @@ -9,7 +9,7 @@ import com.slack.api.socket_mode.listener.WebSocketErrorListener; import com.slack.api.socket_mode.listener.WebSocketMessageListener; import com.slack.api.socket_mode.queue.SocketModeMessageQueue; -import com.slack.api.socket_mode.queue.impl.ConcurrentLinkedMessageQueue; +import com.slack.api.socket_mode.queue.impl.TimedMessageQueue; import com.slack.api.socket_mode.request.EventsApiEnvelope; import com.slack.api.socket_mode.request.InteractiveEnvelope; import com.slack.api.socket_mode.request.SlashCommandsEnvelope; @@ -95,7 +95,7 @@ public SocketModeClientTyrusImpl( appToken, wssUrl, concurrency, - new ConcurrentLinkedMessageQueue(), + new TimedMessageQueue(), true, true, DEFAULT_SESSION_MONITOR_INTERVAL_MILLISECONDS diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/listener/WebSocketMessageListener.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/listener/WebSocketMessageListener.java index 54fefe72d..d1db381ca 100644 --- a/slack-api-client/src/main/java/com/slack/api/socket_mode/listener/WebSocketMessageListener.java +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/listener/WebSocketMessageListener.java @@ -1,8 +1,15 @@ package com.slack.api.socket_mode.listener; +import com.slack.api.socket_mode.SocketModeMessage; + @FunctionalInterface public interface WebSocketMessageListener { void handle(String message); + /** Receives queue timing when available; existing string listeners continue to work unchanged. */ + default void handleWithMetadata(SocketModeMessage message) { + handle(message.getPayload()); + } + } diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/queue/SocketModeMessageQueue.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/queue/SocketModeMessageQueue.java index 99aaf4a1a..dc5da313c 100644 --- a/slack-api-client/src/main/java/com/slack/api/socket_mode/queue/SocketModeMessageQueue.java +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/queue/SocketModeMessageQueue.java @@ -1,5 +1,7 @@ package com.slack.api.socket_mode.queue; +import com.slack.api.socket_mode.SocketModeMessage; + /** * Abstraction of a message queue that provides * a way to handle multiple messages at the same time in SocketModeClient. @@ -20,4 +22,13 @@ public interface SocketModeMessageQueue { */ String poll(); + /** + * Pops a message with its receive time when the queue records one. + * Existing custom queues remain supported and report an unknown receive time. + */ + default SocketModeMessage pollMessage() { + String payload = poll(); + return payload == null ? null : new SocketModeMessage(payload, null); + } + } diff --git a/slack-api-client/src/main/java/com/slack/api/socket_mode/queue/impl/TimedMessageQueue.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/queue/impl/TimedMessageQueue.java new file mode 100644 index 000000000..5eae4b36e --- /dev/null +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/queue/impl/TimedMessageQueue.java @@ -0,0 +1,28 @@ +package com.slack.api.socket_mode.queue.impl; + +import com.slack.api.socket_mode.SocketModeMessage; +import com.slack.api.socket_mode.queue.SocketModeMessageQueue; + +import java.util.Objects; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** The default in-memory queue, retaining the receive time with each delivery. */ +public class TimedMessageQueue implements SocketModeMessageQueue { + private final ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue<>(); + + @Override + public void add(String message) { + messages.add(new SocketModeMessage(Objects.requireNonNull(message), System.nanoTime())); + } + + @Override + public String poll() { + SocketModeMessage message = pollMessage(); + return message == null ? null : message.getPayload(); + } + + @Override + public SocketModeMessage pollMessage() { + return messages.poll(); + } +} diff --git a/slack-api-client/src/test/java/test_locally/api/socket_mode/SocketModeMessageQueueTest.java b/slack-api-client/src/test/java/test_locally/api/socket_mode/SocketModeMessageQueueTest.java new file mode 100644 index 000000000..53fda6ac4 --- /dev/null +++ b/slack-api-client/src/test/java/test_locally/api/socket_mode/SocketModeMessageQueueTest.java @@ -0,0 +1,153 @@ +package test_locally.api.socket_mode; + +import com.slack.api.Slack; +import com.slack.api.SlackConfig; +import com.slack.api.socket_mode.SocketModeClient; +import com.slack.api.socket_mode.SocketModeMessage; +import com.slack.api.socket_mode.listener.WebSocketMessageListener; +import com.slack.api.socket_mode.queue.SocketModeMessageQueue; +import com.slack.api.socket_mode.queue.impl.ConcurrentLinkedMessageQueue; +import com.slack.api.socket_mode.queue.impl.TimedMessageQueue; +import com.slack.api.util.thread.ExecutorServiceProvider; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +import java.io.IOException; +import java.util.Arrays; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +public class SocketModeMessageQueueTest { + @Test + public void receivesEachDeliverySeparatelyAndPreservesFifo() { + SocketModeMessageQueue queue = new TimedMessageQueue(); + long before = System.nanoTime(); + queue.add("same payload"); + long between = System.nanoTime(); + queue.add("same payload"); + long after = System.nanoTime(); + + SocketModeMessage first = queue.pollMessage(); + SocketModeMessage second = queue.pollMessage(); + assertEquals("same payload", first.getPayload()); + assertEquals("same payload", second.getPayload()); + assertTrue(first.getReceivedAtNanos() >= before); + assertTrue(first.getReceivedAtNanos() <= between); + assertTrue(second.getReceivedAtNanos() >= between); + assertTrue(second.getReceivedAtNanos() <= after); + assertNull(queue.pollMessage()); + } + + @Test + public void existingStringPollRemainsSupported() { + SocketModeMessageQueue queue = new TimedMessageQueue(); + queue.add("first"); + queue.add("second"); + assertEquals("first", queue.poll()); + assertEquals("second", queue.pollMessage().getPayload()); + assertNull(queue.poll()); + } + + @Test + public void existingQueuesReportUnknownTimingRatherThanZeroWait() { + SocketModeMessageQueue queue = new ConcurrentLinkedMessageQueue(); + queue.add("legacy"); + SocketModeMessage message = queue.pollMessage(); + assertEquals("legacy", message.getPayload()); + assertNull(message.getReceivedAtNanos()); + assertNull(queue.pollMessage()); + } + + @Test + public void dispatcherPreservesOverridesAndClearsTimingAfterFailure() throws Exception { + Dispatcher dispatcher = new Dispatcher(); + SocketModeClient client = dispatcher.client; + String payload = "{\"type\":\"hello\"}"; + doThrow(new IOException("Failed overridden handler")).doCallRealMethod().when(client).processMessage(payload); + client.enqueueMessage(payload); + dispatcher.poll.run(); + verify(client).processMessage(payload); + assertNull(dispatcher.received.get()); + assertNull(dispatcher.rawMessage.get()); + + // A direct dispatch must not inherit the failed override's receive time. + client.processMessage(payload); + assertNull(dispatcher.received.get().getReceivedAtNanos()); + assertEquals(payload, dispatcher.rawMessage.get()); + + // An override delegating to the default implementation retains queued timing. + client.enqueueMessage(payload); + dispatcher.poll.run(); + verify(client, times(3)).processMessage(payload); + assertNotNull(dispatcher.received.get().getReceivedAtNanos()); + assertEquals(payload, dispatcher.rawMessage.get()); + + // Timing is also cleared after a successful dispatch. + client.processMessage(payload); + assertNull(dispatcher.received.get().getReceivedAtNanos()); + assertEquals(payload, dispatcher.received.get().getPayload()); + } + + @Test + public void nestedDirectDispatchDoesNotInheritQueuedTiming() throws Exception { + Dispatcher dispatcher = new Dispatcher(); + String payload = "{\"type\":\"hello\"}"; + String nestedPayload = "{\"type\":\"hello\",\"nested\":true}"; + AtomicReference queued = new AtomicReference<>(); + doAnswer(invocation -> { + invocation.callRealMethod(); + queued.set(dispatcher.received.get()); + // The override is still running, but this is a separate direct delivery. + dispatcher.client.processMessage(nestedPayload); + return null; + }).when(dispatcher.client).processMessage(payload); + + dispatcher.client.enqueueMessage(payload); + dispatcher.poll.run(); + + assertNotNull(queued.get().getReceivedAtNanos()); + assertEquals(payload, queued.get().getPayload()); + assertEquals(nestedPayload, dispatcher.received.get().getPayload()); + assertNull(dispatcher.received.get().getReceivedAtNanos()); + } + + private static class Dispatcher { + private final SocketModeClient client = mock(SocketModeClient.class, CALLS_REAL_METHODS); + private final AtomicReference received = new AtomicReference<>(); + private final AtomicReference rawMessage = new AtomicReference<>(); + private final Runnable poll; + + Dispatcher() { + ScheduledExecutorService executor = mock(ScheduledExecutorService.class); + ExecutorServiceProvider provider = mock(ExecutorServiceProvider.class); + when(provider.createThreadScheduledExecutor(anyString())).thenReturn(executor); + SlackConfig config = new SlackConfig(); + config.setExecutorServiceProvider(provider); + Slack slack = mock(Slack.class); + when(slack.getConfig()).thenReturn(config); + when(client.getSlack()).thenReturn(slack); + when(client.getMessageQueue()).thenReturn(new TimedMessageQueue()); + WebSocketMessageListener listener = new WebSocketMessageListener() { + @Override + public void handle(String message) { + fail("The dispatcher should supply the message metadata"); + } + + @Override + public void handleWithMetadata(SocketModeMessage message) { + received.set(message); + } + }; + when(client.getWebSocketMessageListeners()).thenReturn(Arrays.asList(listener, rawMessage::set)); + client.initializeMessageProcessorExecutor(1); + ArgumentCaptor task = ArgumentCaptor.forClass(Runnable.class); + verify(executor).scheduleAtFixedRate(task.capture(), eq(0L), eq(10L), eq(TimeUnit.MILLISECONDS)); + poll = task.getValue(); + } + } +} diff --git a/slack-jakarta-socket-mode-client/src/main/java/com/slack/api/jakarta_socket_mode/impl/JakartaSocketModeClientTyrusImpl.java b/slack-jakarta-socket-mode-client/src/main/java/com/slack/api/jakarta_socket_mode/impl/JakartaSocketModeClientTyrusImpl.java index 8586de5f8..e9ccf6393 100644 --- a/slack-jakarta-socket-mode-client/src/main/java/com/slack/api/jakarta_socket_mode/impl/JakartaSocketModeClientTyrusImpl.java +++ b/slack-jakarta-socket-mode-client/src/main/java/com/slack/api/jakarta_socket_mode/impl/JakartaSocketModeClientTyrusImpl.java @@ -9,7 +9,7 @@ import com.slack.api.socket_mode.listener.WebSocketErrorListener; import com.slack.api.socket_mode.listener.WebSocketMessageListener; import com.slack.api.socket_mode.queue.SocketModeMessageQueue; -import com.slack.api.socket_mode.queue.impl.ConcurrentLinkedMessageQueue; +import com.slack.api.socket_mode.queue.impl.TimedMessageQueue; import com.slack.api.socket_mode.request.EventsApiEnvelope; import com.slack.api.socket_mode.request.InteractiveEnvelope; import com.slack.api.socket_mode.request.SlashCommandsEnvelope; @@ -95,7 +95,7 @@ public JakartaSocketModeClientTyrusImpl( appToken, wssUrl, concurrency, - new ConcurrentLinkedMessageQueue(), + new TimedMessageQueue(), true, true, DEFAULT_SESSION_MONITOR_INTERVAL_MILLISECONDS