jamesnetherton commented on code in PR #9018: URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805508069
########## extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java: ########## @@ -0,0 +1,276 @@ +/* + * 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.quarkus.component.langchain4j.ingest; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; + +import dev.langchain4j.data.segment.TextSegment; +import dev.langchain4j.model.embedding.EmbeddingModel; +import dev.langchain4j.store.embedding.EmbeddingStore; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.Any; +import jakarta.enterprise.inject.Instance; +import jakarta.inject.Inject; +import org.apache.camel.Exchange; +import org.apache.camel.Expression; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService; +import org.apache.camel.support.builder.ExpressionBuilder; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.apache.camel.util.URISupport; +import org.jboss.logging.Logger; + +import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file; + +/** + * Generates one Camel route per configured ingestion pipeline. Users never see these routes — + * they are the implementation of the configuration. + */ +@ApplicationScoped +public class IngestRoutes extends RouteBuilder { + + private static final Logger LOG = Logger.getLogger(IngestRoutes.class); + + @Inject + IngestBuildTimeConfig buildTimeConfig; + + @Inject + IngestRunTimeConfig runTimeConfig; + + @Inject + IngestBuilderPipelines builderPipelines; + + @Inject + @Any + Instance<EmbeddingStore<TextSegment>> storeCandidates; + + @Inject + @Any + Instance<EmbeddingModel> modelCandidates; + + @Override + public void configure() { + // a pipeline may be declared entirely through runtime properties - the documented + // minimum is a directory and nothing else - so the two config roots are unioned. Keying + // off the build-time map alone would make that configuration a silent no-op, since + // SmallRye only materialises a map key for the mapping whose structure a property matches + Set<String> builderDeclared = builderPipelines.entries().stream() + .map(IngestBuilderPipelines.Entry::name) + .collect(Collectors.toSet()); + Set<String> names = new TreeSet<>(buildTimeConfig.pipelines().keySet()); + names.addAll(runTimeConfig.pipelines().keySet()); + names.removeAll(builderDeclared); + + for (String name : names) { + IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline = buildTimeConfig.pipelines().get(name); + IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(name); + + if (runtime != null && !runtime.enabled()) { + LOG.infof("Ingestion pipeline '%s' is disabled", name); + continue; + } + + IngestService service = new IngestService( + name, + resolveStore(name, pipeline == null ? null : pipeline.embeddingStore().orElse(null)), + resolveModel(name, pipeline == null ? null : pipeline.embeddingModel().orElse(null)), + pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE : pipeline.maxSegmentSize(), + pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize()); + + // a consumer URI says "consume from this"; its absence says "read that directory" + String uri = pipeline == null ? null : pipeline.source().uri().orElse(null); + if (uri != null && runtime != null && runtime.source().directory().isPresent()) { + throw new IllegalStateException("Ingestion pipeline '" + name + "' sets both source.uri ('" + uri + + "') and source.directory ('" + runtime.source().directory().get() + "'). A pipeline " + + "reads one source: keep the URI, or drop it to read the directory."); + } + if (uri == null) { + configureFileSource(name, runtime, service); + LOG.infof("Ingestion pipeline '%s': source=file", name); + } else { + configureEndpointSource(name, uri, runtime, service); + LOG.infof("Ingestion pipeline '%s': source=%s", name, URISupport.sanitizeUri(uri)); + } + } + + for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) { + configureBuilderPipeline(entry); + } + } + + /** An {@code @Ingest}-declared pipeline: the builder twin of the configuration path. */ + private void configureBuilderPipeline(IngestBuilderPipelines.Entry entry) { + String name = entry.name(); + // configuration can still switch a builder-declared pipeline off, and the check precedes + // the invocation so a disabled pipeline's method never runs + IngestRunTimeConfig.PipelineRunTimeConfig external = runTimeConfig.pipelines().get(name); + if (external != null && !external.enabled()) { + LOG.infof("Ingestion pipeline '%s' (builder) is disabled", name); + return; + } + // enabled is the one thing configuration may say about a builder pipeline; anything about + // its source would be quietly overruled by the @Ingest method, so it is an error instead + // (source.recursive cannot be told apart from its default, so it alone goes undetected - + // Source.recursive() is its builder twin) + if (external != null && (external.source().directory().isPresent() + || external.source().documentId().isPresent())) { + throw new IllegalStateException("Ingestion pipeline '" + name + "' is declared in Java, so its source " + + "comes from the @Ingest method. Remove quarkus.camel.ai.ingest." + name + ".source.* , or " + + "declare the pipeline in configuration instead."); + } + + IngestPipeline definition = builderPipelines.definition(entry); + IngestRunTimeConfig.PipelineRunTimeConfig runtime = definition.asRunTimeConfig(); + + IngestService service = new IngestService( + name, + resolveStore(name, definition.embeddingStoreName().orElse(null)), + resolveModel(name, definition.embeddingModelName().orElse(null)), + definition.maxSegmentSize(), + definition.maxOverlapSize()); + + switch (definition.sourceType()) { + case "file" -> configureFileSource(name, runtime, service); + case "endpoint" -> configureEndpointSource(name, definition.sourceUri(), runtime, service); + default -> throw new IllegalStateException("Unknown source type " + definition.sourceType()); + } + + LOG.infof("Ingestion pipeline '%s' (builder): source=%s", name, + URISupport.sanitizeUri(definition.sourceUri() == null ? definition.sourceType() : definition.sourceUri())); + } + + private void configureFileSource(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime, + IngestService service) { + String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null), + "source.directory"); + // built with the Endpoint DSL rather than concatenated: a directory containing ? # & or a + // space would otherwise mis-parse, and a crafted one could inject options - delete=true + // is honoured ahead of noop and would delete the user's documents after reading them. + // noop leaves the documents where they are (a knowledge base reads its source, it does + // not consume it), idempotent keeps the same file from being ingested twice, and the + // changed read lock waits for a file still being copied in rather than embedding half of + // it - the truncation would be permanent, since idempotent keys on the path. The register + // is sized explicitly: Camel's default caps at 1000 entries, and beyond that eviction + // would re-ingest a large directory steadily during normal operation, not just on restart + Expression documentId = documentIdExpression(runtime, Exchange.FILE_NAME); + from(file(directory) + .noop(true) + .idempotent(true) + .idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000)) + .recursive(runtime.source().recursive()) + .readLock("changed") + .charset(StandardCharsets.UTF_8.name())) + .routeId(routeId(name)) + .process(exchange -> service.ingest(documentId.evaluate(exchange, String.class), + exchange.getIn().getBody(String.class))); + } + + /** + * The escape hatch: any Camel consumer feeds the pipeline. Which part of the exchange + * identifies the document is the consumer's business, so {@code source.document-id} says it — + * {@code ${header.CamelAwsS3Key}} for an S3 consumer, the message header otherwise. + */ + private void configureEndpointSource(String name, String uri, + IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService service) { + Expression documentId = documentIdExpression(runtime, IngestHeaders.DOCUMENT_ID); + from(uri) + .routeId(routeId(name)) + .process(exchange -> { + String id = documentId.evaluate(exchange, String.class); + if (id == null) { + throw new IllegalArgumentException("Ingestion pipeline '" + name + "': no document id. " + + "Set the " + IngestHeaders.DOCUMENT_ID + " header, or point " + + "quarkus.camel.ai.ingest." + name + ".source.document-id at where the " + + "consumer puts it."); + } + exchange.getIn().setBody(service.ingest(id, exchange.getIn().getBody(String.class))); + }); + } + + private Expression documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime, + String defaultHeader) { + String configured = runtime == null ? null : runtime.source().documentId().orElse(null); + if (configured == null) { + return ExpressionBuilder.headerExpression(defaultHeader); + } + // a bare header name is read as a header directly rather than parsed: a dotted name such + // as a dotted header name sends the simple parser into OGNL, and ${...} in a properties file is + // MicroProfile Config expansion, which would consume an expression before Camel saw it + return configured.contains("${") ? simple(configured) : ExpressionBuilder.headerExpression(configured); + } + + private static String routeId(String name) { + return "camel-quarkus-ai-ingest-" + name; + } + + private static String required(String name, String value, String property) { + if (value == null) { + throw new IllegalStateException("Ingestion pipeline '" + name + "' has no " + property + + ". Set quarkus.camel.ai.ingest." + name + "." + property); + } + return value; + } + + private EmbeddingStore<TextSegment> resolveStore(String name, String configured) { Review Comment: Reopening this one — the change works for the *named* path but breaks the unnamed one. `getRegistry().findByType(EmbeddingStore.class)` reaches `RuntimeBeanRepository.findByType`, which is `Arc.container().listAll(type)`. That is a raw required type, and CDI treats a parameterized bean type as assignable to a raw one only when its type parameters are unbounded variables or `Object` — so a bean typed `EmbeddingStore<TextSegment>` never matches. On this branch: ``` registry.findByType(EmbeddingStore) = 0 registry.findByTypeWithName(EmbeddingStore) = [] Arc.listAll(EmbeddingStore) = 0 Arc.listAll(EmbeddingModel) = 1 <- not generic, so it does match ``` with `@Named("store")` set and `quarkus.arc.remove-unused-beans=false`, so neither naming nor bean removal is the variable. Separately, the `@Inject @Any Instance<...>` fields that went away were the injection points keeping unnamed store and model beans unremovable. Even for the non-generic `EmbeddingModel`, an unnamed producer is now removed as unused; it resolves again with `quarkus.arc.remove-unused-beans=false`. Net effect: a configuration-declared pipeline with exactly one unnamed store and model — the arrangement `usage.adoc` documents — starts at `dd193707` and fails at `2541fc3a` with `needs an embedding store, but no bean of that type exists`. Going the other way would fix both at once: keep the `Instance` fields and resolve the named case through them, `candidates.select(NamedLiteral.of(configured))`. If the registry is the mechanism you would rather keep, it needs an `UnremovableBeanBuildItem` for both types and a `TypeLiteral<EmbeddingStore<TextSegment>>` lookup for the store. Worth a test either way — `IngestNoStoreBeanTest` asserts the "no bean of that type" message, so it currently passes whether or not resolution works. ########## extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java: ########## @@ -0,0 +1,85 @@ +/* + * 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.quarkus.component.langchain4j.ingest; + +import java.util.Map; +import java.util.Optional; + +import io.quarkus.runtime.annotations.ConfigDocMapKey; +import io.quarkus.runtime.annotations.ConfigPhase; +import io.quarkus.runtime.annotations.ConfigRoot; +import io.smallrye.config.ConfigMapping; +import io.smallrye.config.WithDefault; +import io.smallrye.config.WithParentName; + +/** + * Runtime configuration of ingestion pipelines: concrete locations and switches that may differ + * per deployment. The pipeline topology is build-time, see {@link IngestBuildTimeConfig}. + */ +@ConfigMapping(prefix = "quarkus.camel.ai.ingest") +@ConfigRoot(phase = ConfigPhase.RUN_TIME) +public interface IngestRunTimeConfig { + + /** + * Ingestion pipelines by name. + */ + @WithParentName + @ConfigDocMapKey("pipeline-name") + Map<String, PipelineRunTimeConfig> pipelines(); + + interface PipelineRunTimeConfig { + + /** + * Whether this pipeline starts. Useful to switch ingestion off in dev mode. + */ + @WithDefault("true") + boolean enabled(); + + /** + * The document source. + */ + SourceRunTimeConfig source(); + + interface SourceRunTimeConfig { + + /** + * The directory to ingest documents from, for a pipeline that has no `source.uri`. A + * path is a deployment concern, so unlike the URI it stays runtime configuration. + * Setting both is an error. + */ + Optional<String> directory(); + + /** + * Whether subdirectories are ingested too, when reading a directory. + */ + @WithDefault("true") + boolean recursive(); + + /** + * Where the document id lives in the exchange the consumer delivers: normally the + * name of a header, such as `CamelAwsS3Key` for an S3 consumer or `CamelKafkaKey` for + * a Kafka one. A value containing a dollar-brace placeholder is taken as a Review Comment: `$simple{...}` is the right call, and the end-to-end coverage on the `custom` pipeline is good. One narrowing though: the check is `configured.startsWith("$simple{")`, while Camel's own `LanguageSupport.hasSimpleFunction` uses `contains` for both tokens (`SIMPLE_FUNCTION_START = {"${", "$simple{"}`). So `id-$simple{header.Foo}` matches neither branch here and is silently taken as a header name, even though Camel would happily parse it. The `${` branch already uses `contains`, so the two halves disagree with each other too — `contains("$simple{")` lines both up with Camel. ########## extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java: ########## @@ -0,0 +1,224 @@ +/* + * 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.quarkus.component.langchain4j.ingest.deployment; + +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import io.quarkus.arc.deployment.AdditionalBeanBuildItem; +import io.quarkus.arc.deployment.SyntheticBeanBuildItem; +import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem; +import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem; +import io.quarkus.deployment.annotations.BuildProducer; +import io.quarkus.deployment.annotations.BuildStep; +import io.quarkus.deployment.annotations.Consume; +import io.quarkus.deployment.annotations.ExecutionTime; +import io.quarkus.deployment.annotations.Record; +import io.quarkus.deployment.builditem.ApplicationArchivesBuildItem; +import io.quarkus.deployment.builditem.CombinedIndexBuildItem; +import io.quarkus.deployment.builditem.FeatureBuildItem; +import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem; +import io.quarkus.runtime.configuration.ConfigurationException; +import jakarta.inject.Singleton; +import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes; +import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder; +import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem; +import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem; +import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem; +import org.apache.camel.quarkus.core.deployment.util.CamelSupport; +import org.apache.camel.quarkus.core.deployment.util.PathFilter; +import org.apache.camel.util.URISupport; +import org.jboss.jandex.AnnotationInstance; +import org.jboss.jandex.AnnotationTarget; +import org.jboss.jandex.DotName; +import org.jboss.jandex.MethodInfo; + +class Langchain4jIngestProcessor { + + private static final String FEATURE = "camel-langchain4j-ingest"; + + @BuildStep + FeatureBuildItem feature() { + return new FeatureBuildItem(FEATURE); + } + + @BuildStep + AdditionalBeanBuildItem beans() { + return AdditionalBeanBuildItem.builder() + .addBeanClasses(IngestRoutes.class) + .setUnremovable() + .build(); + } + + /** + * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters, + * unique names, no collision with configuration-declared pipelines), invoked reflectively once + * at startup. + */ + @BuildStep + @Record(ExecutionTime.STATIC_INIT) + void discoverBuilderPipelines( + CombinedIndexBuildItem combinedIndex, + IngestBuildTimeConfig config, + Langchain4jIngestRecorder recorder, + BuildProducer<AdditionalBeanBuildItem> beans, + BuildProducer<ReflectiveClassBuildItem> reflectiveClasses, + BuildProducer<SyntheticBeanBuildItem> syntheticBeans) { + + DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName()); + DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName()); + + List<String> flatEntries = new ArrayList<>(); + Set<String> names = new HashSet<>(); + Set<String> beanClasses = new HashSet<>(); + + for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) { + if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) { + continue; + } + MethodInfo method = annotation.target().asMethod(); + String name = annotation.value().asString(); + String location = method.declaringClass().name() + "#" + method.name(); + + if (name.isBlank()) { + throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name"); + } + if (!method.returnType().name().equals(pipelineType)) { + throw new ConfigurationException("@Ingest method " + location + " must return " + + IngestPipeline.class.getSimpleName()); + } + if (!method.parameters().isEmpty()) { + throw new ConfigurationException("@Ingest method " + location + " must take no parameters"); + } + // the method is invoked on a CDI bean instance, which a static method would bypass + // and a private one would run against the client proxy, seeing null injected fields + if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) { + throw new ConfigurationException("@Ingest method " + location + " must not be private or static"); Review Comment: Thanks — the check and the test do what I was after. One small over-reach: it rejects `final` on pseudo-scoped beans too. `@Singleton` and `@Dependent` get no client proxy, so a final method there is harmless. Narrowing the check to normal-scoped declaring beans would keep the message for the case that actually breaks. Not a blocker, and erring strict is defensible — just noting the false positive. ########## extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java: ########## @@ -0,0 +1,224 @@ +/* + * 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.quarkus.component.langchain4j.ingest.deployment; + +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import io.quarkus.arc.deployment.AdditionalBeanBuildItem; +import io.quarkus.arc.deployment.SyntheticBeanBuildItem; +import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem; +import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem; +import io.quarkus.deployment.annotations.BuildProducer; +import io.quarkus.deployment.annotations.BuildStep; +import io.quarkus.deployment.annotations.Consume; +import io.quarkus.deployment.annotations.ExecutionTime; +import io.quarkus.deployment.annotations.Record; +import io.quarkus.deployment.builditem.ApplicationArchivesBuildItem; +import io.quarkus.deployment.builditem.CombinedIndexBuildItem; +import io.quarkus.deployment.builditem.FeatureBuildItem; +import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem; +import io.quarkus.runtime.configuration.ConfigurationException; +import jakarta.inject.Singleton; +import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline; +import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes; +import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder; +import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem; +import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem; +import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem; +import org.apache.camel.quarkus.core.deployment.util.CamelSupport; +import org.apache.camel.quarkus.core.deployment.util.PathFilter; +import org.apache.camel.util.URISupport; +import org.jboss.jandex.AnnotationInstance; +import org.jboss.jandex.AnnotationTarget; +import org.jboss.jandex.DotName; +import org.jboss.jandex.MethodInfo; + +class Langchain4jIngestProcessor { + + private static final String FEATURE = "camel-langchain4j-ingest"; + + @BuildStep + FeatureBuildItem feature() { + return new FeatureBuildItem(FEATURE); + } + + @BuildStep + AdditionalBeanBuildItem beans() { + return AdditionalBeanBuildItem.builder() + .addBeanClasses(IngestRoutes.class) + .setUnremovable() + .build(); + } + + /** + * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters, + * unique names, no collision with configuration-declared pipelines), invoked reflectively once + * at startup. + */ + @BuildStep + @Record(ExecutionTime.STATIC_INIT) + void discoverBuilderPipelines( + CombinedIndexBuildItem combinedIndex, + IngestBuildTimeConfig config, + Langchain4jIngestRecorder recorder, + BuildProducer<AdditionalBeanBuildItem> beans, + BuildProducer<ReflectiveClassBuildItem> reflectiveClasses, + BuildProducer<SyntheticBeanBuildItem> syntheticBeans) { + + DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName()); + DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName()); + + List<String> flatEntries = new ArrayList<>(); + Set<String> names = new HashSet<>(); + Set<String> beanClasses = new HashSet<>(); + + for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) { + if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) { + continue; + } + MethodInfo method = annotation.target().asMethod(); + String name = annotation.value().asString(); + String location = method.declaringClass().name() + "#" + method.name(); + + if (name.isBlank()) { + throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name"); + } + if (!method.returnType().name().equals(pipelineType)) { + throw new ConfigurationException("@Ingest method " + location + " must return " + + IngestPipeline.class.getSimpleName()); + } + if (!method.parameters().isEmpty()) { + throw new ConfigurationException("@Ingest method " + location + " must take no parameters"); + } + // the method is invoked on a CDI bean instance, which a static method would bypass + // and a private one would run against the client proxy, seeing null injected fields + if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) { + throw new ConfigurationException("@Ingest method " + location + " must not be private or static"); + } + if (!names.add(name) || config.pipelines().containsKey(name)) { + throw new ConfigurationException("Ingestion pipeline '" + name + "' is declared more than once " + + "(builder and/or configuration). Pipeline names must be unique."); + } + + flatEntries.add(name); + flatEntries.add(method.declaringClass().name().toString()); + flatEntries.add(method.name()); + beanClasses.add(method.declaringClass().name().toString()); + } + + if (!beanClasses.isEmpty()) { + beans.produce(AdditionalBeanBuildItem.builder() + .addBeanClasses(beanClasses.toArray(new String[0])) + .setUnremovable() + .build()); + reflectiveClasses.produce(ReflectiveClassBuildItem.builder(beanClasses.toArray(new String[0])) + .methods() + .build()); + } + + syntheticBeans.produce(SyntheticBeanBuildItem.configure(IngestBuilderPipelines.class) + .scope(Singleton.class) + .unremovable() + .runtimeValue(recorder.createBuilderPipelines(flatEntries)) + .done()); + } + + /** + * The pre-start half of the component-presence check: recorded as a Camel runtime task, it + * runs after ArC is fully initialised but before the Camel runtime is assembled — and thus + * before Camel Main binds {@code camel.component.*} properties, whose failure for a missing + * component would otherwise preempt the friendlier add-extension hint. + */ + @BuildStep + @Record(ExecutionTime.RUNTIME_INIT) + @Consume(SyntheticBeansRuntimeInitBuildItem.class) + CamelRuntimeTaskBuildItem checkComponentsPresent(Langchain4jIngestRecorder recorder, + CamelContextBuildItem camelContext) { + recorder.checkComponentsPresent(camelContext.getCamelContext()); + return new CamelRuntimeTaskBuildItem("langchain4j-ingest-components"); + } + + /** + * A pipeline whose consumer URI names a component that is not on the classpath stops the + * build, with the command that fixes it rather than a startup failure. Only configured URIs + * can be checked: a builder-declared pipeline composes its URI at startup, where the pre-start + * task above applies the same hint. Component services are REGISTRY-destination, so they are + * read from the application archives directly — they never appear among the DISCOVERY + * {@code CamelServiceBuildItem}s. + */ + @BuildStep + void validateConnectorsPresent(IngestBuildTimeConfig config, ApplicationArchivesBuildItem applicationArchives, + BuildProducer<ValidationErrorBuildItem> validationErrors) { + PathFilter pathFilter = new PathFilter.Builder() + .include("META-INF/services/org/apache/camel/component/*") + .build(); + Set<String> components = CamelSupport.services(applicationArchives, pathFilter) + .map(CamelServiceBuildItem::getName) + .collect(Collectors.toSet()); + + for (Map.Entry<String, IngestBuildTimeConfig.PipelineBuildTimeConfig> entry : config.pipelines().entrySet()) { + String uri = entry.getValue().source().uri().orElse(null); + if (uri == null) { + continue; + } + int colon = uri.indexOf(':'); + String scheme = colon < 1 ? uri : uri.substring(0, colon); + // a placeholder resolves at startup, so its scheme cannot be known here + if (scheme.contains("{{") || scheme.contains("$")) { + continue; + } + if (!components.contains(scheme)) { + // the URI is sanitized: a consumer URI may legitimately carry credentials, and a + // build log is no place for them. The artifact hint is a heuristic - multi-scheme + // components (smtp -> camel-quarkus-mail) name their extension differently + validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException( + "Ingestion pipeline '" + entry.getKey() + "' consumes from '" + + URISupport.sanitizeUri(uri) + "', but the Camel component '" + scheme + + "' is not on the classpath.\nAdd the extension that provides it, usually:" + + " ./mvnw quarkus:add-extension -Dextensions=camel-quarkus-" + scheme))); Review Comment: The message reads well now, in both checks. The class javadoc on `IngestComponentPresence` still describes the old form though — "Fails a pipeline whose consumer component is not on the classpath with the same add-extension hint the build gives for configured URIs". ########## integration-tests/langchain4j-ingest/pom.xml: ########## @@ -0,0 +1,183 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + + 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. + +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-build-parent-it</artifactId> + <version>3.39.0-SNAPSHOT</version> + <relativePath>../../poms/build-parent-it/pom.xml</relativePath> + </parent> + + <artifactId>camel-quarkus-integration-test-langchain4j-ingest</artifactId> + <name>Camel Quarkus :: Integration Tests :: LangChain4j Ingest</name> + <description>Integration tests for the declarative AI document ingestion extension</description> + + <dependencies> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-langchain4j-ingest</artifactId> + </dependency> + <dependency> + <groupId>dev.langchain4j</groupId> + <artifactId>langchain4j</artifactId> + </dependency> + <dependency> + <groupId>io.quarkus</groupId> + <artifactId>quarkus-rest</artifactId> + </dependency> + <dependency> + <groupId>io.quarkus</groupId> + <artifactId>quarkus-rest-jackson</artifactId> + </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-direct</artifactId> + </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-aws2-s3</artifactId> + </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-kafka</artifactId> + </dependency> + + <!-- test dependencies --> + <dependency> + <groupId>io.quarkus</groupId> + <artifactId>quarkus-junit</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>io.rest-assured</groupId> + <artifactId>rest-assured</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-integration-tests-support-kafka</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.testcontainers</groupId> + <artifactId>testcontainers</artifactId> + <scope>test</scope> + </dependency> + </dependencies> + + <profiles> + <profile> + <id>native</id> + <activation> + <property> + <name>native</name> + </property> + </activation> + <properties> + <quarkus.native.enabled>true</quarkus.native.enabled> + </properties> + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-failsafe-plugin</artifactId> + <executions> + <execution> + <goals> + <goal>integration-test</goal> + <goal>verify</goal> + </goals> + </execution> + </executions> + </plugin> + </plugins> + </build> + </profile> + <profile> + <id>virtualDependencies</id> + <activation> + <property> + <name>!noVirtualDependencies</name> + </property> + </activation> + <dependencies> + <!-- The following dependencies guarantee that this module is built after them. You can update them by running `mvn process-resources -Pformat -N` from the source tree root directory --> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-aws2-s3-deployment</artifactId> + <version>${project.version}</version> + <type>pom</type> + <scope>test</scope> + <exclusions> + <exclusion> + <groupId>*</groupId> + <artifactId>*</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-direct-deployment</artifactId> + <version>${project.version}</version> + <type>pom</type> + <scope>test</scope> + <exclusions> + <exclusion> + <groupId>*</groupId> + <artifactId>*</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-kafka-deployment</artifactId> + <version>${project.version}</version> + <type>pom</type> + <scope>test</scope> + <exclusions> + <exclusion> + <groupId>*</groupId> + <artifactId>*</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-langchain4j-ingest-deployment</artifactId> + <version>${project.version}</version> + <type>pom</type> + <scope>test</scope> + <exclusions> + <exclusion> + <groupId>*</groupId> + <artifactId>*</artifactId> + </exclusion> + </exclusions> + </dependency> + </dependencies> + </profile> + </profiles> Review Comment: Confirmed locally: `mvn verify -Dskip-testcontainers-tests` on this module runs the four non-container tests, starts no container, and passes. The surefire-only exclusion is fine as things stand — failsafe is bound only in the `native` profile, and no workflow combines `-Dnative` with `-Dskip-testcontainers-tests`, so the `*IT` classes never run on those jobs. Worth keeping in mind if one ever does, since the 74 modules using `<skipTests>true</skipTests>` would be covered there and this one would not. -- 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]
