orpiske commented on code in PR #23711: URL: https://github.com/apache/camel/pull/23711#discussion_r3347387364
########## components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java: ########## @@ -0,0 +1,1278 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.component.a2a; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.camel.Consumer; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.component.a2a.model.A2AError; +import org.apache.camel.component.a2a.model.AgentCard; +import org.apache.camel.component.a2a.model.Message; +import org.apache.camel.component.a2a.model.SendMessageRequest; +import org.apache.camel.component.a2a.model.SendMessageResponse; +import org.apache.camel.component.a2a.model.StreamResponse; +import org.apache.camel.component.a2a.model.Task; +import org.apache.camel.component.a2a.model.TaskPushNotificationConfig; +import org.apache.camel.component.a2a.model.TaskState; +import org.apache.camel.component.a2a.model.TaskStatus; +import org.apache.camel.component.a2a.model.TaskStatusUpdateEvent; +import org.apache.camel.component.a2a.operation.A2AOperations; +import org.apache.camel.component.a2a.operation.MessageSendOperation; +import org.apache.camel.component.a2a.protocol.JsonRpcProtocol; +import org.apache.camel.component.a2a.state.A2ATaskStore; +import org.apache.camel.component.a2a.state.A2ATaskSubscriber; +import org.apache.camel.component.a2a.streaming.DefaultStreamEmitter; +import org.apache.camel.component.a2a.streaming.QueueStreamEmitter; +import org.apache.camel.component.a2a.streaming.SseQueueInputStream; +import org.apache.camel.component.a2a.streaming.StreamSubscriber; +import org.apache.camel.component.a2a.util.A2AJsonMapper; +import org.apache.camel.component.rest.RestEndpoint; +import org.apache.camel.spi.RestConfiguration; +import org.apache.camel.spi.RestConsumerFactory; +import org.apache.camel.support.DefaultConsumer; +import org.apache.camel.support.DefaultExchange; +import org.apache.camel.support.service.ServiceHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A2A consumer that automatically registers HTTP endpoints via {@link RestConsumerFactory} SPI. + * <p> + * On startup, discovers a {@link RestConsumerFactory} (typically provided by camel-platform-http) and registers routes + * for all A2A operations. For REST binding, creates separate routes per operation path. For JSON-RPC binding, creates a + * single POST route that dispatches via the JSON-RPC method field. + * <p> + * The agent card is always served at {@code /.well-known/agent-card.json} regardless of protocol binding. + */ +public class A2AConsumer extends DefaultConsumer { + + private static final Logger LOG = LoggerFactory.getLogger(A2AConsumer.class); + private static final ObjectMapper OBJECT_MAPPER = A2AJsonMapper.instance(); + + private final MessageSendOperation messageSendOperation; + private final List<Consumer> httpConsumers = new ArrayList<>(); + private final ConcurrentHashMap<String, Future<?>> inFlightTasks = new ConcurrentHashMap<>(); + private final ConcurrentHashMap<String, java.util.concurrent.ScheduledFuture<?>> timeoutFutures = new ConcurrentHashMap<>(); + private ExecutorService asyncExecutor; + private ScheduledExecutorService asyncTimeoutScheduler; + private Semaphore taskPermits; + private LinkedBlockingQueue<PendingTask> pendingTaskQueue; + private boolean corsEnabled; + private Map<String, String> corsHeaders; + + public A2AConsumer(A2AEndpoint endpoint, Processor processor) { + super(endpoint, processor); + this.messageSendOperation = new MessageSendOperation(); + } + + @Override + public A2AEndpoint getEndpoint() { + return (A2AEndpoint) super.getEndpoint(); + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + + AgentCard card = getEndpoint().getResolvedCard(); + String agentName = card != null ? card.getName() : "unknown"; + LOG.info("A2A Consumer starting for agent: {}", agentName); + + asyncExecutor = getEndpoint().getCamelContext() + .getExecutorServiceManager() + .newThreadPool(this, "A2AAsyncProcessor", 1, 10); + asyncTimeoutScheduler = getEndpoint().getCamelContext() + .getExecutorServiceManager() + .newScheduledThreadPool(this, "A2AAsyncTimeout", 1); + + int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks(); + if (maxTasks > 0) { + taskPermits = new Semaphore(maxTasks); + int queueSize = getEndpoint().getConfiguration().getTaskQueueSize(); + if (queueSize > 0) { + pendingTaskQueue = new LinkedBlockingQueue<>(queueSize); + } + LOG.info("A2A capacity limits: maxConcurrentTasks={}, taskQueueSize={}", maxTasks, queueSize); + } + + RestConsumerFactory factory = resolveRestConsumerFactory(); + if (factory == null) { + LOG.warn("No RestConsumerFactory found — HTTP endpoints will not be registered automatically. " + + "Add camel-platform-http or another REST consumer implementation to the classpath."); + return; + } + + RestConfiguration restConfig = getEndpoint().getCamelContext().getRestConfiguration(); + String basePath = getEndpoint().getConfiguration().getBasePath(); + + corsEnabled = restConfig.isEnableCORS(); + corsHeaders = restConfig.getCorsHeaders(); + + // Agent card is always public — no auth required + registerRoute(factory, restConfig, "GET", basePath + A2AConstants.WELL_KNOWN_PATH, + this::handleAgentCardRequest, true); + + boolean isJsonRpc + = A2AConstants.PROTOCOL_JSONRPC.equalsIgnoreCase(getEndpoint().getConfiguration().getProtocolBinding()); + + if (isJsonRpc) { + registerSseRoute(factory, restConfig, "POST", basePath + "/", + this::handleJsonRpcDispatch); + } else { + registerRoute(factory, restConfig, "POST", basePath + "/message:send", + this::handleSendMessage); + registerSseRoute(factory, restConfig, "POST", basePath + "/message:stream", + this::handleMessageStream); + registerRoute(factory, restConfig, "GET", basePath + "/tasks", + this::handleListTasks); + registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}", + this::handleGetTask); + registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}:cancel", + this::handleCancelTask); + registerSseRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}:subscribe", + this::handleTaskSubscribe); + registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}/pushNotificationConfigs", + this::handlePushConfigCreate); + registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}/pushNotificationConfigs", + this::handlePushConfigList); + registerRoute(factory, restConfig, "GET", + basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}", + this::handlePushConfigGet); + registerRoute(factory, restConfig, "DELETE", + basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}", + this::handlePushConfigDelete); + } + + if (corsEnabled) { + registerRoute(factory, restConfig, "OPTIONS", basePath + A2AConstants.WELL_KNOWN_PATH, + this::handleCorsPreFlight, true); + if (isJsonRpc) { + registerRoute(factory, restConfig, "OPTIONS", basePath + "/", + this::handleCorsPreFlight, true); + } else { + for (String path : List.of( + "/message:send", "/message:stream", "/tasks", "/tasks/{taskId}", + "/tasks/{taskId}:cancel", "/tasks/{taskId}:subscribe", + "/tasks/{taskId}/pushNotificationConfigs", + "/tasks/{taskId}/pushNotificationConfigs/{configId}")) { + registerRoute(factory, restConfig, "OPTIONS", basePath + path, + this::handleCorsPreFlight, true); + } + } + } + + for (Consumer consumer : httpConsumers) { + ServiceHelper.startService(consumer); + } + + LOG.info("A2A Consumer registered {} HTTP endpoint(s) for agent '{}'", + httpConsumers.size(), agentName); + } + + @Override + protected void doStop() throws Exception { + cancelAllInFlight(); + for (Consumer consumer : httpConsumers) { + ServiceHelper.stopService(consumer); + } + httpConsumers.clear(); + super.doStop(); + } + + @Override + protected void doShutdown() throws Exception { + failPendingQueuedTasks(); + cancelAllInFlight(); + if (asyncExecutor != null) { + getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncExecutor); + } + if (asyncTimeoutScheduler != null) { + getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncTimeoutScheduler); + } + for (Consumer consumer : httpConsumers) { + ServiceHelper.stopAndShutdownService(consumer); + } + httpConsumers.clear(); + super.doShutdown(); + } + + private void cancelAllInFlight() { + for (Future<?> future : inFlightTasks.values()) { + future.cancel(true); + } + inFlightTasks.clear(); + } + + // ---- Capacity limiting ---- + + private boolean tryAcquirePermit() { + return taskPermits == null || taskPermits.tryAcquire(); + } + + private void releasePermit() { + if (taskPermits != null) { + taskPermits.release(); + drainPendingQueue(); + } + } + + private void drainPendingQueue() { + if (pendingTaskQueue == null || pendingTaskQueue.isEmpty()) { + return; + } + if (!taskPermits.tryAcquire()) { + return; + } + PendingTask pending = pendingTaskQueue.poll(); + if (pending == null) { + taskPermits.release(); + return; + } + submitAsyncTask(pending.taskId, pending.contextId, pending.processorExchange); + } + + private void failPendingQueuedTasks() { + if (pendingTaskQueue == null) { + return; + } + A2ATaskStore store = getEndpoint().getTaskStore(); + PendingTask pending; + while ((pending = pendingTaskQueue.poll()) != null) { + Task failedTask = Task.builder() + .id(pending.taskId) + .contextId(pending.contextId) + .status(new TaskStatus(TaskState.FAILED)) + .build(); + store.put(pending.taskId, failedTask); + store.notifySubscribers(pending.taskId, + StreamResponse.ofStatusUpdate(buildStatusEvent(failedTask))); + } + } + + private void writeServerBusyError(Exchange exchange) throws Exception { + int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks(); + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError( + "ServerBusyError", + "Agent at capacity: " + maxTasks + " concurrent tasks")); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 429); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + + static class ServerBusyException extends RuntimeException { + ServerBusyException(String message) { + super(message); + } + } + + private record PendingTask(String taskId, String contextId, Exchange processorExchange) { + } + + private void registerRoute( + RestConsumerFactory factory, RestConfiguration restConfig, + String verb, String path, A2ARequestHandler handler) + throws Exception { + registerRoute(factory, restConfig, verb, path, handler, false, A2AConstants.CONTENT_TYPE); + } + + private void registerRoute( + RestConsumerFactory factory, RestConfiguration restConfig, + String verb, String path, A2ARequestHandler handler, boolean isPublic) + throws Exception { + registerRoute(factory, restConfig, verb, path, handler, isPublic, A2AConstants.CONTENT_TYPE); + } + + private void registerSseRoute( + RestConsumerFactory factory, RestConfiguration restConfig, + String verb, String path, A2ARequestHandler handler) + throws Exception { + Consumer consumer = factory.createConsumer( + getEndpoint().getCamelContext(), + createDispatchProcessor(handler, false), + verb, path, null, + null, SSE_CONTENT_TYPE, + restConfig, Map.of("useStreaming", "true")); + httpConsumers.add(consumer); + LOG.debug("Registered A2A SSE route: {} {}", verb, path); + } + + private void registerRoute( + RestConsumerFactory factory, RestConfiguration restConfig, + String verb, String path, A2ARequestHandler handler, boolean isPublic, + String produces) + throws Exception { + Consumer consumer = factory.createConsumer( + getEndpoint().getCamelContext(), + createDispatchProcessor(handler, isPublic), + verb, path, null, + null, produces, + restConfig, Collections.emptyMap()); + httpConsumers.add(consumer); + LOG.debug("Registered A2A route: {} {}", verb, path); + } + + private Processor createDispatchProcessor(A2ARequestHandler handler, boolean isPublic) { + return exchange -> { + try { + if (!isPublic) { + Map<String, Object> profile = getEndpoint().getAuthHandler() + .validateConsumerAuth(exchange, getEndpoint().getResolvedCard()); + if (profile != null) { + exchange.getMessage().setHeader(A2AConstants.USER_PROFILE, profile); + } + } + handler.handle(exchange); + } catch (SecurityException e) { + LOG.debug("Authentication failed: {}", e.getMessage()); + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("AuthenticationError", e.getMessage())); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 401); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } catch (Exception e) { + LOG.error("A2A request handling failed", e); + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("InternalError", e.getMessage())); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 500); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } finally { + if (corsEnabled) { + setCorsHeaders(exchange); + } + } + }; + } + + void handleCorsPreFlight(Exchange exchange) { + exchange.getMessage().setBody(""); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 200); + } + + private void setCorsHeaders(Exchange exchange) { + String allowOrigin = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Origin") : null; + if (allowOrigin == null) { + allowOrigin = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_ORIGIN; + } + String allowMethods = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Methods") : null; + if (allowMethods == null) { + allowMethods = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_METHODS; + } + String allowHeaders = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Headers") : null; + if (allowHeaders == null) { + allowHeaders = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_HEADERS + + ", A2A-Version, A2A-Extensions, Authorization"; + } + String maxAge = corsHeaders != null ? corsHeaders.get("Access-Control-Max-Age") : null; + if (maxAge == null) { + maxAge = RestConfiguration.CORS_ACCESS_CONTROL_MAX_AGE; + } + String allowCredentials = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Credentials") : null; + + // Restrict the origin if credentials are allowed (W3C CORS spec section 6.1, point 3) + String origin = exchange.getIn().getHeader("Origin", String.class); + if ("true".equalsIgnoreCase(allowCredentials) && "*".equals(allowOrigin) && origin != null) { + allowOrigin = origin; + } + + exchange.getMessage().setHeader("Access-Control-Allow-Origin", allowOrigin); + exchange.getMessage().setHeader("Access-Control-Allow-Methods", allowMethods); + exchange.getMessage().setHeader("Access-Control-Allow-Headers", allowHeaders); + exchange.getMessage().setHeader("Access-Control-Max-Age", maxAge); + if (allowCredentials != null) { + exchange.getMessage().setHeader("Access-Control-Allow-Credentials", allowCredentials); + } + } + + private RestConsumerFactory resolveRestConsumerFactory() { + // 1. Explicit httpServerComponent takes priority + String serverComponent = getEndpoint().getConfiguration().getHttpServerComponent(); + if (serverComponent != null) { + Object component = getEndpoint().getCamelContext().getComponent(serverComponent, true); + if (component instanceof RestConsumerFactory rcf) { + return rcf; + } + throw new IllegalArgumentException( + "httpServerComponent '" + serverComponent + "' does not implement RestConsumerFactory"); + } + + // 2. Scan existing components + for (String name : getEndpoint().getCamelContext().getComponentNames()) { + Object component = getEndpoint().getCamelContext().getComponent(name, false); + if (component instanceof RestConsumerFactory rcf) { + return rcf; + } + } + + // 3. Auto-discover from registry + Set<RestConsumerFactory> factories = getEndpoint().getCamelContext() + .getRegistry().findByType(RestConsumerFactory.class); + if (!factories.isEmpty()) { + return factories.iterator().next(); + } + + // 4. Auto-create default REST consumer components (same list as REST DSL). + // Skip platform-http — it needs a pre-configured engine and is already covered by step 2. + for (String name : RestEndpoint.DEFAULT_REST_CONSUMER_COMPONENTS) { + if ("platform-http".equals(name)) { + continue; + } + try { + Object component = getEndpoint().getCamelContext().getComponent(name, true); + if (component instanceof RestConsumerFactory rcf) { + LOG.info("Auto-discovered {} as RestConsumerFactory", name); + return rcf; + } + } catch (Exception e) { + LOG.debug("Could not auto-create component {}: {}", name, e.getMessage()); + } + } + return null; + } + + // ---- JSON-RPC dispatch ---- + + void handleJsonRpcDispatch(Exchange exchange) throws Exception { + filterInboundHeaders(exchange); + byte[] body = exchange.getMessage().getBody(byte[].class); + if (body == null || body.length == 0) { + writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: empty body", null); + return; + } + + long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize(); + if (body.length > maxSize) { + writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, + "Request body exceeds maximum size: " + maxSize + " bytes", null); + return; + } + + JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol(); + String method; + Object requestId; + Map<String, Object> params; + try { + method = jsonRpc.detectMethod(body); + requestId = jsonRpc.extractId(body); + params = jsonRpc.extractParams(body); + } catch (Exception e) { + writeJsonRpcError(exchange, JsonRpcProtocol.PARSE_ERROR, "Parse error: " + e.getMessage(), null); + return; + } + + if (method == null) { + writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: missing method", requestId); + return; + } + + // Streaming methods produce text/event-stream, not a single JSON-RPC envelope + A2AOperations op = A2AOperations.fromMethodName(method); + LOG.info("JSON-RPC dispatch: method={}, op={}", method, op); + if (op == A2AOperations.MESSAGE_STREAM) { + LOG.info("Entering handleJsonRpcStream for taskId={}", requestId); + try { + handleJsonRpcStream(exchange, params, requestId); + LOG.info("handleJsonRpcStream completed, contentType={}", exchange.getMessage().getHeader("Content-Type")); + } catch (Exception e) { + LOG.error("handleJsonRpcStream failed: {}", e.getMessage(), e); + writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR, + "Internal error: " + e.getMessage(), requestId); + } + return; + } + + Object result; + try { + result = dispatchJsonRpcMethod(method, params, exchange); + } catch (ServerBusyException e) { + writeJsonRpcError(exchange, -32000, e.getMessage(), requestId); + return; + } catch (IllegalArgumentException e) { + writeJsonRpcError(exchange, JsonRpcProtocol.METHOD_NOT_FOUND, "Method not found: " + method, requestId); + return; + } catch (Exception e) { + writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR, "Internal error: " + e.getMessage(), requestId); + return; + } + + byte[] response = jsonRpc.wrapJsonRpcResponse(result, requestId); + exchange.getMessage().setBody(response); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + + private Object dispatchJsonRpcMethod(String method, Map<String, Object> params, Exchange exchange) throws Exception { + A2AOperations op = A2AOperations.fromMethodName(method); + if (op == null) { + throw new IllegalArgumentException("Unknown method: " + method); + } + return switch (op) { + case MESSAGE_SEND -> { + SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class); + yield processSendMessage(request, exchange); + } + case TASK_GET -> { + String taskId = params != null ? (String) params.get("id") : null; + yield processGetTask(taskId); + } + case TASK_CANCEL -> { + String taskId = params != null ? (String) params.get("id") : null; + yield processCancelTask(taskId); + } + case TASK_LIST -> { + Integer pageSize = params != null ? (Integer) params.get("pageSize") : null; + String contextId = params != null ? (String) params.get("contextId") : null; + yield processListTasks(contextId, pageSize); + } + case PUSH_CONFIG_CREATE -> { + String taskId = params != null ? (String) params.get("taskId") : null; + TaskPushNotificationConfig config = OBJECT_MAPPER.convertValue(params, TaskPushNotificationConfig.class); + yield processPushConfigCreate(taskId, config); + } + case PUSH_CONFIG_GET -> { + String taskId = params != null ? (String) params.get("taskId") : null; + String configId = params != null ? (String) params.get("id") : null; + yield processPushConfigGet(taskId, configId); + } + case PUSH_CONFIG_LIST -> { + String taskId = params != null ? (String) params.get("taskId") : null; + yield processPushConfigList(taskId); + } + case PUSH_CONFIG_DELETE -> { + String taskId = params != null ? (String) params.get("taskId") : null; + String configId = params != null ? (String) params.get("id") : null; + yield processPushConfigDelete(taskId, configId); + } + default -> throw new IllegalArgumentException("Unsupported method: " + method); + }; + } + + private void writeJsonRpcError(Exchange exchange, int code, String message, Object requestId) throws Exception { + JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol(); + byte[] error = jsonRpc.wrapJsonRpcError(code, message, requestId); + exchange.getMessage().setBody(error); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + + // ---- JSON-RPC streaming ---- + + private void handleJsonRpcStream(Exchange exchange, Map<String, Object> params, Object requestId) throws Exception { + if (!tryAcquirePermit()) { + writeJsonRpcError(exchange, -32000, + "Agent at capacity: " + getEndpoint().getConfiguration().getMaxConcurrentTasks() + + " concurrent tasks", + requestId); + return; + } + + SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class); + String taskId = UUID.randomUUID().toString(); + String contextId = request.getMessage().contextId(); + if (contextId == null || contextId.isEmpty()) { + contextId = UUID.randomUUID().toString(); + } + + A2ATaskStore store = getEndpoint().getTaskStore(); + store.put(taskId, Task.builder() + .id(taskId).contextId(contextId) + .status(new TaskStatus(TaskState.SUBMITTED)).build()); + + JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol(); + final Object rpcRequestId = requestId; + Function<StreamResponse, String> jsonRpcEncoder = response -> { + byte[] envelope = jsonRpc.wrapJsonRpcResponse(response, rpcRequestId); + return "data: " + new String(envelope, StandardCharsets.UTF_8) + "\n\n"; + }; + + LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(); + QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, contextId, queue, jsonRpcEncoder); + SseQueueInputStream inputStream = new SseQueueInputStream( + queue, + getEndpoint().getConfiguration().getAsyncTimeout()); + + A2ATaskSubscriber subscriber = new StreamSubscriber(emitter); + store.addSubscriber(taskId, subscriber); + inputStream.setOnClose(() -> { + store.removeSubscriber(taskId, subscriber); + emitter.close(); + }); + + submitStreamProcessing(taskId, contextId, request, emitter); + + exchange.getExchangeExtension().setStreamCacheDisabled(true); + exchange.getMessage().setBody(inputStream); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE); + exchange.getMessage().setHeader("Cache-Control", "no-cache"); + exchange.getMessage().setHeader("Connection", "keep-alive"); + } + + // ---- REST handler methods ---- + + void handleAgentCardRequest(Exchange exchange) throws Exception { + filterInboundHeaders(exchange); + AgentCard card = getEndpoint().getResolvedCard(); + byte[] cardJson = OBJECT_MAPPER.writeValueAsBytes(card); + + exchange.getMessage().setBody(cardJson); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + + void handleSendMessage(Exchange exchange) throws Exception { + filterInboundHeaders(exchange); + String requestBody = exchange.getMessage().getBody(String.class); + if (requestBody == null || requestBody.isBlank()) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("ContentTypeNotSupportedError", "Request body is empty")); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + return; + } + + long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize(); + if (requestBody.length() > maxSize) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError( + "ContentTypeNotSupportedError", + "Request body exceeds maximum size: " + maxSize + " bytes")); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 413); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + return; + } + + SendMessageRequest request = OBJECT_MAPPER.readValue(requestBody, SendMessageRequest.class); + + SendMessageResponse response; + try { + response = processSendMessage(request, exchange); + } catch (ServerBusyException e) { + writeServerBusyError(exchange); + return; + } + + byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(response); + exchange.getMessage().setBody(responseJson); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + + void handleGetTask(Exchange exchange) throws Exception { + filterInboundHeaders(exchange); + String taskId = extractTaskIdFromPath(exchange); + + try { + Task task = processGetTask(taskId); + byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task); + exchange.getMessage().setBody(responseJson); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } catch (TaskNotFoundException e) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("TaskNotFoundError", e.getMessage())); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + } + + void handleCancelTask(Exchange exchange) throws Exception { + filterInboundHeaders(exchange); + String taskId = extractTaskIdFromPath(exchange); + + try { + Task task = processCancelTask(taskId); + byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task); + exchange.getMessage().setBody(responseJson); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } catch (TaskNotFoundException e) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("TaskNotFoundError", e.getMessage())); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } catch (IllegalStateException e) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("TaskNotCancelableError", e.getMessage())); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 409); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + } + + void handleListTasks(Exchange exchange) throws Exception { + Integer pageSize = exchange.getMessage().getHeader(A2AConstants.LIST_PAGE_SIZE, Integer.class); + String contextId = exchange.getMessage().getHeader(A2AConstants.LIST_CONTEXT_ID, String.class); + + filterInboundHeaders(exchange); + + List<Task> tasks = processListTasks(contextId, pageSize); + + byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(tasks); + exchange.getMessage().setBody(responseJson); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + } + + // ---- SSE streaming handlers ---- + + private static final String SSE_CONTENT_TYPE = "text/event-stream"; + + void handleMessageStream(Exchange exchange) throws Exception { + filterInboundHeaders(exchange); + String requestBody = exchange.getMessage().getBody(String.class); + if (requestBody == null || requestBody.isBlank()) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("ContentTypeNotSupportedError", "Request body is empty")); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + return; + } + + long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize(); + if (requestBody.length() > maxSize) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError( + "ContentTypeNotSupportedError", + "Request body exceeds maximum size: " + maxSize + " bytes")); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 413); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + return; + } + + SendMessageRequest request = OBJECT_MAPPER.readValue(requestBody, SendMessageRequest.class); + + if (resolveReturnImmediately(request, exchange)) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError( + "UnsupportedOperationError", + "returnImmediately is not supported for streaming operations per A2A spec")); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + return; + } + + if (!tryAcquirePermit()) { + writeServerBusyError(exchange); + return; + } + + String taskId = UUID.randomUUID().toString(); + String contextId = request.getMessage().contextId(); + if (contextId == null || contextId.isEmpty()) { + contextId = UUID.randomUUID().toString(); + } + + A2ATaskStore store = getEndpoint().getTaskStore(); + store.put(taskId, Task.builder() + .id(taskId).contextId(contextId) + .status(new TaskStatus(TaskState.SUBMITTED)).build()); + + LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(); + QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, contextId, queue); + SseQueueInputStream inputStream = new SseQueueInputStream( + queue, + getEndpoint().getConfiguration().getAsyncTimeout()); + + A2ATaskSubscriber subscriber = new StreamSubscriber(emitter); + store.addSubscriber(taskId, subscriber); + inputStream.setOnClose(() -> { + store.removeSubscriber(taskId, subscriber); + emitter.close(); + }); + + submitStreamProcessing(taskId, contextId, request, emitter); + + exchange.getExchangeExtension().setStreamCacheDisabled(true); + exchange.getMessage().setBody(inputStream); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE); + exchange.getMessage().setHeader("Cache-Control", "no-cache"); + exchange.getMessage().setHeader("Connection", "keep-alive"); + } + + void handleTaskSubscribe(Exchange exchange) throws Exception { + filterInboundHeaders(exchange); + String taskId = extractTaskIdFromPath(exchange); + A2ATaskStore store = getEndpoint().getTaskStore(); + + Task task = store.get(taskId); + if (task == null) { + byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes( + new A2AError("TaskNotFoundError", "Task not found: " + taskId)); + exchange.getMessage().setBody(errorJson); + exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE); + return; + } + + long timeout = getEndpoint().getConfiguration().getAsyncTimeout(); + LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(); + QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, task.contextId(), queue); + SseQueueInputStream inputStream = new SseQueueInputStream(queue, timeout); + + A2ATaskSubscriber subscriber = new StreamSubscriber(emitter); + store.addSubscriber(taskId, subscriber); + + // Cleanup: unsubscribe when the HTTP component closes the stream + inputStream.setOnClose(() -> { + store.removeSubscriber(taskId, subscriber); + emitter.close(); + }); + + // Emit current state as the first event + if (task.status() != null) { + emitter.emitStatus(task.status().state(), null); + } + + // If already terminal, close immediately — queued events + EOF will be read before -1 + if (task.status() != null && task.status().state() != null + && task.status().state().isTerminal()) { + emitter.close(); + } + + exchange.getExchangeExtension().setStreamCacheDisabled(true); + exchange.getMessage().setBody(inputStream); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE); + exchange.getMessage().setHeader("Cache-Control", "no-cache"); + exchange.getMessage().setHeader("Connection", "keep-alive"); + } + + // ---- Shared processing logic (used by both REST and JSON-RPC paths) ---- + + private SendMessageResponse processSendMessage(SendMessageRequest request, Exchange exchange) throws Exception { + DefaultExchange processorExchange = new DefaultExchange(getEndpoint().getCamelContext()); + processorExchange.getExchangeExtension().setFromEndpoint(getEndpoint()); + messageSendOperation.parseRequest(processorExchange, request); + + String taskId = UUID.randomUUID().toString(); + String contextId = request.getMessage().contextId(); + if (contextId == null || contextId.isEmpty()) { + contextId = UUID.randomUUID().toString(); + } + + processorExchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + processorExchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId); + + if (resolveReturnImmediately(request, exchange)) { + return processSendMessageAsync(processorExchange, taskId, contextId); + } + + if (!tryAcquirePermit()) { + throw new ServerBusyException( + "Agent at capacity: " + + getEndpoint().getConfiguration().getMaxConcurrentTasks() + " concurrent tasks"); + } + try { + getProcessor().process(processorExchange); + } finally { + releasePermit(); + } + + SendMessageResponse response = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange); + + if (response.isTaskResponse()) { + Task task = response.getTask(); + A2ATaskStore store = getEndpoint().getTaskStore(); + store.put(task.id(), task); + if (task.status() != null) { + store.notifySubscribers(task.id(), + StreamResponse.ofStatusUpdate(buildStatusEvent(task))); + } + } + + return response; + } + + private SendMessageResponse processSendMessageAsync( + Exchange processorExchange, String taskId, String contextId) { + A2ATaskStore store = getEndpoint().getTaskStore(); + + Task submittedTask = Task.builder() + .id(taskId) + .contextId(contextId) + .status(new TaskStatus(TaskState.SUBMITTED)) + .build(); + + if (tryAcquirePermit()) { + store.put(taskId, submittedTask); + store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(submittedTask))); + submitAsyncTask(taskId, contextId, processorExchange); + } else if (pendingTaskQueue != null + && pendingTaskQueue.offer(new PendingTask(taskId, contextId, processorExchange))) { + store.put(taskId, submittedTask); + store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(submittedTask))); + LOG.debug("Task {} queued (queue size: {})", taskId, pendingTaskQueue.size()); + } else { + throw new ServerBusyException( + "Agent at capacity: " + + getEndpoint().getConfiguration().getMaxConcurrentTasks() + " concurrent tasks"); + } + + Task responseTask = Task.builder() + .id(taskId) + .contextId(contextId) + .status(new TaskStatus(TaskState.SUBMITTED)) + .build(); + + SendMessageResponse response = new SendMessageResponse(); + response.setTask(responseTask); + return response; + } + + private void submitAsyncTask(String taskId, String contextId, Exchange processorExchange) { + A2ATaskStore store = getEndpoint().getTaskStore(); + + Future<?> future = asyncExecutor.submit(() -> { + try { + Task workingTask = store.get(taskId); + if (workingTask != null) { + store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.WORKING)); + } + getProcessor().process(processorExchange); + SendMessageResponse asyncResponse + = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange); + if (asyncResponse.isTaskResponse()) { + Task completed = Task.builder(asyncResponse.getTask()) + .id(taskId) + .contextId(contextId) + .build(); + store.put(taskId, completed); + store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(completed))); + } + } catch (Exception e) { + LOG.error("Async task processing failed for taskId={}", taskId, e); + store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED)); + } finally { + inFlightTasks.remove(taskId); + java.util.concurrent.ScheduledFuture<?> tf = timeoutFutures.remove(taskId); + if (tf != null) { + tf.cancel(false); + } + releasePermit(); + } + }); + inFlightTasks.put(taskId, future); + + long asyncTimeout = getEndpoint().getConfiguration().getAsyncTimeout(); + if (asyncTimeout > 0) { + java.util.concurrent.ScheduledFuture<?> timeoutFuture = asyncTimeoutScheduler.schedule(() -> { + timeoutFutures.remove(taskId); + Future<?> f = inFlightTasks.get(taskId); + if (f != null && !f.isDone()) { + f.cancel(true); + store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED)); + inFlightTasks.remove(taskId); + LOG.warn("Async task timed out after {}ms for taskId={}", asyncTimeout, taskId); + } + }, asyncTimeout, TimeUnit.MILLISECONDS); + timeoutFutures.put(taskId, timeoutFuture); + } + } + + private boolean resolveReturnImmediately(SendMessageRequest request, Exchange exchange) { + if (request.getConfiguration() != null) { + Object val = request.getConfiguration().get("returnImmediately"); + if (val instanceof Boolean) { + return (Boolean) val; + } + if (val instanceof String) { + return Boolean.parseBoolean((String) val); + } + } + Boolean header = exchange.getMessage().getHeader(A2AConstants.RETURN_IMMEDIATELY, Boolean.class); + if (header != null) { + return header; + } + return getEndpoint().getConfiguration().isReturnImmediately(); + } + + private Task processGetTask(String taskId) throws TaskNotFoundException { + A2ATaskStore store = getEndpoint().getTaskStore(); + Task task = store.get(taskId); + if (task == null) { + throw new TaskNotFoundException("Task not found: " + taskId); + } + return task; + } + + private Task processCancelTask(String taskId) throws TaskNotFoundException { + A2ATaskStore store = getEndpoint().getTaskStore(); + Task task = store.cancelIfNotTerminal(taskId); + if (task == null) { + throw new TaskNotFoundException("Task not found: " + taskId); + } + Future<?> inFlight = inFlightTasks.remove(taskId); + if (inFlight != null) { + inFlight.cancel(true); + } + store.notifySubscribers(taskId, + StreamResponse.ofStatusUpdate(buildStatusEvent(task))); + return task; + } + + private void submitStreamProcessing( + String taskId, String contextId, SendMessageRequest request, + QueueStreamEmitter emitter) { + DefaultExchange processorExchange = new DefaultExchange(getEndpoint().getCamelContext()); + processorExchange.getExchangeExtension().setFromEndpoint(getEndpoint()); + processorExchange.getMessage().setBody(request.getMessage()); + processorExchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + processorExchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId); + processorExchange.getMessage().setHeader(A2AConstants.OPERATION, A2AOperations.MESSAGE_STREAM.getMethodName()); + + asyncExecutor.submit(() -> { + try { + getProcessor().process(processorExchange); + SendMessageResponse resp = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange); + if (resp.isTaskResponse()) { + Task task = resp.getTask(); + if (task.history() != null && !task.history().isEmpty()) { + Message agentMessage = task.history().get(task.history().size() - 1); + A2AProgress.emitMessage(processorExchange, agentMessage); + } + } + } catch (Exception e) { + LOG.error("Stream processing failed for task {}: {}", taskId, e.getMessage(), e); + A2ATaskStore store = getEndpoint().getTaskStore(); + store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED)); + } finally { + emitter.close(); + releasePermit(); + } + }); + } + + private DefaultStreamEmitter executeStreamProcessing(SendMessageRequest request) throws Exception { Review Comment: This one was flagged as unused. Please, can you double check @luigidemasi ? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
