From fd4dceabba4ebfdb0b89db246717395d06d2d879 Mon Sep 17 00:00:00 2001 From: Adam Schildkraut Date: Thu, 5 Nov 2026 15:54:50 +1100 Subject: [PATCH 1/4] Add a Socket Mode request timing listener --- .../jakarta_socket_mode/SocketModeApp.java | 76 +++++-- .../java/test_locally/SocketModeAppTest.java | 14 ++ .../api/bolt/socket_mode/SocketModeApp.java | 72 +++++-- .../java/test_locally/SocketModeAppTest.java | 25 +++ .../test_locally/SocketModeTimingTest.java | 186 ++++++++++++++++++ .../listener/SocketModeRequestListener.java | 32 +++ .../api/socket_mode/SocketModeClient.java | 7 +- .../api/socket_mode/SocketModeMessage.java | 17 ++ .../socket_mode/SocketModeMessageContext.java | 28 +++ .../impl/SocketModeClientJavaWSImpl.java | 4 +- .../impl/SocketModeClientTyrusImpl.java | 4 +- .../listener/WebSocketMessageListener.java | 7 + .../queue/SocketModeMessageQueue.java | 11 ++ .../queue/impl/TimedMessageQueue.java | 28 +++ .../SocketModeMessageQueueTest.java | 110 +++++++++++ .../JakartaSocketModeClientTyrusImpl.java | 4 +- 16 files changed, 580 insertions(+), 45 deletions(-) create mode 100644 bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java create mode 100644 bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java create mode 100644 slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessage.java create mode 100644 slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java create mode 100644 slack-api-client/src/main/java/com/slack/api/socket_mode/queue/impl/TimedMessageQueue.java create mode 100644 slack-api-client/src/test/java/test_locally/api/socket_mode/SocketModeMessageQueueTest.java 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..6db63feea 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 @@ -7,8 +7,11 @@ import com.slack.api.bolt.jakarta_socket_mode.request.SocketModeRequestParser; import com.slack.api.bolt.request.Request; import com.slack.api.bolt.response.Response; +import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; 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; @@ -18,7 +21,9 @@ import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.function.Function; import java.util.function.Supplier; @@ -30,6 +35,7 @@ public class SocketModeApp { private final Supplier clientFactory; private SocketModeClient client; private final ExecutorService executorService; + private final List requestListeners = new CopyOnWriteArrayList<>(); private static final Function DEFAULT_ERROR_HANDLER = (context) -> { Exception e = context.getException(); @@ -69,25 +75,30 @@ private static void sendSocketModeResponse( } } - private static Supplier buildSocketModeClientFactory( - App app, + private Supplier buildSocketModeClientFactory( String appToken, - Function errorHandler, - ExecutorService executorService + Function errorHandler ) { return () -> { try { 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) { + handle(new SocketModeMessage(message, null)); + } + + @Override + public void handle(SocketModeMessage message) { + if (executorService != null) { + // asynchronous: retain the receive time while this task waits for a worker + executorService.execute(() -> runBoltApp( + message, client, requestParser, errorHandler, gson)); + } else { + runBoltApp(message, client, requestParser, errorHandler, gson); + } } }); return client; @@ -98,17 +109,29 @@ private static Supplier buildSocketModeClientFactory( }; } - private static void runBoltApp( - String message, - App app, + private void runBoltApp( + SocketModeMessage message, SocketModeClient client, SocketModeRequestParser requestParser, Function errorHandler, Gson gson ) { long startMillis = System.currentTimeMillis(); - SocketModeRequest req = requestParser.parse(message); + SocketModeRequest req = requestParser.parse(message.getPayload()); if (req != null) { + if (!requestListeners.isEmpty()) { + Long receivedAtNanos = message.getReceivedAtNanos(); + Long dispatchDelayNanos = receivedAtNanos == null ? null : System.nanoTime() - receivedAtNanos; + SocketModeRequestListener.State state = new SocketModeRequestListener.State( + req.getBoltRequest(), req.getEnvelope().getEnvelopeId(), dispatchDelayNanos); + for (SocketModeRequestListener listener : requestListeners) { + try { + listener.accept(state); + } catch (RuntimeException e) { + log.warn("Failed to notify a Socket Mode request listener", e); + } + } + } try { Response boltResponse = app.run(req.getBoltRequest()); if (boltResponse.getStatusCode() != 200) { @@ -158,7 +181,7 @@ public SocketModeApp( Function errorHandler, App app ) throws IOException { - this(buildSocketModeClientFactory(app, appToken, errorHandler, null), app); + this(appToken, errorHandler, app, null); } public SocketModeApp( @@ -166,7 +189,7 @@ public SocketModeApp( App app, Function errorHandler ) throws IOException { - this(buildSocketModeClientFactory(app, appToken, errorHandler, null), app); + this(appToken, errorHandler, app, null); } public SocketModeApp(Supplier clientFactory, App app) { @@ -181,7 +204,9 @@ private SocketModeApp( App app, ExecutorService executorService ) throws IOException { - this(buildSocketModeClientFactory(app, appToken, errorHandler, executorService), app, executorService); + this.app = app; + this.executorService = executorService; + this.clientFactory = buildSocketModeClientFactory(appToken, errorHandler); } // intentionally private to avoid exposing the ExecutorService initialization @@ -264,6 +289,19 @@ public void close() throws Exception { // Accessors // ------------------------------------------- + /** Observers for requests dispatched by this adapter's built-in Bolt handler. */ + public List getRequestListeners() { + return requestListeners; + } + + public void addRequestListener(SocketModeRequestListener listener) { + requestListeners.add(listener); + } + + public void removeRequestListener(SocketModeRequestListener listener) { + requestListeners.remove(listener); + } + public boolean isClientStopped() { return clientStopped; } 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..c6fd6be6d 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 @@ -6,6 +6,7 @@ import com.slack.api.bolt.AppConfig; import com.slack.api.bolt.jakarta_socket_mode.SocketModeApp; import com.slack.api.bolt.response.Response; +import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; import com.slack.api.model.event.AppMentionEvent; import org.junit.After; import org.junit.Before; @@ -16,6 +17,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 +56,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 +71,16 @@ public void payloadHandling() throws Exception { return ctx.ack(); }); SocketModeApp socketModeApp = new SocketModeApp(VALID_APP_TOKEN, app); + socketModeApp.addRequestListener(new SocketModeRequestListener() { + @Override + public void accept(State state) { + assertNotNull(state.getRequest()); + assertNotNull(state.getEnvelopeId()); + assertNotNull(state.getDispatchDelayNanos()); + assertTrue(state.getDispatchDelayNanos() >= 0); + requestObserved.set(true); + } + }); 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..d99b2ac74 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 @@ -5,9 +5,12 @@ import com.slack.api.bolt.App; import com.slack.api.bolt.request.Request; import com.slack.api.bolt.response.Response; +import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; 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; @@ -17,7 +20,9 @@ import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.function.Function; import java.util.function.Supplier; @@ -29,6 +34,7 @@ public class SocketModeApp { private final Supplier clientFactory; private SocketModeClient client; private final ExecutorService executorService; + private final List requestListeners = new CopyOnWriteArrayList<>(); private static final Function DEFAULT_ERROR_HANDLER = (context) -> { Exception e = context.getException(); @@ -68,26 +74,31 @@ private static void sendSocketModeResponse( } } - private static Supplier buildSocketModeClientFactory( - App app, + private Supplier buildSocketModeClientFactory( String appToken, SocketModeClient.Backend backend, - Function errorHandler, - ExecutorService executorService + Function errorHandler ) { return () -> { try { 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) { + handle(new SocketModeMessage(message, null)); + } + + @Override + public void handle(SocketModeMessage message) { + if (executorService != null) { + // asynchronous: retain the receive time while this task waits for a worker + executorService.execute(() -> runBoltApp( + message, client, requestParser, errorHandler, gson)); + } else { + runBoltApp(message, client, requestParser, errorHandler, gson); + } } }); return client; @@ -98,17 +109,29 @@ private static Supplier buildSocketModeClientFactory( }; } - private static void runBoltApp( - String message, - App app, + private void runBoltApp( + SocketModeMessage message, SocketModeClient client, SocketModeRequestParser requestParser, Function errorHandler, Gson gson ) { long startMillis = System.currentTimeMillis(); - SocketModeRequest req = requestParser.parse(message); + SocketModeRequest req = requestParser.parse(message.getPayload()); if (req != null) { + if (!requestListeners.isEmpty()) { + Long receivedAtNanos = message.getReceivedAtNanos(); + Long dispatchDelayNanos = receivedAtNanos == null ? null : System.nanoTime() - receivedAtNanos; + SocketModeRequestListener.State state = new SocketModeRequestListener.State( + req.getBoltRequest(), req.getEnvelope().getEnvelopeId(), dispatchDelayNanos); + for (SocketModeRequestListener listener : requestListeners) { + try { + listener.accept(state); + } catch (RuntimeException e) { + log.warn("Failed to notify a Socket Mode request listener", e); + } + } + } try { Response boltResponse = app.run(req.getBoltRequest()); if (boltResponse.getStatusCode() != 200) { @@ -228,7 +251,9 @@ private SocketModeApp( App app, ExecutorService executorService ) throws IOException { - this(buildSocketModeClientFactory(app, appToken, backend, errorHandler, executorService), app, executorService); + this.app = app; + this.executorService = executorService; + this.clientFactory = buildSocketModeClientFactory(appToken, backend, errorHandler); } // intentionally private to avoid exposing the ExecutorService initialization @@ -311,6 +336,19 @@ public void close() throws Exception { // Accessors // ------------------------------------------- + /** Observers for requests dispatched by this adapter's built-in Bolt handler. */ + public List getRequestListeners() { + return requestListeners; + } + + public void addRequestListener(SocketModeRequestListener listener) { + requestListeners.add(listener); + } + + public void removeRequestListener(SocketModeRequestListener listener) { + requestListeners.remove(listener); + } + public boolean isClientStopped() { return clientStopped; } 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..434fe2b75 100644 --- a/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java +++ b/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java @@ -5,6 +5,7 @@ import com.slack.api.bolt.App; import com.slack.api.bolt.AppConfig; import com.slack.api.bolt.response.Response; +import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; import com.slack.api.bolt.socket_mode.SocketModeApp; import com.slack.api.model.event.AppMentionEvent; import com.slack.api.socket_mode.SocketModeClient; @@ -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,16 @@ public void payloadHandling() throws Exception { return ctx.ack(); }); SocketModeApp socketModeApp = new SocketModeApp(VALID_APP_TOKEN, app); + socketModeApp.addRequestListener(new SocketModeRequestListener() { + @Override + public void accept(State state) { + assertNotNull(state.getRequest()); + assertNotNull(state.getEnvelopeId()); + assertNotNull(state.getDispatchDelayNanos()); + assertTrue(state.getDispatchDelayNanos() >= 0); + requestObserved.set(true); + } + }); socketModeApp.startAsync(); try { Thread.sleep(3_000L); @@ -110,7 +123,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 +138,16 @@ public void payloadHandling_JavaWebSocket() throws Exception { SocketModeClient.Backend.JavaWebSocket, app ); + socketModeApp.addRequestListener(new SocketModeRequestListener() { + @Override + public void accept(State state) { + assertNotNull(state.getRequest()); + assertNotNull(state.getEnvelopeId()); + assertNotNull(state.getDispatchDelayNanos()); + assertTrue(state.getDispatchDelayNanos() >= 0); + requestObserved.set(true); + } + }); 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..e4f72f066 --- /dev/null +++ b/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java @@ -0,0 +1,186 @@ +package test_locally; + +import com.google.gson.JsonParser; +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.bolt.socket_mode.listener.SocketModeRequestListener; +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 org.junit.Test; +import org.mockito.ArgumentCaptor; + +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.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +public class SocketModeTimingTest { + @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 secondState = 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.socketApp.addRequestListener(new SocketModeRequestListener() { + @Override + public void accept(State state) { + if ("queued-delivery".equals(state.getEnvelopeId())) { + secondState.set(state); + } + } + }); + when(fixture.app.run(any())).thenAnswer(invocation -> { + if (handled.getAndIncrement() == 0) { + handlerHeld.countDown(); + await(releaseHandler); + } else { + secondStarted.set(System.nanoTime()); + secondRequest.set(invocation.getArgument(0)); + secondHandled.countDown(); + } + return Response.ok(); + }); + try { + fixture.socketApp.startAsync(); + fixture.socket.enqueueMessage(command("first-delivery")); + assertTrue(receiverHeld.await(5, TimeUnit.SECONDS)); + long beforeEnqueue = System.nanoTime(); + fixture.socket.enqueueMessage(command("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(secondState.get()); + assertNull(fixture.acknowledgements.poll()); + long handlerReleased = System.nanoTime(); + releaseHandler.countDown(); + assertTrue(secondHandled.await(5, TimeUnit.SECONDS)); + + SocketModeRequestListener.State state = secondState.get(); + assertEquals("queued-delivery", state.getEnvelopeId()); + assertSame(secondRequest.get(), state.getRequest()); + assertNotNull(state.getDispatchDelayNanos()); + assertTrue(state.getDispatchDelayNanos() >= handlerReleased - afterEnqueue); + assertTrue(state.getDispatchDelayNanos() <= secondStarted.get() - beforeEnqueue); + assertEquals("first-delivery", envelopeId(fixture.acknowledgements.poll(5, TimeUnit.SECONDS))); + assertEquals("queued-delivery", envelopeId(fixture.acknowledgements.poll(5, TimeUnit.SECONDS))); + } finally { + releaseReceiver.countDown(); + releaseHandler.countDown(); + } + } + } + + @Test + public void unknownTimingAndListenerFailureDoNotPreventHandlingOrAcknowledgement() throws Exception { + try (Fixture fixture = new Fixture(new ConcurrentLinkedMessageQueue())) { + fixture.socketApp.addRequestListener(new SocketModeRequestListener() { + @Override + public void accept(State state) { + throw new IllegalStateException("Metrics listener unavailable"); + } + }); + BlockingQueue observed = new LinkedBlockingQueue<>(); + fixture.socketApp.addRequestListener(new SocketModeRequestListener() { + @Override + public void accept(State state) { + observed.add(state); + } + }); + fixture.socketApp.startAsync(); + fixture.socket.enqueueMessage(command("legacy-delivery")); + SocketModeRequestListener.State state = observed.poll(5, TimeUnit.SECONDS); + assertNotNull(state); + assertEquals("legacy-delivery", state.getEnvelopeId()); + assertNull(state.getDispatchDelayNanos()); + assertEquals("legacy-delivery", envelopeId(fixture.acknowledgements.poll(5, TimeUnit.SECONDS))); + ArgumentCaptor request = ArgumentCaptor.forClass(Request.class); + verify(fixture.app).run(request.capture()); + assertSame(state.getRequest(), request.getValue()); + } + } + + private static class Fixture implements AutoCloseable { + // Real receive queue, dispatcher, Bolt parser and handler executor. Only network + // connection/send and the application handler are stubbed. + private final Slack slack = spy(Slack.getInstance(new SlackConfig())); + private final SocketModeClient socket; + private final App app = mock(App.class); + private final SocketModeApp socketApp; + private final BlockingQueue acknowledgements = new LinkedBlockingQueue<>(); + + Fixture(SocketModeMessageQueue queue) 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 -> { + acknowledgements.add(invocation.getArgument(0)); + return null; + }).when(socket).sendSocketModeResponse(anyString()); + when(app.slack()).thenReturn(slack); + when(app.config()).thenReturn(AppConfig.builder().slack(slack).build()); + when(app.run(any())).thenReturn(Response.ok()); + socketApp = new SocketModeApp("xapp-test", app, SocketModeClient.Backend.JavaWebSocket, 1); + } + + @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); + } + } + + private static String envelopeId(String response) { + assertNotNull(response); + return JsonParser.parseString(response).getAsJsonObject().get("envelope_id").getAsString(); + } + + private static String command(String envelopeId) { + return "{\"type\":\"slash_commands\",\"envelope_id\":\"" + envelopeId + + "\",\"accepts_response_payload\":true,\"payload\":{\"command\":\"/test\"," + + "\"team_id\":\"T123\",\"user_id\":\"U123\",\"channel_id\":\"C123\"," + + "\"text\":\"\",\"trigger_id\":\"trigger\"}}"; + } +} diff --git a/bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java b/bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java new file mode 100644 index 000000000..76a081ff9 --- /dev/null +++ b/bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java @@ -0,0 +1,32 @@ +package com.slack.api.bolt.socket_mode.listener; + +import com.slack.api.bolt.request.Request; +import lombok.AllArgsConstructor; +import lombok.Data; + +import java.util.function.Consumer; + +/** + * Observes a parsed Socket Mode request immediately before Bolt handles it. + * Listeners run on the handler thread and should return promptly. Runtime exceptions + * from listeners are logged without preventing request handling or acknowledgement. + */ +public abstract class SocketModeRequestListener implements Consumer { + + public abstract void accept(State state); + + @AllArgsConstructor + @Data + public static class State { + private Request request; + private String envelopeId; + + /** + * Monotonic elapsed time from entering the receive queue to starting request handling, + * in nanoseconds. Includes both SDK queues and dispatch/parsing overhead, but excludes + * these listeners, Bolt middleware, the handler and acknowledgement sending. + * Null means the queue or custom transport did not provide timing information. + */ + private Long dispatchDelayNanos; + } +} 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..e4ee5ff89 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); @@ -382,8 +382,9 @@ default void processMessage(String message) throws IOException { return; } + SocketModeMessage receivedMessage = new SocketModeMessage(message, SocketModeMessageContext.receivedAtNanos()); for (WebSocketMessageListener listener : getWebSocketMessageListeners()) { - listener.handle(message); + listener.handle(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..9c3879939 --- /dev/null +++ b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java @@ -0,0 +1,28 @@ +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 RECEIVED_AT_NANOS = new ThreadLocal<>(); + + private SocketModeMessageContext() {} + + static Long receivedAtNanos() { + return RECEIVED_AT_NANOS.get(); + } + + static void process(SocketModeClient client, SocketModeMessage message) throws IOException { + Long previous = RECEIVED_AT_NANOS.get(); + RECEIVED_AT_NANOS.set(message.getReceivedAtNanos()); + try { + client.processMessage(message.getPayload()); + } finally { + if (previous == null) { + RECEIVED_AT_NANOS.remove(); + } else { + RECEIVED_AT_NANOS.set(previous); + } + } + } +} 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..0c9c73d88 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 handle(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..97a12c4e1 --- /dev/null +++ b/slack-api-client/src/test/java/test_locally/api/socket_mode/SocketModeMessageQueueTest.java @@ -0,0 +1,110 @@ +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.Collections; +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 dispatcherPreservesProcessMessageOverridesAndClearsTimingAfterFailure() throws Exception { + 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); + SocketModeClient client = mock(SocketModeClient.class, CALLS_REAL_METHODS); + when(client.getSlack()).thenReturn(slack); + when(client.getMessageQueue()).thenReturn(new TimedMessageQueue()); + AtomicReference received = new AtomicReference<>(); + WebSocketMessageListener listener = new WebSocketMessageListener() { + @Override + public void handle(String message) { + fail("The dispatcher should supply the message metadata"); + } + + @Override + public void handle(SocketModeMessage message) { + received.set(message); + } + }; + when(client.getWebSocketMessageListeners()).thenReturn(Collections.singletonList(listener)); + client.initializeMessageProcessorExecutor(1); + ArgumentCaptor task = ArgumentCaptor.forClass(Runnable.class); + verify(executor).scheduleAtFixedRate(task.capture(), eq(0L), eq(10L), eq(TimeUnit.MILLISECONDS)); + + String payload = "{\"type\":\"hello\"}"; + doThrow(new IOException("Failed overridden handler")).doCallRealMethod().when(client).processMessage(payload); + client.enqueueMessage(payload); + task.getValue().run(); + verify(client).processMessage(payload); + + // A direct dispatch on the same thread must not inherit the failed delivery's time. + client.processMessage(payload); + assertNull(received.get().getReceivedAtNanos()); + client.enqueueMessage(payload); + task.getValue().run(); + assertNotNull(received.get().getReceivedAtNanos()); + client.processMessage(payload); + assertNull(received.get().getReceivedAtNanos()); + } +} 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 From 8958c434974ea51478ea08c3407277953f7d3e23 Mon Sep 17 00:00:00 2001 From: Adam Schildkraut Date: Thu, 5 Nov 2026 16:04:57 +1100 Subject: [PATCH 2/4] test: reuse Socket Mode fixtures and response models --- .../test_locally/SocketModeTimingTest.java | 43 ++++++++++--------- .../java/util/socket_mode/MockSocketMode.java | 6 +-- 2 files changed, 25 insertions(+), 24 deletions(-) diff --git a/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java b/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java index e4f72f066..b5064f282 100644 --- a/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java +++ b/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java @@ -1,6 +1,6 @@ package test_locally; -import com.google.gson.JsonParser; +import com.google.gson.Gson; import com.slack.api.Slack; import com.slack.api.SlackConfig; import com.slack.api.bolt.App; @@ -14,8 +14,12 @@ 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 org.mockito.ArgumentCaptor; +import util.socket_mode.MockSocketMode; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -30,6 +34,8 @@ 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())) { @@ -72,10 +78,10 @@ public void accept(State state) { }); try { fixture.socketApp.startAsync(); - fixture.socket.enqueueMessage(command("first-delivery")); + fixture.enqueueCommand("first-delivery"); assertTrue(receiverHeld.await(5, TimeUnit.SECONDS)); long beforeEnqueue = System.nanoTime(); - fixture.socket.enqueueMessage(command("queued-delivery")); + fixture.enqueueCommand("queued-delivery"); long afterEnqueue = System.nanoTime(); releaseReceiver.countDown(); assertTrue(handlerHeld.await(5, TimeUnit.SECONDS)); @@ -93,8 +99,8 @@ public void accept(State state) { assertNotNull(state.getDispatchDelayNanos()); assertTrue(state.getDispatchDelayNanos() >= handlerReleased - afterEnqueue); assertTrue(state.getDispatchDelayNanos() <= secondStarted.get() - beforeEnqueue); - assertEquals("first-delivery", envelopeId(fixture.acknowledgements.poll(5, TimeUnit.SECONDS))); - assertEquals("queued-delivery", envelopeId(fixture.acknowledgements.poll(5, TimeUnit.SECONDS))); + 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(); @@ -119,12 +125,12 @@ public void accept(State state) { } }); fixture.socketApp.startAsync(); - fixture.socket.enqueueMessage(command("legacy-delivery")); + fixture.enqueueCommand("legacy-delivery"); SocketModeRequestListener.State state = observed.poll(5, TimeUnit.SECONDS); assertNotNull(state); assertEquals("legacy-delivery", state.getEnvelopeId()); assertNull(state.getDispatchDelayNanos()); - assertEquals("legacy-delivery", envelopeId(fixture.acknowledgements.poll(5, TimeUnit.SECONDS))); + assertEquals(new AckResponse("legacy-delivery"), fixture.acknowledgements.poll(5, TimeUnit.SECONDS)); ArgumentCaptor request = ArgumentCaptor.forClass(Request.class); verify(fixture.app).run(request.capture()); assertSame(state.getRequest(), request.getValue()); @@ -138,7 +144,7 @@ private static class Fixture implements AutoCloseable { private final SocketModeClient socket; private final App app = mock(App.class); private final SocketModeApp socketApp; - private final BlockingQueue acknowledgements = new LinkedBlockingQueue<>(); + private final BlockingQueue acknowledgements = new LinkedBlockingQueue<>(); Fixture(SocketModeMessageQueue queue) throws Exception { socket = spy(new SocketModeClientJavaWSImpl( @@ -146,7 +152,8 @@ private static class Fixture implements AutoCloseable { doNothing().when(socket).connectToNewEndpoint(); doReturn(socket).when(slack).socketMode(anyString(), any(SocketModeClient.Backend.class)); doAnswer(invocation -> { - acknowledgements.add(invocation.getArgument(0)); + String response = invocation.getArgument(0); + acknowledgements.add(GSON.fromJson(response, AckResponse.class)); return null; }).when(socket).sendSocketModeResponse(anyString()); when(app.slack()).thenReturn(slack); @@ -155,6 +162,12 @@ private static class Fixture implements AutoCloseable { socketApp = new SocketModeApp("xapp-test", app, SocketModeClient.Backend.JavaWebSocket, 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(); @@ -171,16 +184,4 @@ private static void await(CountDownLatch latch) { throw new IllegalStateException(e); } } - - private static String envelopeId(String response) { - assertNotNull(response); - return JsonParser.parseString(response).getAsJsonObject().get("envelope_id").getAsString(); - } - - private static String command(String envelopeId) { - return "{\"type\":\"slash_commands\",\"envelope_id\":\"" + envelopeId - + "\",\"accepts_response_payload\":true,\"payload\":{\"command\":\"/test\"," - + "\"team_id\":\"T123\",\"user_id\":\"U123\",\"channel_id\":\"C123\"," - + "\"text\":\"\",\"trigger_id\":\"trigger\"}}"; - } } 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 +} From 3d0ab458dd1258e4136369aeb8251518bf19f36a Mon Sep 17 00:00:00 2001 From: Adam Schildkraut Date: Thu, 5 Nov 2026 16:13:49 +1100 Subject: [PATCH 3/4] refactor: expose Socket Mode timing through Bolt requests --- .../jakarta_socket_mode/SocketModeApp.java | 62 +++------- .../java/test_locally/SocketModeAppTest.java | 17 ++- .../api/bolt/socket_mode/SocketModeApp.java | 58 +++------ .../java/test_locally/SocketModeAppTest.java | 33 +++-- .../test_locally/SocketModeTimingTest.java | 116 +++++++++++------- .../com/slack/api/bolt/request/Request.java | 25 ++++ .../listener/SocketModeRequestListener.java | 32 ----- .../api/socket_mode/SocketModeClient.java | 15 ++- .../socket_mode/SocketModeMessageContext.java | 28 ----- .../listener/WebSocketMessageListener.java | 2 +- .../SocketModeMessageQueueTest.java | 25 ++-- 11 files changed, 180 insertions(+), 233 deletions(-) delete mode 100644 bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java delete mode 100644 slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java 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 6db63feea..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 @@ -7,7 +7,6 @@ import com.slack.api.bolt.jakarta_socket_mode.request.SocketModeRequestParser; import com.slack.api.bolt.request.Request; import com.slack.api.bolt.response.Response; -import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; import com.slack.api.jakarta_socket_mode.JakartaSocketModeClientFactory; import com.slack.api.socket_mode.SocketModeClient; import com.slack.api.socket_mode.SocketModeMessage; @@ -21,9 +20,7 @@ import java.io.IOException; import java.util.HashMap; -import java.util.List; import java.util.Map; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.function.Function; import java.util.function.Supplier; @@ -35,7 +32,6 @@ public class SocketModeApp { private final Supplier clientFactory; private SocketModeClient client; private final ExecutorService executorService; - private final List requestListeners = new CopyOnWriteArrayList<>(); private static final Function DEFAULT_ERROR_HANDLER = (context) -> { Exception e = context.getException(); @@ -75,9 +71,11 @@ private static void sendSocketModeResponse( } } - private Supplier buildSocketModeClientFactory( + private static Supplier buildSocketModeClientFactory( + App app, String appToken, - Function errorHandler + Function errorHandler, + ExecutorService executorService ) { return () -> { try { @@ -87,17 +85,17 @@ private Supplier buildSocketModeClientFactory( client.addWebSocketMessageListener(new WebSocketMessageListener() { @Override public void handle(String message) { - handle(new SocketModeMessage(message, null)); + handleWithMetadata(new SocketModeMessage(message, null)); } @Override - public void handle(SocketModeMessage message) { + public void handleWithMetadata(SocketModeMessage message) { if (executorService != null) { - // asynchronous: retain the receive time while this task waits for a worker + // Retain the receive time while this task waits for a worker. executorService.execute(() -> runBoltApp( - message, client, requestParser, errorHandler, gson)); + message, app, client, requestParser, errorHandler, gson)); } else { - runBoltApp(message, client, requestParser, errorHandler, gson); + runBoltApp(message, app, client, requestParser, errorHandler, gson); } } }); @@ -109,8 +107,9 @@ public void handle(SocketModeMessage message) { }; } - private void runBoltApp( + private static void runBoltApp( SocketModeMessage message, + App app, SocketModeClient client, SocketModeRequestParser requestParser, Function errorHandler, @@ -119,21 +118,15 @@ private void runBoltApp( long startMillis = System.currentTimeMillis(); SocketModeRequest req = requestParser.parse(message.getPayload()); if (req != null) { - if (!requestListeners.isEmpty()) { + Request boltRequest = req.getBoltRequest(); + if (boltRequest != null) { + boltRequest.setSocketModeEnvelopeId(req.getEnvelope().getEnvelopeId()); Long receivedAtNanos = message.getReceivedAtNanos(); - Long dispatchDelayNanos = receivedAtNanos == null ? null : System.nanoTime() - receivedAtNanos; - SocketModeRequestListener.State state = new SocketModeRequestListener.State( - req.getBoltRequest(), req.getEnvelope().getEnvelopeId(), dispatchDelayNanos); - for (SocketModeRequestListener listener : requestListeners) { - try { - listener.accept(state); - } catch (RuntimeException e) { - log.warn("Failed to notify a Socket Mode request listener", e); - } - } + 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()); @@ -181,7 +174,7 @@ public SocketModeApp( Function errorHandler, App app ) throws IOException { - this(appToken, errorHandler, app, null); + this(buildSocketModeClientFactory(app, appToken, errorHandler, null), app); } public SocketModeApp( @@ -189,7 +182,7 @@ public SocketModeApp( App app, Function errorHandler ) throws IOException { - this(appToken, errorHandler, app, null); + this(buildSocketModeClientFactory(app, appToken, errorHandler, null), app); } public SocketModeApp(Supplier clientFactory, App app) { @@ -204,9 +197,7 @@ private SocketModeApp( App app, ExecutorService executorService ) throws IOException { - this.app = app; - this.executorService = executorService; - this.clientFactory = buildSocketModeClientFactory(appToken, errorHandler); + this(buildSocketModeClientFactory(app, appToken, errorHandler, executorService), app, executorService); } // intentionally private to avoid exposing the ExecutorService initialization @@ -289,19 +280,6 @@ public void close() throws Exception { // Accessors // ------------------------------------------- - /** Observers for requests dispatched by this adapter's built-in Bolt handler. */ - public List getRequestListeners() { - return requestListeners; - } - - public void addRequestListener(SocketModeRequestListener listener) { - requestListeners.add(listener); - } - - public void removeRequestListener(SocketModeRequestListener listener) { - requestListeners.remove(listener); - } - public boolean isClientStopped() { return clientStopped; } 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 c6fd6be6d..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 @@ -6,7 +6,6 @@ import com.slack.api.bolt.AppConfig; import com.slack.api.bolt.jakarta_socket_mode.SocketModeApp; import com.slack.api.bolt.response.Response; -import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; import com.slack.api.model.event.AppMentionEvent; import org.junit.After; import org.junit.Before; @@ -71,15 +70,13 @@ public void payloadHandling() throws Exception { return ctx.ack(); }); SocketModeApp socketModeApp = new SocketModeApp(VALID_APP_TOKEN, app); - socketModeApp.addRequestListener(new SocketModeRequestListener() { - @Override - public void accept(State state) { - assertNotNull(state.getRequest()); - assertNotNull(state.getEnvelopeId()); - assertNotNull(state.getDispatchDelayNanos()); - assertTrue(state.getDispatchDelayNanos() >= 0); - requestObserved.set(true); - } + 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 { 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 d99b2ac74..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 @@ -5,7 +5,6 @@ import com.slack.api.bolt.App; import com.slack.api.bolt.request.Request; import com.slack.api.bolt.response.Response; -import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; 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; @@ -20,9 +19,7 @@ import java.io.IOException; import java.util.HashMap; -import java.util.List; import java.util.Map; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.function.Function; import java.util.function.Supplier; @@ -34,7 +31,6 @@ public class SocketModeApp { private final Supplier clientFactory; private SocketModeClient client; private final ExecutorService executorService; - private final List requestListeners = new CopyOnWriteArrayList<>(); private static final Function DEFAULT_ERROR_HANDLER = (context) -> { Exception e = context.getException(); @@ -74,10 +70,12 @@ private static void sendSocketModeResponse( } } - private Supplier buildSocketModeClientFactory( + private static Supplier buildSocketModeClientFactory( + App app, String appToken, SocketModeClient.Backend backend, - Function errorHandler + Function errorHandler, + ExecutorService executorService ) { return () -> { try { @@ -87,17 +85,17 @@ private Supplier buildSocketModeClientFactory( client.addWebSocketMessageListener(new WebSocketMessageListener() { @Override public void handle(String message) { - handle(new SocketModeMessage(message, null)); + handleWithMetadata(new SocketModeMessage(message, null)); } @Override - public void handle(SocketModeMessage message) { + public void handleWithMetadata(SocketModeMessage message) { if (executorService != null) { - // asynchronous: retain the receive time while this task waits for a worker + // Retain the receive time while this task waits for a worker. executorService.execute(() -> runBoltApp( - message, client, requestParser, errorHandler, gson)); + message, app, client, requestParser, errorHandler, gson)); } else { - runBoltApp(message, client, requestParser, errorHandler, gson); + runBoltApp(message, app, client, requestParser, errorHandler, gson); } } }); @@ -109,8 +107,9 @@ public void handle(SocketModeMessage message) { }; } - private void runBoltApp( + private static void runBoltApp( SocketModeMessage message, + App app, SocketModeClient client, SocketModeRequestParser requestParser, Function errorHandler, @@ -119,21 +118,15 @@ private void runBoltApp( long startMillis = System.currentTimeMillis(); SocketModeRequest req = requestParser.parse(message.getPayload()); if (req != null) { - if (!requestListeners.isEmpty()) { + Request boltRequest = req.getBoltRequest(); + if (boltRequest != null) { + boltRequest.setSocketModeEnvelopeId(req.getEnvelope().getEnvelopeId()); Long receivedAtNanos = message.getReceivedAtNanos(); - Long dispatchDelayNanos = receivedAtNanos == null ? null : System.nanoTime() - receivedAtNanos; - SocketModeRequestListener.State state = new SocketModeRequestListener.State( - req.getBoltRequest(), req.getEnvelope().getEnvelopeId(), dispatchDelayNanos); - for (SocketModeRequestListener listener : requestListeners) { - try { - listener.accept(state); - } catch (RuntimeException e) { - log.warn("Failed to notify a Socket Mode request listener", e); - } - } + 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()); @@ -251,9 +244,7 @@ private SocketModeApp( App app, ExecutorService executorService ) throws IOException { - this.app = app; - this.executorService = executorService; - this.clientFactory = buildSocketModeClientFactory(appToken, backend, errorHandler); + this(buildSocketModeClientFactory(app, appToken, backend, errorHandler, executorService), app, executorService); } // intentionally private to avoid exposing the ExecutorService initialization @@ -336,19 +327,6 @@ public void close() throws Exception { // Accessors // ------------------------------------------- - /** Observers for requests dispatched by this adapter's built-in Bolt handler. */ - public List getRequestListeners() { - return requestListeners; - } - - public void addRequestListener(SocketModeRequestListener listener) { - requestListeners.add(listener); - } - - public void removeRequestListener(SocketModeRequestListener listener) { - requestListeners.remove(listener); - } - public boolean isClientStopped() { return clientStopped; } 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 434fe2b75..38ed2a36b 100644 --- a/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java +++ b/bolt-socket-mode/src/test/java/test_locally/SocketModeAppTest.java @@ -5,7 +5,6 @@ import com.slack.api.bolt.App; import com.slack.api.bolt.AppConfig; import com.slack.api.bolt.response.Response; -import com.slack.api.bolt.socket_mode.listener.SocketModeRequestListener; import com.slack.api.bolt.socket_mode.SocketModeApp; import com.slack.api.model.event.AppMentionEvent; import com.slack.api.socket_mode.SocketModeClient; @@ -70,15 +69,13 @@ public void payloadHandling() throws Exception { return ctx.ack(); }); SocketModeApp socketModeApp = new SocketModeApp(VALID_APP_TOKEN, app); - socketModeApp.addRequestListener(new SocketModeRequestListener() { - @Override - public void accept(State state) { - assertNotNull(state.getRequest()); - assertNotNull(state.getEnvelopeId()); - assertNotNull(state.getDispatchDelayNanos()); - assertTrue(state.getDispatchDelayNanos() >= 0); - requestObserved.set(true); - } + 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 { @@ -138,15 +135,13 @@ public void payloadHandling_JavaWebSocket() throws Exception { SocketModeClient.Backend.JavaWebSocket, app ); - socketModeApp.addRequestListener(new SocketModeRequestListener() { - @Override - public void accept(State state) { - assertNotNull(state.getRequest()); - assertNotNull(state.getEnvelopeId()); - assertNotNull(state.getDispatchDelayNanos()); - assertTrue(state.getDispatchDelayNanos() >= 0); - requestObserved.set(true); - } + 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 { diff --git a/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java b/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java index b5064f282..763739b1d 100644 --- a/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java +++ b/bolt-socket-mode/src/test/java/test_locally/SocketModeTimingTest.java @@ -8,7 +8,6 @@ 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.bolt.socket_mode.listener.SocketModeRequestListener; import com.slack.api.socket_mode.SocketModeClient; import com.slack.api.socket_mode.impl.SocketModeClientJavaWSImpl; import com.slack.api.socket_mode.queue.SocketModeMessageQueue; @@ -18,16 +17,18 @@ import com.slack.api.socket_mode.response.AckResponse; import com.slack.api.util.json.GsonFactory; import org.junit.Test; -import org.mockito.ArgumentCaptor; 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.*; @@ -46,7 +47,7 @@ public void receiveTimeSurvivesBothQueuesAndTheSdkStillSendsAcknowledgements() t CountDownLatch secondDispatched = new CountDownLatch(1); CountDownLatch secondHandled = new CountDownLatch(1); AtomicReference> secondRequest = new AtomicReference<>(); - AtomicReference secondState = new AtomicReference<>(); + AtomicReference> observedRequest = new AtomicReference<>(); AtomicInteger handled = new AtomicInteger(); AtomicLong secondStarted = new AtomicLong(); fixture.socket.addWebSocketMessageListener(message -> { @@ -57,24 +58,22 @@ public void receiveTimeSurvivesBothQueuesAndTheSdkStillSendsAcknowledgements() t secondDispatched.countDown(); } }); - fixture.socketApp.addRequestListener(new SocketModeRequestListener() { - @Override - public void accept(State state) { - if ("queued-delivery".equals(state.getEnvelopeId())) { - secondState.set(state); - } + fixture.app.use((request, response, chain) -> { + if ("queued-delivery".equals(request.getSocketModeEnvelopeId())) { + observedRequest.set(request); } + return chain.next(request); }); - when(fixture.app.run(any())).thenAnswer(invocation -> { + fixture.app.command("/hi-socket-mode", (request, context) -> { if (handled.getAndIncrement() == 0) { handlerHeld.countDown(); await(releaseHandler); } else { secondStarted.set(System.nanoTime()); - secondRequest.set(invocation.getArgument(0)); + secondRequest.set(request); secondHandled.countDown(); } - return Response.ok(); + return context.ack(); }); try { fixture.socketApp.startAsync(); @@ -87,18 +86,18 @@ public void accept(State state) { assertTrue(handlerHeld.await(5, TimeUnit.SECONDS)); assertTrue(secondDispatched.await(5, TimeUnit.SECONDS)); assertEquals(1, secondHandled.getCount()); - assertNull(secondState.get()); + assertNull(observedRequest.get()); assertNull(fixture.acknowledgements.poll()); long handlerReleased = System.nanoTime(); releaseHandler.countDown(); assertTrue(secondHandled.await(5, TimeUnit.SECONDS)); - SocketModeRequestListener.State state = secondState.get(); - assertEquals("queued-delivery", state.getEnvelopeId()); - assertSame(secondRequest.get(), state.getRequest()); - assertNotNull(state.getDispatchDelayNanos()); - assertTrue(state.getDispatchDelayNanos() >= handlerReleased - afterEnqueue); - assertTrue(state.getDispatchDelayNanos() <= secondStarted.get() - beforeEnqueue); + 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 { @@ -109,44 +108,72 @@ public void accept(State state) { } @Test - public void unknownTimingAndListenerFailureDoNotPreventHandlingOrAcknowledgement() throws Exception { + public void unknownTimingRemainsAvailableToMiddlewareAndTheHandlerStillAcknowledges() throws Exception { try (Fixture fixture = new Fixture(new ConcurrentLinkedMessageQueue())) { - fixture.socketApp.addRequestListener(new SocketModeRequestListener() { - @Override - public void accept(State state) { - throw new IllegalStateException("Metrics listener unavailable"); - } + BlockingQueue> observed = new LinkedBlockingQueue<>(); + AtomicReference> handledRequest = new AtomicReference<>(); + fixture.app.use((request, response, chain) -> { + observed.add(request); + return chain.next(request); }); - BlockingQueue observed = new LinkedBlockingQueue<>(); - fixture.socketApp.addRequestListener(new SocketModeRequestListener() { - @Override - public void accept(State state) { - observed.add(state); - } + fixture.app.command("/hi-socket-mode", (request, context) -> { + handledRequest.set(request); + return context.ack(); }); fixture.socketApp.startAsync(); fixture.enqueueCommand("legacy-delivery"); - SocketModeRequestListener.State state = observed.poll(5, TimeUnit.SECONDS); - assertNotNull(state); - assertEquals("legacy-delivery", state.getEnvelopeId()); - assertNull(state.getDispatchDelayNanos()); + 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)); - ArgumentCaptor request = ArgumentCaptor.forClass(Request.class); - verify(fixture.app).run(request.capture()); - assertSame(state.getRequest(), request.getValue()); + 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 and handler executor. Only network - // connection/send and the application handler are stubbed. + // 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 = mock(App.class); + 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(); @@ -156,10 +183,9 @@ private static class Fixture implements AutoCloseable { acknowledgements.add(GSON.fromJson(response, AckResponse.class)); return null; }).when(socket).sendSocketModeResponse(anyString()); - when(app.slack()).thenReturn(slack); - when(app.config()).thenReturn(AppConfig.builder().slack(slack).build()); - when(app.run(any())).thenReturn(Response.ok()); - socketApp = new SocketModeApp("xapp-test", app, SocketModeClient.Backend.JavaWebSocket, 1); + 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) { 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/bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java b/bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java deleted file mode 100644 index 76a081ff9..000000000 --- a/bolt/src/main/java/com/slack/api/bolt/socket_mode/listener/SocketModeRequestListener.java +++ /dev/null @@ -1,32 +0,0 @@ -package com.slack.api.bolt.socket_mode.listener; - -import com.slack.api.bolt.request.Request; -import lombok.AllArgsConstructor; -import lombok.Data; - -import java.util.function.Consumer; - -/** - * Observes a parsed Socket Mode request immediately before Bolt handles it. - * Listeners run on the handler thread and should return promptly. Runtime exceptions - * from listeners are logged without preventing request handling or acknowledgement. - */ -public abstract class SocketModeRequestListener implements Consumer { - - public abstract void accept(State state); - - @AllArgsConstructor - @Data - public static class State { - private Request request; - private String envelopeId; - - /** - * Monotonic elapsed time from entering the receive queue to starting request handling, - * in nanoseconds. Includes both SDK queues and dispatch/parsing overhead, but excludes - * these listeners, Bolt middleware, the handler and acknowledgement sending. - * Null means the queue or custom transport did not provide timing information. - */ - private Long dispatchDelayNanos; - } -} 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 e4ee5ff89..86058ce2a 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 @@ -163,7 +163,7 @@ default void initializeMessageProcessorExecutor(int concurrency) { try { SocketModeMessage message = getMessageQueue().pollMessage(); if (message != null) { - SocketModeMessageContext.process(this, message); + processMessageWithMetadata(message); } } catch (Exception e) { getLogger().error("Failed to poll a message or run processMessage (error: {})", e.getMessage(), e); @@ -362,7 +362,17 @@ default Gson getGson() { return GSON; } + /** + * Dispatches a raw message without receive timing. Queued deliveries use + * {@link #processMessageWithMetadata(SocketModeMessage)}; override that method to intercept them. + */ default void processMessage(String message) throws IOException { + processMessageWithMetadata(new SocketModeMessage(message, null)); + } + + /** Dispatches a queued message, retaining its receive time for metadata-aware listeners. */ + default void processMessageWithMetadata(SocketModeMessage receivedMessage) throws IOException { + String message = receivedMessage.getPayload(); if (!message.startsWith("{")) { return; } @@ -382,9 +392,8 @@ default void processMessage(String message) throws IOException { return; } - SocketModeMessage receivedMessage = new SocketModeMessage(message, SocketModeMessageContext.receivedAtNanos()); for (WebSocketMessageListener listener : getWebSocketMessageListeners()) { - listener.handle(receivedMessage); + 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/SocketModeMessageContext.java b/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java deleted file mode 100644 index 9c3879939..000000000 --- a/slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java +++ /dev/null @@ -1,28 +0,0 @@ -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 RECEIVED_AT_NANOS = new ThreadLocal<>(); - - private SocketModeMessageContext() {} - - static Long receivedAtNanos() { - return RECEIVED_AT_NANOS.get(); - } - - static void process(SocketModeClient client, SocketModeMessage message) throws IOException { - Long previous = RECEIVED_AT_NANOS.get(); - RECEIVED_AT_NANOS.set(message.getReceivedAtNanos()); - try { - client.processMessage(message.getPayload()); - } finally { - if (previous == null) { - RECEIVED_AT_NANOS.remove(); - } else { - RECEIVED_AT_NANOS.set(previous); - } - } - } -} 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 0c9c73d88..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 @@ -8,7 +8,7 @@ public interface WebSocketMessageListener { void handle(String message); /** Receives queue timing when available; existing string listeners continue to work unchanged. */ - default void handle(SocketModeMessage message) { + default void handleWithMetadata(SocketModeMessage message) { handle(message.getPayload()); } 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 index 97a12c4e1..c039bfe5c 100644 --- 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 @@ -12,8 +12,7 @@ import org.junit.Test; import org.mockito.ArgumentCaptor; -import java.io.IOException; -import java.util.Collections; +import java.util.Arrays; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -64,7 +63,7 @@ public void existingQueuesReportUnknownTimingRatherThanZeroWait() { } @Test - public void dispatcherPreservesProcessMessageOverridesAndClearsTimingAfterFailure() throws Exception { + public void dispatcherPassesMetadataExplicitlyAndSupportsExistingStringListeners() throws Exception { ScheduledExecutorService executor = mock(ScheduledExecutorService.class); ExecutorServiceProvider provider = mock(ExecutorServiceProvider.class); when(provider.createThreadScheduledExecutor(anyString())).thenReturn(executor); @@ -76,6 +75,7 @@ public void dispatcherPreservesProcessMessageOverridesAndClearsTimingAfterFailur when(client.getSlack()).thenReturn(slack); when(client.getMessageQueue()).thenReturn(new TimedMessageQueue()); AtomicReference received = new AtomicReference<>(); + AtomicReference rawMessage = new AtomicReference<>(); WebSocketMessageListener listener = new WebSocketMessageListener() { @Override public void handle(String message) { @@ -83,28 +83,27 @@ public void handle(String message) { } @Override - public void handle(SocketModeMessage message) { + public void handleWithMetadata(SocketModeMessage message) { received.set(message); } }; - when(client.getWebSocketMessageListeners()).thenReturn(Collections.singletonList(listener)); + 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)); String payload = "{\"type\":\"hello\"}"; - doThrow(new IOException("Failed overridden handler")).doCallRealMethod().when(client).processMessage(payload); - client.enqueueMessage(payload); - task.getValue().run(); - verify(client).processMessage(payload); - - // A direct dispatch on the same thread must not inherit the failed delivery's time. - client.processMessage(payload); - assertNull(received.get().getReceivedAtNanos()); client.enqueueMessage(payload); task.getValue().run(); + ArgumentCaptor delivery = ArgumentCaptor.forClass(SocketModeMessage.class); + verify(client).processMessageWithMetadata(delivery.capture()); + assertSame(delivery.getValue(), received.get()); assertNotNull(received.get().getReceivedAtNanos()); + assertEquals(payload, rawMessage.get()); + + // Direct string dispatch has no receive timestamp; no ambient delivery state is needed. client.processMessage(payload); assertNull(received.get().getReceivedAtNanos()); + assertEquals(payload, received.get().getPayload()); } } From 1e4b06523ab9bdc04c1d4288207833ad098b3e19 Mon Sep 17 00:00:00 2001 From: Adam Schildkraut Date: Thu, 5 Nov 2026 16:32:51 +1100 Subject: [PATCH 4/4] fix: preserve Socket Mode dispatch overrides with timing --- .../api/socket_mode/SocketModeClient.java | 13 +- .../socket_mode/SocketModeMessageContext.java | 33 +++++ .../SocketModeMessageQueueTest.java | 122 ++++++++++++------ 3 files changed, 118 insertions(+), 50 deletions(-) create mode 100644 slack-api-client/src/main/java/com/slack/api/socket_mode/SocketModeMessageContext.java 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 86058ce2a..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 @@ -163,7 +163,7 @@ default void initializeMessageProcessorExecutor(int concurrency) { try { SocketModeMessage message = getMessageQueue().pollMessage(); if (message != null) { - processMessageWithMetadata(message); + SocketModeMessageContext.process(this, message); } } catch (Exception e) { getLogger().error("Failed to poll a message or run processMessage (error: {})", e.getMessage(), e); @@ -362,17 +362,8 @@ default Gson getGson() { return GSON; } - /** - * Dispatches a raw message without receive timing. Queued deliveries use - * {@link #processMessageWithMetadata(SocketModeMessage)}; override that method to intercept them. - */ default void processMessage(String message) throws IOException { - processMessageWithMetadata(new SocketModeMessage(message, null)); - } - - /** Dispatches a queued message, retaining its receive time for metadata-aware listeners. */ - default void processMessageWithMetadata(SocketModeMessage receivedMessage) throws IOException { - String message = receivedMessage.getPayload(); + SocketModeMessage receivedMessage = SocketModeMessageContext.take(message); if (!message.startsWith("{")) { return; } 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/test/java/test_locally/api/socket_mode/SocketModeMessageQueueTest.java b/slack-api-client/src/test/java/test_locally/api/socket_mode/SocketModeMessageQueueTest.java index c039bfe5c..53fda6ac4 100644 --- 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 @@ -12,6 +12,7 @@ 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; @@ -63,47 +64,90 @@ public void existingQueuesReportUnknownTimingRatherThanZeroWait() { } @Test - public void dispatcherPassesMetadataExplicitlyAndSupportsExistingStringListeners() throws Exception { - 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); - SocketModeClient client = mock(SocketModeClient.class, CALLS_REAL_METHODS); - when(client.getSlack()).thenReturn(slack); - when(client.getMessageQueue()).thenReturn(new TimedMessageQueue()); - AtomicReference received = new AtomicReference<>(); - AtomicReference rawMessage = new AtomicReference<>(); - 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)); - + 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); - task.getValue().run(); - ArgumentCaptor delivery = ArgumentCaptor.forClass(SocketModeMessage.class); - verify(client).processMessageWithMetadata(delivery.capture()); - assertSame(delivery.getValue(), received.get()); - assertNotNull(received.get().getReceivedAtNanos()); - assertEquals(payload, rawMessage.get()); - - // Direct string dispatch has no receive timestamp; no ambient delivery state is needed. + 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(received.get().getReceivedAtNanos()); - assertEquals(payload, received.get().getPayload()); + 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(); + } } }