This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 0e2a286dc6de CAMEL-26118: camel-flowable - a producer created before
its channel is deployed sends once the channel is deployed, and fails with a
clear error until then (#28895)
0e2a286dc6de is described below
commit 0e2a286dc6de03faf36caa5b5e03efdc7be3089b
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:37:06 2026 +0100
CAMEL-26118: camel-flowable - a producer created before its channel is
deployed sends once the channel is deployed, and fails with a clear error until
then (#28895)
…s deployed sends once the channel is deployed, and fails with a clear
error until then
A route that sends to flowable:channelKey and starts before the channel
model is
deployed got a bare endpoint without the channel and the event registry
configuration, so every send failed with a NullPointerException, also after
the
channel was deployed. The producer now looks up the endpoint that the
channel
model processor registers on deployment, and fails with a FlowableException
naming the channel key while there is none. The doc gave the URI format as
flowable-event://; it is flowable:channelKey[?options].
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/catalog/docs/flowable-component.adoc | 8 +-
.../src/main/docs/flowable-component.adoc | 8 +-
.../camel/component/flowable/FlowableEndpoint.java | 8 ++
.../camel/component/flowable/FlowableProducer.java | 26 ++++++-
.../FlowableProducerChannelNotDeployedTest.java | 85 ++++++++++++++++++++++
.../resources/channel/userRouteChannel.channel | 10 +++
6 files changed, 142 insertions(+), 3 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flowable-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flowable-component.adoc
index 416941624306..55ecbd0c62bb 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flowable-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flowable-component.adoc
@@ -17,7 +17,7 @@ Communicate with the Flowable Event Registry for sending and
receiving messages
== URI format
------------------------------
-flowable-event://[channelKey]?[options]
+flowable:channelKey[?options]
------------------------------
Where *channelKey* is the unique key value of the channel model deployed in
the Flowable Event Registry.
@@ -35,6 +35,12 @@ include::partial$component-endpoint-headers.adoc[]
The body content is defined by the event model deployed to the Flowable Event
Registry.
The event message can then be mapped to process or case variables in the
process BPMN or case CMMN model.
+=== Sending to a channel that is not deployed yet
+
+A route that sends to `flowable:channelKey` can start before the inbound
channel model with that key is deployed
+in the Flowable Event Registry. Once the channel is deployed, the producer
sends to it.
+Until then, every send fails with a `FlowableException` that names the channel
key.
+
== Examples
The following example sends an event to the Flowable event registry for
further processing:
diff --git a/components/camel-flowable/src/main/docs/flowable-component.adoc
b/components/camel-flowable/src/main/docs/flowable-component.adoc
index 416941624306..55ecbd0c62bb 100644
--- a/components/camel-flowable/src/main/docs/flowable-component.adoc
+++ b/components/camel-flowable/src/main/docs/flowable-component.adoc
@@ -17,7 +17,7 @@ Communicate with the Flowable Event Registry for sending and
receiving messages
== URI format
------------------------------
-flowable-event://[channelKey]?[options]
+flowable:channelKey[?options]
------------------------------
Where *channelKey* is the unique key value of the channel model deployed in
the Flowable Event Registry.
@@ -35,6 +35,12 @@ include::partial$component-endpoint-headers.adoc[]
The body content is defined by the event model deployed to the Flowable Event
Registry.
The event message can then be mapped to process or case variables in the
process BPMN or case CMMN model.
+=== Sending to a channel that is not deployed yet
+
+A route that sends to `flowable:channelKey` can start before the inbound
channel model with that key is deployed
+in the Flowable Event Registry. Once the channel is deployed, the producer
sends to it.
+Until then, every send fails with a `FlowableException` that names the channel
key.
+
== Examples
The following example sends an event to the Flowable event registry for
further processing:
diff --git
a/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableEndpoint.java
b/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableEndpoint.java
index 2d770ba2a704..f927ae3edceb 100644
---
a/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableEndpoint.java
+++
b/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableEndpoint.java
@@ -124,6 +124,10 @@ public class FlowableEndpoint extends DefaultEndpoint {
this.channelKey = channelKey;
}
+ public CamelInboundChannelModel getCamelInboundChannel() {
+ return camelInboundChannel;
+ }
+
public void setCamelInboundChannel(CamelInboundChannelModel
camelInboundChannel) {
this.camelInboundChannel = camelInboundChannel;
}
@@ -132,6 +136,10 @@ public class FlowableEndpoint extends DefaultEndpoint {
this.camelOutboundChannel = camelOutboundChannel;
}
+ public EventRegistryEngineConfiguration
getEventRegistryEngineConfiguration() {
+ return eventRegistryEngineConfiguration;
+ }
+
public void
setEventRegistryEngineConfiguration(EventRegistryEngineConfiguration
eventRegistryEngineConfiguration) {
this.eventRegistryEngineConfiguration =
eventRegistryEngineConfiguration;
}
diff --git
a/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableProducer.java
b/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableProducer.java
index d58eff3e8ec9..d73d7e8d98f2 100644
---
a/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableProducer.java
+++
b/components/camel-flowable/src/main/java/org/apache/camel/component/flowable/FlowableProducer.java
@@ -20,6 +20,7 @@ import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.support.DefaultProducer;
+import org.flowable.common.engine.api.FlowableException;
import org.flowable.eventregistry.impl.DefaultInboundEvent;
import org.flowable.eventregistry.impl.EventRegistryEngineConfiguration;
import org.flowable.eventregistry.model.CamelInboundChannelModel;
@@ -39,8 +40,31 @@ public class FlowableProducer extends DefaultProducer {
@Override
public void process(Exchange exchange) throws Exception {
+ CamelInboundChannelModel channel = camelInboundChannel;
+ EventRegistryEngineConfiguration configuration =
eventRegistryEngineConfiguration;
+ if (channel == null || configuration == null) {
+ // this producer was created before the channel was deployed (for
example a route that starts before the
+ // channel model), so use the endpoint that the channel model
processor registered when it was deployed
+ FlowableEndpoint deployed = lookupDeployedEndpoint();
+ channel = deployed.getCamelInboundChannel();
+ configuration = deployed.getEventRegistryEngineConfiguration();
+ }
+
Message message = exchange.getMessage();
DefaultInboundEvent inboundEvent = new
DefaultInboundEvent(message.getBody(), message.getHeaders());
-
eventRegistryEngineConfiguration.getEventRegistry().eventReceived(camelInboundChannel,
inboundEvent);
+ configuration.getEventRegistry().eventReceived(channel, inboundEvent);
+ }
+
+ private FlowableEndpoint lookupDeployedEndpoint() {
+ Endpoint endpoint =
getEndpoint().getCamelContext().hasEndpoint(getEndpoint().getEndpointUri());
+ if (endpoint instanceof FlowableEndpoint flowableEndpoint &&
flowableEndpoint.getCamelInboundChannel() != null
+ && flowableEndpoint.getEventRegistryEngineConfiguration() !=
null) {
+ return flowableEndpoint;
+ }
+ String channelKey = getEndpoint() instanceof FlowableEndpoint
flowableEndpoint
+ && flowableEndpoint.getChannelKey() != null ?
flowableEndpoint.getChannelKey() : getEndpoint().getEndpointUri();
+ throw new FlowableException(
+ "No inbound channel with key " + channelKey + " is deployed in
the Flowable Event Registry, so "
+ + getEndpoint().getEndpointUri() + "
cannot send the event");
}
}
diff --git
a/components/camel-flowable/src/test/java/org/apache/camel/component/flowable/FlowableProducerChannelNotDeployedTest.java
b/components/camel-flowable/src/test/java/org/apache/camel/component/flowable/FlowableProducerChannelNotDeployedTest.java
new file mode 100644
index 000000000000..d4874d28aea1
--- /dev/null
+++
b/components/camel-flowable/src/test/java/org/apache/camel/component/flowable/FlowableProducerChannelNotDeployedTest.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.component.flowable;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.flowable.common.engine.api.FlowableException;
+import org.flowable.engine.runtime.ProcessInstance;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * A route that sends to a channel starts before the channel model is deployed
in the Flowable Event Registry.
+ */
+public class FlowableProducerChannelNotDeployedTest extends
CamelFlowableTestCase {
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:send").to("flowable:userRouteChannel");
+ }
+ };
+ }
+
+ @Test
+ public void testSendBeforeChannelIsDeployed() {
+ Exchange exchange = template.send("direct:send", e ->
e.getIn().setBody(userBody()));
+
+ FlowableException cause = assertInstanceOf(FlowableException.class,
exchange.getException());
+ assertEquals("No inbound channel with key userRouteChannel is deployed
in the Flowable Event Registry,"
+ + " so flowable:userRouteChannel cannot send the event",
+ cause.getMessage());
+ }
+
+ @Test
+ public void testSendAfterChannelIsDeployed() {
+ String deploymentId =
deployProcessDefinition("process/start.bpmn20.xml");
+ try {
+
eventRegistryEngineConfiguration.getEventRepositoryService().createDeployment()
+ .addClasspathResource("channel/userRouteChannel.channel")
+ .addClasspathResource("event/userEvent.event")
+ .deploy();
+
+ Exchange exchange = template.send("direct:send", e ->
e.getIn().setBody(userBody()));
+ assertNull(exchange.getException());
+
+ ProcessInstance processInstance
+ =
runtimeService.createProcessInstanceQuery().processDefinitionKey("camelProcess").singleResult();
+ assertNotNull(processInstance);
+ assertEquals("John Doe",
runtimeService.getVariable(processInstance.getId(), "name"));
+ assertEquals(23,
runtimeService.getVariable(processInstance.getId(), "age"));
+ } finally {
+ repositoryService.deleteDeployment(deploymentId, true);
+ }
+ }
+
+ private static ObjectNode userBody() {
+ ObjectNode bodyNode = new ObjectMapper().createObjectNode();
+ bodyNode.put("name", "John Doe");
+ bodyNode.put("age", 23);
+ return bodyNode;
+ }
+}
diff --git
a/components/camel-flowable/src/test/resources/channel/userRouteChannel.channel
b/components/camel-flowable/src/test/resources/channel/userRouteChannel.channel
new file mode 100644
index 000000000000..4d95da0c75a0
--- /dev/null
+++
b/components/camel-flowable/src/test/resources/channel/userRouteChannel.channel
@@ -0,0 +1,10 @@
+{
+ "name": "Camel user route channel",
+ "key": "userRouteChannel",
+ "channelType": "inbound",
+ "type": "camel",
+ "deserializerType": "json",
+ "channelEventKeyDetection": {
+ "fixedValue": "userEvent"
+ }
+}
