feat: chat stream helper

Author: srtaalej

slack-api-client/src/main/java/com/slack/api/methods/AsyncChatStreamHelper.java

@@ -0,0 +1,185 @@
+package com.slack.api.methods;
+
+import com.slack.api.methods.request.chat.ChatAppendStreamRequest;
+import com.slack.api.methods.request.chat.ChatStartStreamRequest;
+import com.slack.api.methods.request.chat.ChatStopStreamRequest;
+import com.slack.api.methods.response.chat.ChatAppendStreamResponse;
+import com.slack.api.methods.response.chat.ChatStartStreamResponse;
+import com.slack.api.methods.response.chat.ChatStopStreamResponse;
+import com.slack.api.model.Message;
+import com.slack.api.model.block.LayoutBlock;
+import lombok.Builder;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Async variant of {@link ChatStreamHelper} for {@link AsyncMethodsClient}.
+ * <p>
+ * This helper buffers markdown text and flushes via chat.startStream / chat.appendStream, then finalizes via
+ * chat.stopStream.
+ * <p>
+ *
+ */
+@Data
+@Slf4j
+@Builder
+public class AsyncChatStreamHelper {
+
+    public enum State {
+        STARTING,
+        IN_PROGRESS,
+        COMPLETED
+    }
+
+    private final AsyncMethodsClient client;
+    private final String channel;
+    private final String threadTs;
+    private final String recipientTeamId;
+    private final String recipientUserId;
+
+    @Builder.Default
+    private final int bufferSize = 100;
+
+    @Builder.Default
+    private StringBuilder buffer = new StringBuilder();
+    @Builder.Default
+    private State state = State.STARTING;
+    private String streamTs;
+
+    /**
+     * Append text to the stream.
+     *
+     * @param markdownText markdown text to append
+     * @return a future that completes with a response if the buffer was flushed; completes with null if buffering
+     */
+    public CompletableFuture<ChatAppendStreamResponse> append(String markdownText) {
+        if (state == State.COMPLETED) {
+            CompletableFuture<ChatAppendStreamResponse> f = new CompletableFuture<>();
+            f.completeExceptionally(new SlackChatStreamException("Cannot append to stream: stream state is " + state));
+            return f;
+        }
+
+        buffer.append(markdownText);
+
+        if (buffer.length() >= bufferSize) {
+            return flushBuffer();
+        }
+
+        if (log.isDebugEnabled()) {
+            log.debug("AsyncChatStream appended to buffer: bufferLength={}, bufferSize={}, channel={}, " +
+                            "recipientTeamId={}, recipientUserId={}, threadTs={}",
+                    buffer.length(), bufferSize, channel, recipientTeamId, recipientUserId, threadTs);
+        }
+        return CompletableFuture.completedFuture(null);
+    }
+
+    public CompletableFuture<ChatStopStreamResponse> stop() {
+        return stop(null, null, null);
+    }
+
+    public CompletableFuture<ChatStopStreamResponse> stop(String markdownText) {
+        return stop(markdownText, null, null);
+    }
+
+    public CompletableFuture<ChatStopStreamResponse> stop(
+            String markdownText,
+            List<LayoutBlock> blocks,
+            Message.Metadata metadata
+    ) {
+        if (state == State.COMPLETED) {
+            CompletableFuture<ChatStopStreamResponse> f = new CompletableFuture<>();
+            f.completeExceptionally(new SlackChatStreamException("Cannot stop stream: stream state is " + state));
+            return f;
+        }
+
+        if (markdownText != null) {
+            buffer.append(markdownText);
+        }
+
+        CompletableFuture<Void> ensureStarted;
+        if (streamTs == null) {
+            ensureStarted = client.chatStartStream(ChatStartStreamRequest.builder()
+                            .channel(channel)
+                            .threadTs(threadTs)
+                            .recipientTeamId(recipientTeamId)
+                            .recipientUserId(recipientUserId)
+                            .build())
+                    .thenApply(startResponse -> {
+                        if (!startResponse.isOk() || startResponse.getTs() == null) {
+                            SlackChatStreamException ex = new SlackChatStreamException(
+                                    "Failed to stop stream: stream not started - " + startResponse.getError());
+                            ex.setStartResponse(startResponse);
+                            throw ex;
+                        }
+                        streamTs = startResponse.getTs();
+                        state = State.IN_PROGRESS;
+                        return null;
+                    });
+        } else {
+            ensureStarted = CompletableFuture.completedFuture(null);
+        }
+
+        return ensureStarted.thenCompose(ignored -> client.chatStopStream(ChatStopStreamRequest.builder()
+                        .channel(channel)
+                        .ts(streamTs)
+                        .markdownText(buffer.toString())
+                        .blocks(blocks)
+                        .metadata(metadata)
+                        .build())
+                .thenApply(resp -> {
+                    state = State.COMPLETED;
+                    return resp;
+                }));
+    }
+
+    private CompletableFuture<ChatAppendStreamResponse> flushBuffer() {
+        if (streamTs == null) {
+            return client.chatStartStream(ChatStartStreamRequest.builder()
+                            .channel(channel)
+                            .threadTs(threadTs)
+                            .recipientTeamId(recipientTeamId)
+                            .recipientUserId(recipientUserId)
+                            .markdownText(buffer.toString())
+                            .build())
+                    .thenApply(startResponse -> {
+                        if (!startResponse.isOk()) {
+                            SlackChatStreamException ex = new SlackChatStreamException(
+                                    "Failed to start stream: " + startResponse.getError());
+                            ex.setStartResponse(startResponse);
+                            throw ex;
+                        }
+                        streamTs = startResponse.getTs();
+                        state = State.IN_PROGRESS;
+                        ChatAppendStreamResponse synth = new ChatAppendStreamResponse();
+                        synth.setOk(startResponse.isOk());
+                        synth.setChannel(startResponse.getChannel());
+                        synth.setTs(startResponse.getTs());
+                        synth.setWarning(startResponse.getWarning());
+                        synth.setError(startResponse.getError());
+                        buffer.setLength(0);
+                        return synth;
+                    });
+        } else {
+            return client.chatAppendStream(ChatAppendStreamRequest.builder()
+                            .channel(channel)
+                            .ts(streamTs)
+                            .markdownText(buffer.toString())
+                            .build())
+                    .thenApply(resp -> {
+                        if (!resp.isOk()) {
+                            SlackChatStreamException ex = new SlackChatStreamException(
+                                    "Failed to append to stream: " + resp.getError());
+                            ex.getAppendResponses().add(resp);
+                            throw ex;
+                        }
+                        buffer.setLength(0);
+                        return resp;
+                    });
+        }
+    }
+}
+
+