Gateway: accept HTTP request for producing messages

langstream-api-gateway-auth/langstream-google-api-gateway-auth/src/main/java/ai/langstream/apigateway/auth/impl/google/GoogleAuthenticationProvider.java

@@ -59,7 +59,12 @@ public void initialize(Map<String, Object> configuration) {
     @Override
     public GatewayAuthenticationResult authenticate(GatewayRequestContext context) {
         try {
-            GoogleIdToken idToken = verifier.verify(context.credentials());
+            final String credentials = context.credentials();
+            if (credentials == null) {
+                return GatewayAuthenticationResult.authenticationFailed(
+                        "Credentials not provided.");
+            }
+            GoogleIdToken idToken = verifier.verify(credentials);
             if (idToken != null) {
                 final GoogleIdToken.Payload payload = idToken.getPayload();
                 Map<String, String> result = new HashMap<>();
@@ -69,7 +74,7 @@ public GatewayAuthenticationResult authenticate(GatewayRequestContext context) {
                 result.put(FIELD_LOCALE, (String) payload.get("locale"));
                 return GatewayAuthenticationResult.authenticationSuccessful(result);
             } else {
-                return GatewayAuthenticationResult.authenticationFailed("Invalid token.");
+                return GatewayAuthenticationResult.authenticationFailed("Invalid credentials.");
             }
         } catch (Exception e) {
             throw new RuntimeException(e);

langstream-api-gateway-auth/langstream-http-api-gateway-auth/src/main/java/ai/langstream/apigateway/auth/impl/jwt/admin/HttpAuthenticationProvider.java

@@ -64,7 +64,8 @@ public GatewayAuthenticationResult authenticate(GatewayRequestContext context) {
         final HttpRequest.Builder builder = HttpRequest.newBuilder().uri(URI.create(url));
 
         httpConfiguration.getHeaders().forEach(builder::header);
-        builder.header("Authorization", "Bearer " + context.credentials());
+        final String credentials = context.credentials();
+        builder.header("Authorization", "Bearer " + (credentials == null ? "" : credentials));
         final HttpRequest request = builder.GET().build();
 
         final HttpResponse<Void> response;

langstream-api-gateway-auth/langstream-jwt-api-gateway-auth/src/main/java/ai/langstream/apigateway/auth/impl/jwt/admin/JwtAuthenticationProvider.java

@@ -66,7 +66,8 @@ public void initialize(Map<String, Object> configuration) {
     public GatewayAuthenticationResult authenticate(GatewayRequestContext context) {
         String role;
         try {
-            role = authenticationProviderToken.authenticate(context.credentials());
+            final String credentials = context.credentials();
+            role = authenticationProviderToken.authenticate(credentials == null ? "" : credentials);
         } catch (AuthenticationProviderToken.AuthenticationException ex) {
             return GatewayAuthenticationResult.authenticationFailed(ex.getMessage());
         }

langstream-api-gateway/pom.xml

@@ -61,6 +61,13 @@
 			<scope>runtime</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>langstream-pulsar-runtime</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-configuration-processor</artifactId>
@@ -70,6 +77,16 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-websocket</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springdoc</groupId>
+			<artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
+			<version>${springdoc-openapi-starter-webmvc.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springdoc</groupId>
+			<artifactId>springdoc-openapi-starter-webmvc-api</artifactId>
+			<version>${springdoc-openapi-starter-webmvc.version}</version>
+		</dependency>
 		<dependency>
 			<groupId>ch.qos.logback</groupId>
 			<artifactId>logback-core</artifactId>

langstream-api-gateway/src/main/java/ai/langstream/apigateway/gateways/ConsumeGateway.java

@@ -0,0 +1,240 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.apigateway.gateways;
+
+import ai.langstream.api.model.Gateway;
+import ai.langstream.api.model.StreamingCluster;
+import ai.langstream.api.runner.code.Header;
+import ai.langstream.api.runner.code.Record;
+import ai.langstream.api.runner.topics.TopicConnectionsRuntime;
+import ai.langstream.api.runner.topics.TopicConnectionsRuntimeRegistry;
+import ai.langstream.api.runner.topics.TopicOffsetPosition;
+import ai.langstream.api.runner.topics.TopicReadResult;
+import ai.langstream.api.runner.topics.TopicReader;
+import ai.langstream.apigateway.websocket.AuthenticatedGatewayRequestContext;
+import ai.langstream.apigateway.websocket.api.ConsumePushMessage;
+import ai.langstream.apigateway.websocket.api.ProduceResponse;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.Closeable;
+import java.util.Base64;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class ConsumeGateway implements Closeable {
+
+    protected static final ObjectMapper mapper = new ObjectMapper();
+
+    @Getter
+    public static class ProduceException extends Exception {
+
+        private final ProduceResponse.Status status;
+
+        public ProduceException(String message, ProduceResponse.Status status) {
+            super(message);
+            this.status = status;
+        }
+    }
+
+    public static class ProduceGatewayRequestValidator
+            implements GatewayRequestHandler.GatewayRequestValidator {
+        @Override
+        public List<String> getAllRequiredParameters(Gateway gateway) {
+            return gateway.getParameters();
+        }
+
+        @Override
+        public void validateOptions(Map<String, String> options) {
+            for (Map.Entry<String, String> option : options.entrySet()) {
+                switch (option.getKey()) {
+                    default -> throw new IllegalArgumentException(
+                            "Unknown option " + option.getKey());
+                }
+            }
+        }
+    }
+
+    private final TopicConnectionsRuntimeRegistry topicConnectionsRuntimeRegistry;
+
+    private volatile TopicReader reader;
+    private volatile boolean interrupted;
+    private volatile String logRef;
+    private CompletableFuture<Void> readerFuture;
+    private AuthenticatedGatewayRequestContext requestContext;
+    private List<Function<Record, Boolean>> filters;
+
+    public ConsumeGateway(TopicConnectionsRuntimeRegistry topicConnectionsRuntimeRegistry) {
+        this.topicConnectionsRuntimeRegistry = topicConnectionsRuntimeRegistry;
+    }
+
+    public void setup(
+            String topic,
+            List<Function<Record, Boolean>> filters,
+            AuthenticatedGatewayRequestContext requestContext)
+            throws Exception {
+        this.logRef =
+                "%s/%s/%s"
+                        .formatted(
+                                requestContext.tenant(),
+                                requestContext.applicationId(),
+                                requestContext.gateway().getId());
+        this.requestContext = requestContext;
+        this.filters = filters == null ? List.of() : filters;
+
+        final StreamingCluster streamingCluster =
+                requestContext.application().getInstance().streamingCluster();
+        final TopicConnectionsRuntime topicConnectionsRuntime =
+                topicConnectionsRuntimeRegistry
+                        .getTopicConnectionsRuntime(streamingCluster)
+                        .asTopicConnectionsRuntime();
+
+        topicConnectionsRuntime.init(streamingCluster);
+
+        final String positionParameter =
+                requestContext.options().getOrDefault("position", "latest");
+        TopicOffsetPosition position =
+                switch (positionParameter) {
+                    case "latest" -> TopicOffsetPosition.LATEST;
+                    case "earliest" -> TopicOffsetPosition.EARLIEST;
+                    default -> TopicOffsetPosition.absolute(
+                            Base64.getDecoder().decode(positionParameter));
+                };
+        reader =
+                topicConnectionsRuntime.createReader(
+                        streamingCluster, Map.of("topic", topic), position);
+        reader.start();
+    }
+
+    public void startReadingAsync(
+            Executor executor, Supplier<Boolean> stop, Consumer<String> onMessage) {
+        if (requestContext == null || reader == null) {
+            throw new IllegalStateException("Not initialized");
+        }
+        if (readerFuture != null) {
+            throw new IllegalStateException("Already started");
+        }
+        readerFuture =
+                CompletableFuture.runAsync(
+                        () -> {
+                            try {
+                                log.debug("[{}] Started reader", logRef);
+                                readMessages(stop, onMessage);
+                            } catch (Throwable ex) {
+                                throw new RuntimeException(ex);
+                            } finally {
+                                closeReader();
+                            }
+                        },
+                        executor);
+    }
+
+    protected void readMessages(Supplier<Boolean> stop, Consumer<String> onMessage)
+            throws Exception {
+        while (true) {
+            if (interrupted) {
+                return;
+            }
+            if (stop.get()) {
+                return;
+            }
+            final TopicReadResult readResult = reader.read();
+            final List<Record> records = readResult.records();
+            for (Record record : records) {
+                log.debug("[{}] Received record {}", logRef, record);
+                boolean skip = false;
+                if (filters != null) {
+                    for (Function<Record, Boolean> filter : filters) {
+                        if (!filter.apply(record)) {
+                            skip = true;
+                            log.debug("[{}] Skipping record {}", logRef, record);
+                            break;
+                        }
+                    }
+                }
+                if (!skip) {
+                    final Map<String, String> messageHeaders = computeMessageHeaders(record);
+                    final String offset = computeOffset(readResult);
+
+                    final ConsumePushMessage message =
+                            new ConsumePushMessage(
+                                    new ConsumePushMessage.Record(
+                                            record.key(), record.value(), messageHeaders),
+                                    offset);
+                    final String jsonMessage = mapper.writeValueAsString(message);
+                    onMessage.accept(jsonMessage);
+                }
+            }
+        }
+    }
+
+    private static Map<String, String> computeMessageHeaders(Record record) {
+        final Collection<Header> headers = record.headers();
+        final Map<String, String> messageHeaders;
+        if (headers == null) {
+            messageHeaders = Map.of();
+        } else {
+            messageHeaders = new HashMap<>();
+            headers.forEach(h -> messageHeaders.put(h.key(), h.valueAsString()));
+        }
+        return messageHeaders;
+    }
+
+    private static String computeOffset(TopicReadResult readResult) {
+        final byte[] offset = readResult.offset();
+        if (offset == null) {
+            return null;
+        }
+        return Base64.getEncoder().encodeToString(offset);
+    }
+
+    private void closeReader() {
+        if (reader != null) {
+            try {
+                reader.close();
+            } catch (Exception e) {
+                log.warn("error closing reader", e);
+            }
+        }
+    }
+
+    @Override
+    public void close() {
+        if (readerFuture != null) {
+
+            interrupted = true;
+            try {
+                // reader.close must be done by the same thread that started the consumer
+                readerFuture.get(10, TimeUnit.SECONDS);
+            } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                log.debug("error waiting for reader to stop", e);
+            }
+        } else {
+            closeReader();
+        }
+    }
+}