luigidemasi commented on code in PR #26684: URL: https://github.com/apache/camel/pull/26684#discussion_r4063943003
########## components/camel-ai/camel-jev/src/main/java/org/apache/camel/component/jev/JevClient.java: ########## @@ -0,0 +1,114 @@ +/* + * 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.jev; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.apache.camel.util.json.JsonObject; +import org.apache.camel.util.json.Jsoner; + +/** HTTP transport shared by the endpoint's producer and predicates. */ +final class JevClient implements AutoCloseable { + private final HttpClient http; + private final URI uri; + private final String apiKey; + private final String model; + private final Duration timeout; + private final Set<CompletableFuture<HttpResponse<String>>> pending = new HashSet<>(); Review Comment: Fixed in 74f145e83ea7654b3dadedd22d222ac27b1025b4. `maxConcurrentRequests` now defaults to 64 per endpoint and is configurable through component properties or endpoint options. Producers and predicates using the same endpoint share the limit. Admission uses the client's existing lock and happens before JSON processing or HTTP submission. Excess evaluations fail immediately with `RejectedExecutionException`; there is no request queue. Capacity is released in `finally`, including validation/transport failures, timeouts, interruption and cancellation. The tests cover saturation, producer/predicate sharing, independent endpoints, properties-only configuration, Camel error handling, and recovery after failure or cancellation. _AI-generated by OpenAI Codex on behalf of @luigidemasi via /oss-address-review._ ########## components/camel-ai/camel-jev/src/main/java/org/apache/camel/component/jev/JevPredicate.java: ########## @@ -0,0 +1,113 @@ +/* + * 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.jev; + +import java.util.Map; +import java.util.Objects; + +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.Expression; +import org.apache.camel.Predicate; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.util.json.JsonObject; + +/** + * A synchronous Noul predicate. Each invocation submits freshly selected state, preserves the message body and stores + * the complete response in {@link #RESULT}. The supplied state expression must be thread-safe. + */ +public final class JevPredicate implements Predicate { + public static final String RESULT = "CamelJevResult"; + + public enum UncertaintyPolicy { + NonMatch, + Fail + } + + private final String endpointUri; + private final Expression state; + private final String question; + private final double threshold; + private final double uncertainty; + private final UncertaintyPolicy uncertaintyPolicy; + private JevEndpoint endpoint; + + public JevPredicate(String endpointUri, Expression state, String instructions, double threshold) { + this(endpointUri, state, Map.of("type", "noul", "instructions", instructions), threshold); + } + + public JevPredicate(String endpointUri, Expression state, Map<String, Object> question, double threshold) { + this(endpointUri, state, question, threshold, 0, UncertaintyPolicy.NonMatch); + } + + /** + * @param endpointUri the configured Jev endpoint to share with producers and other predicates + * @param state selects only the exchange data to submit + * @param question the Noul proposition + * @param threshold a probability at or above this value matches, outside the uncertainty band + * @param uncertainty half-width of the inclusive band around threshold; zero disables the band + * @param uncertaintyPolicy whether uncertainty yields a non-match or an exception + */ + public JevPredicate(String endpointUri, Expression state, Map<String, Object> question, + double threshold, double uncertainty, UncertaintyPolicy uncertaintyPolicy) { + this.endpointUri = Objects.requireNonNull(endpointUri, "endpointUri"); + this.state = Objects.requireNonNull(state, "state"); + this.question = JevJson.noulQuestion(Objects.requireNonNull(question, "question")); + this.uncertaintyPolicy = Objects.requireNonNull(uncertaintyPolicy, "uncertaintyPolicy"); + if (!Double.isFinite(threshold) || threshold < 0 || threshold > 1 + || !Double.isFinite(uncertainty) || uncertainty < 0 + || threshold - uncertainty < 0 || threshold + uncertainty > 1) { + throw new IllegalArgumentException("The threshold and its uncertainty band must be within [0,1]"); + } + this.threshold = threshold; + this.uncertainty = uncertainty; + } + + @Override + public void init(CamelContext context) { + state.init(context); + // Register the endpoint even when it is used only by a predicate, so Camel owns its lifecycle. + endpoint = context.getEndpoint(endpointUri, JevEndpoint.class); + } + + @Override + public boolean matches(Exchange exchange) { + exchange.removeProperty(RESULT); + try { + if (endpoint == null) { + throw new IllegalStateException("Jev predicate must be initialized"); + } + Object selected = state.evaluate(exchange, Object.class); + if (selected == null) { + throw new IllegalArgumentException("Jev state must not be null"); + } + JsonObject result = endpoint.evaluate(Map.of("state", selected, + "questions", Map.of("predicate", JevJson.parse(question)))); Review Comment: Fixed in 74f145e83ea7654b3dadedd22d222ac27b1025b4. The predicate now stores a private `JsonObject` snapshot parsed once during construction. `matches()` reuses it, removing the extra per-evaluation question parse. The snapshot remains detached from the supplied map and is only read after construction. The concurrent predicate test also checks that later mutations to structured instructions do not affect requests and that nested null values are preserved. _AI-generated by OpenAI Codex on behalf of @luigidemasi via /oss-address-review._ -- 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]
