From ab5d656affbdb9139327069a01d6a0ab2e295e2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Fri, 23 Oct 2026 23:16:54 +0300 Subject: [PATCH 01/10] CAMEL-25617: camel-plc4x - Lock the triggered consumer behaviour with route tests The test MockDriver can now be programmed per tag address with the values successive reads return (the last value repeats), and can refuse connections. Requests whose tags are not programmed keep the previous deep stub behaviour, so the existing component, tag and pollEnrich tests are unchanged. The driver prepares every tag as a boolean tag, which the scraper needs to validate the trigger. Plc4XConsumerRouteTest drives plc4x:mock: routes and asserts only what arrives at mock endpoints: - one exchange per rising edge of the trigger (false, false, true, true, false, true gives exactly two); - the body is a map of the data tags, without the trigger; - a trigger that is already true on the first read emits once; - a stopped route delivers nothing, and a restarted route delivers again without duplicates; - a consumer without a trigger reads the tags once; - refused connections neither fail the route start nor log a warning per period. These tests replace Plc4XConsumerTest and Plc4XConsumerStopTest, which mocked the endpoint and captured the scraper's constructors with Mockito. Co-Authored-By: Claude Fable 5.1 --- .../camel/component/plc4x/MockDriver.java | 124 +++++++++- .../plc4x/Plc4XConsumerRouteTest.java | 215 ++++++++++++++++++ .../plc4x/Plc4XConsumerStopTest.java | 67 ------ .../component/plc4x/Plc4XConsumerTest.java | 90 -------- 4 files changed, 338 insertions(+), 158 deletions(-) create mode 100644 components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java delete mode 100644 components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerStopTest.java delete mode 100644 components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerTest.java diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java index 71815047e3a75..c83e162854e59 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java @@ -16,20 +16,31 @@ */ package org.apache.camel.component.plc4x; +import java.util.Arrays; import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.plc4x.java.api.PlcConnection; import org.apache.plc4x.java.api.PlcDriver; import org.apache.plc4x.java.api.authentication.PlcAuthentication; import org.apache.plc4x.java.api.exceptions.PlcConnectionException; import org.apache.plc4x.java.api.messages.PlcReadRequest; +import org.apache.plc4x.java.api.messages.PlcReadResponse; import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest; import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest; import org.apache.plc4x.java.api.messages.PlcWriteRequest; +import org.apache.plc4x.java.api.model.PlcTag; +import org.apache.plc4x.java.api.types.PlcResponseCode; +import org.apache.plc4x.java.api.types.PlcValueType; import org.apache.plc4x.java.spi.messages.DefaultPlcSubscriptionResponse; import org.apache.plc4x.java.spi.messages.PlcSubscriber; import org.slf4j.Logger; @@ -37,12 +48,55 @@ import static org.mockito.Mockito.*; +/** + * Test driver for the {@code mock} protocol. A read returns the values a test programmed with + * {@link #programRead(String, Object...)} when every tag in the request is programmed, and Mockito deep stubs + * otherwise. The programming is static because every driver manager loads its own driver instance; tests call + * {@link #reset()}. + */ public class MockDriver implements PlcDriver { public static final Logger LOGGER = LoggerFactory.getLogger(MockDriver.class); + private static final Map PROGRAMMED_TAGS = new ConcurrentHashMap<>(); + private static final AtomicInteger REFUSED_CONNECTIONS = new AtomicInteger(); + private static volatile boolean refuseConnections; + ExecutorService executorService = Executors.newFixedThreadPool(10); + /** + * Programs the values that successive reads of a tag address return; the last value repeats once they are used up. + */ + public static void programRead(String tagAddress, Object... values) { + if (values.length == 0) { + throw new IllegalArgumentException("At least one value is required for " + tagAddress); + } + PROGRAMMED_TAGS.put(tagAddress, new ProgrammedTag(Arrays.asList(values))); + } + + /** + * Makes every following connection attempt fail. + */ + public static void refuseConnections() { + refuseConnections = true; + } + + /** + * @return the number of connection attempts refused since the last {@link #reset()} + */ + public static int refusedConnections() { + return REFUSED_CONNECTIONS.get(); + } + + /** + * Forgets all programmed tag values and accepts connections again. + */ + public static void reset() { + PROGRAMMED_TAGS.clear(); + refuseConnections = false; + REFUSED_CONNECTIONS.set(0); + } + @Override public String getProtocolCode() { return "mock"; @@ -55,11 +109,15 @@ public String getProtocolName() { @Override public PlcConnection getConnection(String url) throws PlcConnectionException { + if (refuseConnections) { + REFUSED_CONNECTIONS.incrementAndGet(); + throw new PlcConnectionException("Connection refused by the mock driver"); + } // Mock a connection. PlcConnection plcConnectionMock = mock(PlcConnection.class, RETURNS_DEEP_STUBS); when(plcConnectionMock.getMetadata().isReadSupported()).thenReturn(true); when(plcConnectionMock.getMetadata().isWriteSupported()).thenReturn(true); - when(plcConnectionMock.readRequestBuilder()).thenReturn(mock(PlcReadRequest.Builder.class, RETURNS_DEEP_STUBS)); + when(plcConnectionMock.readRequestBuilder()).thenAnswer(invocation -> readRequestBuilder()); when(plcConnectionMock.writeRequestBuilder()).thenReturn(mock(PlcWriteRequest.Builder.class, RETURNS_DEEP_STUBS)); when(plcConnectionMock.subscriptionRequestBuilder()) .thenReturn(mock(PlcSubscriptionRequest.Builder.class, RETURNS_DEEP_STUBS)); @@ -103,4 +161,68 @@ public PlcConnection getConnection(String url, PlcAuthentication authentication) return getConnection(null); } + /** + * Every tag is a boolean tag. Only the scraper prepares tags, to validate its trigger. + */ + @Override + public PlcTag prepareTag(String tagAddress) { + return mock(PlcTag.class, invocation -> switch (invocation.getMethod().getName()) { + case "getAddressString" -> tagAddress; + case "getPlcValueType" -> PlcValueType.BOOL; + default -> RETURNS_DEFAULTS.answer(invocation); + }); + } + + private static PlcReadRequest.Builder readRequestBuilder() { + PlcReadRequest.Builder unprogrammed = mock(PlcReadRequest.Builder.class, RETURNS_DEEP_STUBS); + Map tags = new LinkedHashMap<>(); + return mock(PlcReadRequest.Builder.class, invocation -> switch (invocation.getMethod().getName()) { + case "addTagAddress" -> { + tags.put(invocation.getArgument(0), invocation.getArgument(1)); + unprogrammed.addTagAddress(invocation.getArgument(0), invocation.getArgument(1)); + yield invocation.getMock(); + } + case "build" -> !tags.isEmpty() && PROGRAMMED_TAGS.keySet().containsAll(tags.values()) + ? programmedRequest(tags) : unprogrammed.build(); + default -> RETURNS_DEFAULTS.answer(invocation); + }); + } + + private static PlcReadRequest programmedRequest(Map tags) { + return mock(PlcReadRequest.class, invocation -> switch (invocation.getMethod().getName()) { + case "execute" -> CompletableFuture.completedFuture(read(tags)); + case "getTagNames" -> new LinkedHashSet<>(tags.keySet()); + default -> RETURNS_DEFAULTS.answer(invocation); + }); + } + + private static PlcReadResponse read(Map tags) { + Map values = new LinkedHashMap<>(); + tags.forEach((name, address) -> values.put(name, PROGRAMMED_TAGS.get(address).next())); + return mock(PlcReadResponse.class, invocation -> switch (invocation.getMethod().getName()) { + case "getTagNames" -> List.copyOf(values.keySet()); + case "getObject" -> values.get(invocation. getArgument(0)); + case "getResponseCode" -> PlcResponseCode.OK; + default -> RETURNS_DEFAULTS.answer(invocation); + }); + } + + private static final class ProgrammedTag { + + private final List values; + private int next; + + ProgrammedTag(List values) { + this.values = values; + } + + synchronized Object next() { + Object value = values.get(next); + if (next < values.size() - 1) { + next++; + } + return value; + } + } + } diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java new file mode 100644 index 0000000000000..eedc51deb7f2f --- /dev/null +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java @@ -0,0 +1,215 @@ +/* + * 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.plc4x; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The consumer's contract at the route seam: a route reads a {@code plc4x:mock:} endpoint whose tag values are + * programmed in {@link MockDriver}, and only what arrives at the mock endpoint is asserted. + */ +public class Plc4XConsumerRouteTest extends CamelTestSupport { + + private static final String PLC = "plc4x:mock:10.10.10.1/1/1?autoReconnect=true"; + private static final String TRIGGERED = PLC + "&trigger=TRIG&period=100&tag.temperature=TEMP&tag.pressure=PRES"; + private static final String UNTRIGGERED = PLC + "&tag.temperature=TEMP&tag.pressure=PRES"; + private static final String REFUSED = PLC + "&trigger=TRIG&period=50&tag.temperature=TEMP&tag.pressure=PRES"; + + @BeforeEach + void resetMockDriver() { + MockDriver.reset(); + } + + @AfterAll + static void leaveMockDriverUnprogrammed() { + MockDriver.reset(); + } + + @Test + public void emitsOncePerRisingEdgeOfTheTrigger() throws Exception { + MockDriver.programRead("TRIG", false, false, true, true, false, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:triggered"); + result.expectedMessageCount(2); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("triggered"); + + result.assertIsSatisfied(10000); + } + + @Test + public void deliversTheDataTagsWithoutTheTrigger() throws Exception { + MockDriver.programRead("TRIG", false, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:triggered"); + result.expectedMessageCount(1); + + context.getRouteController().startRoute("triggered"); + + result.assertIsSatisfied(10000); + assertEquals(Map.of("temperature", 21, "pressure", 1013), result.getExchanges().get(0).getIn().getBody(), + "the body holds the data tags, not the trigger"); + } + + @Test + public void emitsOnceForATriggerThatIsAlreadyTrueOnTheFirstRead() throws Exception { + MockDriver.programRead("TRIG", true, true, false); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:triggered"); + result.expectedMessageCount(1); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("triggered"); + + result.assertIsSatisfied(10000); + } + + @Test + public void stopsReadingWhenTheRouteStopsAndResumesWithoutDuplicatesAfterARestart() throws Exception { + MockDriver.programRead("TRIG", false, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:triggered"); + result.expectedMessageCount(3); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("triggered"); + await().atMost(10, TimeUnit.SECONDS).until(() -> result.getReceivedCounter() == 1); + context.getRouteController().stopRoute("triggered"); + + MockDriver.programRead("TRIG", false, true, false, true, false); + await().during(1, TimeUnit.SECONDS).atMost(2, TimeUnit.SECONDS).until(() -> result.getReceivedCounter() == 1); + + context.getRouteController().startRoute("triggered"); + + result.assertIsSatisfied(10000); + } + + @Test + public void readsTheTagsOnceWithoutATrigger() throws Exception { + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:untriggered"); + result.expectedMessageCount(1); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("untriggered"); + + result.assertIsSatisfied(10000); + assertEquals(Map.of("temperature", 21, "pressure", 1013), result.getExchanges().get(0).getIn().getBody()); + } + + @Test + public void aRefusedConnectionNeitherFailsTheRouteStartNorFloodsTheLog() throws Exception { + MockDriver.programRead("TRIG", false, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockDriver.refuseConnections(); + MockEndpoint result = getMockEndpoint("mock:refused"); + result.expectedMessageCount(0); + result.setAssertPeriod(1000); + WarningCapture warnings = new WarningCapture(); + + try { + warnings.attach(); + context.getRouteController().startRoute("refused"); + assertTrue(context.getRouteController().getRouteStatus("refused").isStarted()); + // the connection set up on start and at least one connection for reading the trigger were refused + await().atMost(10, TimeUnit.SECONDS).until(() -> MockDriver.refusedConnections() >= 2); + result.assertIsSatisfied(); + } finally { + warnings.detach(); + } + + // more than 40 periods of 50 ms have elapsed; a warning per period would be a flood + assertTrue(warnings.messages().size() <= 5, () -> "Too many warnings: " + warnings.messages()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from(TRIGGERED).routeId("triggered").autoStartup(false).to("mock:triggered"); + from(UNTRIGGERED).routeId("untriggered").autoStartup(false).to("mock:untriggered"); + from(REFUSED).routeId("refused").autoStartup(false).to("mock:refused"); + } + }; + } + + /** + * Captures WARN and ERROR events of the component and PLC4X loggers. + */ + private static final class WarningCapture extends AbstractAppender { + + private final List messages = new CopyOnWriteArrayList<>(); + + WarningCapture() { + super("plc4x-warnings", null, null, true, Property.EMPTY_ARRAY); + } + + @Override + public void append(LogEvent event) { + String logger = event.getLoggerName(); + if (event.getLevel().isMoreSpecificThan(Level.WARN) + && (logger.startsWith("org.apache.camel.component.plc4x") || logger.startsWith("org.apache.plc4x"))) { + messages.add(event.getLevel() + " " + logger + " " + event.getMessage().getFormattedMessage()); + } + } + + void attach() { + start(); + LoggerContext loggerContext = LoggerContext.getContext(false); + loggerContext.getConfiguration().getRootLogger().addAppender(this, Level.WARN, null); + loggerContext.updateLoggers(); + } + + void detach() { + LoggerContext loggerContext = LoggerContext.getContext(false); + loggerContext.getConfiguration().getRootLogger().removeAppender(getName()); + loggerContext.updateLoggers(); + stop(); + } + + List messages() { + return messages; + } + } +} diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerStopTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerStopTest.java deleted file mode 100644 index f2362156a1479..0000000000000 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerStopTest.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * 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.plc4x; - -import java.util.Map; - -import org.apache.camel.Processor; -import org.apache.camel.impl.DefaultCamelContext; -import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperImpl; -import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollectorImpl; -import org.junit.jupiter.api.Test; -import org.mockito.MockedConstruction; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.mockConstruction; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -/** - * A stopped triggered consumer must stop its scraper, or the scraper keeps reading the PLC and sending exchanges to the - * route, also after the route is started again with a new consumer (duplicates). - */ -class Plc4XConsumerStopTest { - - @Test - void testStopStopsTheScraper() throws Exception { - Plc4XEndpoint endpoint = mock(Plc4XEndpoint.class); - when(endpoint.getCamelContext()).thenReturn(new DefaultCamelContext()); - when(endpoint.getTrigger()).thenReturn("(PLC4X_TRIGGER_VAR,10,(%DB1:DBX0.0:BOOL)==(true))"); - when(endpoint.getTags()).thenReturn(Map.of("tag1", "%DB1:DBW2:INT")); - when(endpoint.getPeriod()).thenReturn(1000); - when(endpoint.getUri()).thenReturn("mock:plc"); - - try (MockedConstruction scrapers = mockConstruction(TriggeredScraperImpl.class); - MockedConstruction collectors = mockConstruction(TriggerCollectorImpl.class)) { - Plc4XConsumer consumer = new Plc4XConsumer(endpoint, mock(Processor.class)); - consumer.start(); - - assertEquals(1, scrapers.constructed().size()); - TriggeredScraperImpl scraper = scrapers.constructed().get(0); - TriggerCollectorImpl collector = collectors.constructed().get(0); - verify(scraper).start(); - verify(scraper, never()).stop(); - - consumer.stop(); - - verify(scraper).stop(); - verify(collector).stop(); - } - } -} diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerTest.java deleted file mode 100644 index fa595463a25ee..0000000000000 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerTest.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * 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.plc4x; - -import java.util.Collections; -import java.util.Map; - -import org.apache.camel.Exchange; -import org.apache.camel.Processor; -import org.apache.camel.impl.DefaultCamelContext; -import org.apache.plc4x.java.api.exceptions.PlcConnectionException; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import static org.junit.jupiter.api.Assertions.*; -import static org.mockito.Mockito.*; - -class Plc4XConsumerTest { - - private Plc4XEndpoint endpoint; - private Processor processor; - private Plc4XConsumer consumer; - - @BeforeEach - void setUp() { - endpoint = mock(Plc4XEndpoint.class); - processor = mock(Processor.class); - - when(endpoint.getTags()).thenReturn(Collections.emptyMap()); - when(endpoint.getTrigger()).thenReturn(null); // untriggered - when(endpoint.getCamelContext()).thenReturn(new DefaultCamelContext()); - - consumer = new Plc4XConsumer(endpoint, processor); - } - - @Test - void doStart() throws Exception { - - doNothing().when(endpoint).setupConnection(); - consumer.doStart(); - verify(endpoint, times(1)).setupConnection(); - } - - @Test - void doStartBadStart() throws Exception { - - doThrow(new PlcConnectionException("fail")) - .when(endpoint).setupConnection(); - consumer.doStart(); - verify(endpoint).setupConnection(); - assertFalse(consumer.isStarted()); - } - - @Test - void shouldStartTriggeredScraperWhenTriggerPresent() throws Exception { - - when(endpoint.getTrigger()).thenReturn("someTrigger"); - when(endpoint.getTags()).thenReturn(Map.of("tag1", "value1")); - when(endpoint.getPeriod()).thenReturn(1000); - when(endpoint.getUri()).thenReturn("mock:plc"); - - // Avoid real connection - doNothing().when(endpoint).setupConnection(); - doNothing().when(endpoint).reconnectIfNeeded(); - when(endpoint.createExchange()).thenReturn(mock(Exchange.class)); - consumer.doStart(); - verify(endpoint, atLeastOnce()).getTrigger(); - } - - @Test - void doStop() throws Exception { - consumer.doStop(); - assertFalse(consumer.isStarted(), "Consumer should not be started after doStop"); - } - -} From 54cffbd44d9a8e87e3e046d7b65728817660964d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sat, 24 Oct 2026 11:18:39 +0300 Subject: [PATCH 02/10] CAMEL-25617: camel-plc4x - Upgrade to PLC4X 1.0.0 and port the triggered consumer to ScheduledPollConsumer PLC4X 1.0.0 no longer publishes plc4j-scraper, which the triggered consumer was built on; no artifact replaces it. The ten driver artifacts keep their names, and the driver SPI (PlcDriver, DefaultPlcDriverManager) stays in plc4j-api. With a trigger set, the endpoint now creates Plc4XTriggeredConsumer, a ScheduledPollConsumer with a fixed delay of period milliseconds and the first poll at start. Each poll reads the trigger tag and the tags in one request on the endpoint's connection, after reconnectIfNeeded(), and emits the tags once when the trigger goes from false or unknown to true. The data therefore comes from the same read that saw the edge, where the scraper read the tags in a second request. A failed poll (connection, read, or a tag answered with an error code) goes to the consumer's exception handler, forgets the trigger value so a trigger that is still true emits again, and polling continues. A read that completes after the consumer stopped is not processed. A trigger without a positive period, or a tag named like the reserved trigger name TRIGGER_VAR, fails the route start; so does a connection that cannot be set up at start without autoReconnect, since no poll would reconnect it. No endpoint option is added or changed: the endpoint stays a DefaultEndpoint, so Camel's scheduler options do not leak into it. The untriggered consumer keeps its one-shot read and loses the scraper code; it and the polling consumer share the read response conversion. The scraper-only Constants.PLC_NAME is removed, and so is the MockDriver subscriber stub, which used SPI classes that moved in 1.0.0. MockDriver can now fail a whole read or answer a tag with an error code for one cycle, accept connections again after refusing them, and hold a read until the test releases it. New route tests cover the data of the edge cycle, the trigger reset after a failed read, an error code on a data tag, a read that completes while the route stops, a trigger without a period, the reserved trigger name and a refused connection at start without autoReconnect. The refused-connection test now asserts the ScheduledPollConsumer contract: the route starts, each failed poll reaches the exception handler, and delivery resumes once the PLC accepts connections, instead of capping the number of warnings. Co-Authored-By: Claude Fable 5.1 --- components/camel-plc4x/pom.xml | 5 - .../camel/component/plc4x/Constants.java | 1 - .../camel/component/plc4x/Plc4XConsumer.java | 71 +----- .../camel/component/plc4x/Plc4XEndpoint.java | 27 ++- .../component/plc4x/Plc4XPollingConsumer.java | 8 +- .../plc4x/Plc4XTriggeredConsumer.java | 139 ++++++++++++ .../camel/component/plc4x/MockDriver.java | 147 +++++++------ .../plc4x/Plc4XConsumerRouteTest.java | 205 ++++++++++++++---- parent/pom.xml | 2 +- 9 files changed, 408 insertions(+), 197 deletions(-) create mode 100644 components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java diff --git a/components/camel-plc4x/pom.xml b/components/camel-plc4x/pom.xml index 32f826e824f05..0bea7af6f9ebe 100644 --- a/components/camel-plc4x/pom.xml +++ b/components/camel-plc4x/pom.xml @@ -44,11 +44,6 @@ plc4j-api ${plc4x-version} - - org.apache.plc4x - plc4j-scraper - ${plc4x-version} - diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java index d973e4e37ddd7..6f995dde53fa0 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java @@ -21,7 +21,6 @@ public final class Constants { public static final String FIELD_NAME_HEADER = "fieldName"; public static final String FIELD_QUERY_HEADER = "fieldQuery"; public static final String TRIGGER = "TRIGGER_VAR"; - public static final String PLC_NAME = "PLC"; private Constants() { throw new IllegalStateException("Utility class!"); diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java index 581191a7929d8..e3865b5e84287 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java @@ -16,10 +16,6 @@ */ package org.apache.camel.component.plc4x; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -30,33 +26,20 @@ import org.apache.camel.support.DefaultConsumer; import org.apache.plc4x.java.api.exceptions.PlcConnectionException; import org.apache.plc4x.java.api.messages.PlcReadRequest; -import org.apache.plc4x.java.scraper.config.JobConfigurationImpl; -import org.apache.plc4x.java.scraper.config.ScraperConfiguration; -import org.apache.plc4x.java.scraper.config.triggeredscraper.ScraperConfigurationTriggeredImpl; -import org.apache.plc4x.java.scraper.exception.ScraperException; -import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperImpl; -import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollector; -import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollectorImpl; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class Plc4XConsumer extends DefaultConsumer { private static final Logger LOGGER = LoggerFactory.getLogger(Plc4XConsumer.class); - private final Map tags; - private final String trigger; private final Plc4XEndpoint plc4XEndpoint; private ScheduledExecutorService executorService; private ScheduledFuture future; - private TriggeredScraperImpl scraper; - private TriggerCollector collector; public Plc4XConsumer(Plc4XEndpoint endpoint, Processor processor) { super(endpoint, processor); this.plc4XEndpoint = endpoint; - this.tags = endpoint.getTags(); - this.trigger = endpoint.getTrigger(); } @Override @@ -82,11 +65,7 @@ protected void doStart() throws Exception { } doStop(); } - if (trigger == null) { - startUnTriggered(); - } else { - startTriggered(); - } + startUnTriggered(); } private void startUnTriggered() { @@ -111,11 +90,7 @@ private void startUnTriggered() { future = executorService.schedule(() -> request.execute().thenAccept(response -> { try { Exchange exchange = plc4XEndpoint.createExchange(); - Map rsp = new HashMap<>(); - for (String field : response.getTagNames()) { - rsp.put(field, response.getObject(field)); - } - exchange.getIn().setBody(rsp); + exchange.getIn().setBody(Plc4XEndpoint.tagValues(response)); getProcessor().process(exchange); } catch (Exception e) { getExceptionHandler().handleException(e); @@ -123,39 +98,6 @@ private void startUnTriggered() { }), 500, TimeUnit.MILLISECONDS); } - private void startTriggered() throws ScraperException { - ScraperConfiguration configuration = getScraperConfig(tags); - collector = new TriggerCollectorImpl(plc4XEndpoint.getPlcDriverManager()); - - scraper = new TriggeredScraperImpl(configuration, (job, alias, response) -> { - try { - plc4XEndpoint.reconnectIfNeeded(); - - Exchange exchange = plc4XEndpoint.createExchange(); - exchange.getIn().setBody(response); - getProcessor().process(exchange); - } catch (PlcConnectionException e) { - if (LOGGER.isTraceEnabled()) { - LOGGER.warn("Unable to reconnect, skipping request", e); - } else { - LOGGER.warn("Unable to reconnect, skipping request"); - } - } catch (Exception e) { - getExceptionHandler().handleException(e); - } - }, collector); - scraper.start(); - collector.start(); - } - - private ScraperConfigurationTriggeredImpl getScraperConfig(Map tagList) { - String config = "(TRIGGER_VAR," + plc4XEndpoint.getPeriod() + ",(" + plc4XEndpoint.getTrigger() + ")==(true))"; - List job = Collections.singletonList( - new JobConfigurationImpl("PLC4X-Camel", config, 0, Collections.singletonList(Constants.PLC_NAME), tagList)); - Map source = Collections.singletonMap(Constants.PLC_NAME, plc4XEndpoint.getUri()); - return new ScraperConfigurationTriggeredImpl(source, job); - } - @Override protected void doStop() throws Exception { // First stop the polling process @@ -167,15 +109,6 @@ protected void doStop() throws Exception { getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executorService); executorService = null; } - // a scraper that is not stopped keeps reading the PLC and sending exchanges to the route - if (scraper != null) { - scraper.stop(); - scraper = null; - } - if (collector != null) { - collector.stop(); - collector = null; - } super.doStop(); } diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java index 55fd8e9238816..112404b8cb5ec 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java @@ -16,6 +16,7 @@ */ package org.apache.camel.component.plc4x; +import java.util.HashMap; import java.util.Map; import org.apache.camel.Category; @@ -39,6 +40,7 @@ import org.apache.plc4x.java.api.exceptions.PlcConnectionException; import org.apache.plc4x.java.api.exceptions.PlcIncompatibleDatatypeException; import org.apache.plc4x.java.api.messages.PlcReadRequest; +import org.apache.plc4x.java.api.messages.PlcReadResponse; import org.apache.plc4x.java.api.messages.PlcWriteRequest; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -200,7 +202,8 @@ public Producer createProducer() { @Override public Consumer createConsumer(Processor processor) throws Exception { - Plc4XConsumer consumer = new Plc4XConsumer(this, processor); + Consumer consumer = trigger == null + ? new Plc4XConsumer(this, processor) : new Plc4XTriggeredConsumer(this, processor); configureConsumer(consumer); return consumer; } @@ -214,7 +217,18 @@ public PollingConsumer createPollingConsumer() { * Build a {@link PlcReadRequest} using the tags specified in the endpoint. */ public PlcReadRequest buildPlcReadRequest() { + return buildPlcReadRequest(Map.of()); + } + + /** + * Build a {@link PlcReadRequest} using the given tags and the tags specified in the endpoint. + * + * @param additionalTags tag addresses by name to read in addition to the tags of the endpoint + * @return {@link PlcReadRequest} + */ + PlcReadRequest buildPlcReadRequest(Map additionalTags) { PlcReadRequest.Builder builder = connection.readRequestBuilder(); + additionalTags.forEach(builder::addTagAddress); if (tags != null) { for (Map.Entry tag : tags.entrySet()) { try { @@ -228,6 +242,17 @@ public PlcReadRequest buildPlcReadRequest() { return builder.build(); } + /** + * @return the values of a read response by tag name + */ + static Map tagValues(PlcReadResponse response) { + Map values = new HashMap<>(); + for (String tag : response.getTagNames()) { + values.put(tag, response.getObject(tag)); + } + return values; + } + /** * Build a {@link PlcWriteRequest}. *

diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java index 6fcf7242244f1..dd063bfa8a1f7 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java @@ -16,8 +16,6 @@ */ package org.apache.camel.component.plc4x; -import java.util.HashMap; -import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -101,11 +99,7 @@ protected Exchange doReceive(long timeout) { response = future.get(); } - Map rsp = new HashMap<>(); - for (String field : response.getTagNames()) { - rsp.put(field, response.getObject(field)); - } - exchange.getIn().setBody(rsp); + exchange.getIn().setBody(Plc4XEndpoint.tagValues(response)); } catch (TimeoutException e) { // nothing received within the timeout LOGGER.debug("No response from the PLC within {} millis", timeout); diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java new file mode 100644 index 0000000000000..6852aecbb6672 --- /dev/null +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java @@ -0,0 +1,139 @@ +/* + * 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.plc4x; + +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.support.ScheduledPollConsumer; +import org.apache.plc4x.java.api.exceptions.PlcConnectionException; +import org.apache.plc4x.java.api.exceptions.PlcException; +import org.apache.plc4x.java.api.messages.PlcReadResponse; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +/** + * Reads the trigger tag together with the tags every {@code period} milliseconds and emits the tags once on each rising + * edge of the trigger. + */ +public class Plc4XTriggeredConsumer extends ScheduledPollConsumer { + private final Plc4XEndpoint plc4XEndpoint; + private final String trigger; + private final int period; + + // whether the trigger was true on the last poll; null when unknown, before the first poll or after a failed one + private Boolean lastTrigger; + + public Plc4XTriggeredConsumer(Plc4XEndpoint endpoint, Processor processor) { + super(endpoint, processor); + this.plc4XEndpoint = endpoint; + this.trigger = endpoint.getTrigger(); + this.period = endpoint.getPeriod(); + setInitialDelay(0); + setDelay(period); + setTimeUnit(TimeUnit.MILLISECONDS); + } + + @Override + public Plc4XEndpoint getEndpoint() { + return plc4XEndpoint; + } + + @Override + protected void doStart() throws Exception { + if (period <= 0) { + throw new IllegalArgumentException( + "The period must be a positive number of milliseconds when a trigger is set, but was " + period); + } + Map tags = plc4XEndpoint.getTags(); + if (tags != null && tags.containsKey(Constants.TRIGGER)) { + throw new IllegalArgumentException( + "The tag name " + Constants.TRIGGER + " is reserved for the trigger when a trigger is set"); + } + lastTrigger = null; + // without autoReconnect a connection that cannot be set up fails the start, as no poll would reconnect + plc4XEndpoint.setupConnection(); + super.doStart(); + } + + @Override + protected int poll() throws Exception { + PlcReadResponse response; + try { + response = read(); + } catch (Exception e) { + lastTrigger = null; + throw e; + } + if (!isRunAllowed()) { + // the consumer stopped while the read was in progress + return 0; + } + boolean triggered = Boolean.TRUE.equals(response.getObject(Constants.TRIGGER)); + boolean risingEdge = triggered && !Boolean.TRUE.equals(lastTrigger); + lastTrigger = triggered; + if (!risingEdge) { + return 0; + } + + Map tags = Plc4XEndpoint.tagValues(response); + tags.remove(Constants.TRIGGER); + Exchange exchange = createExchange(false); + exchange.getIn().setBody(tags); + try { + getProcessor().process(exchange); + } catch (Exception e) { + exchange.setException(e); + } finally { + if (exchange.getException() != null) { + getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException()); + } + releaseExchange(exchange, false); + } + return 1; + } + + /** + * Reads the trigger and the tags in one request; fails unless every tag was read. + */ + private PlcReadResponse read() throws Exception { + plc4XEndpoint.reconnectIfNeeded(); + if (plc4XEndpoint.connection == null) { + throw new PlcConnectionException("Cannot establish connection"); + } + PlcReadResponse response; + try { + response = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER, trigger)).execute().get(); + } catch (ExecutionException e) { + throw e.getCause() instanceof Exception cause ? cause : e; + } + for (String tag : response.getTagNames()) { + PlcResponseCode code = response.getResponseCode(tag); + if (code != PlcResponseCode.OK) { + throw new PlcException("Reading the tag " + tag + " failed with response code " + code); + } + } + return response; + } + + @Override + public String toString() { + return "Plc4XTriggeredConsumer[" + plc4XEndpoint + "]"; + } +} diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java index c83e162854e59..6c0eed1c917d7 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java @@ -17,53 +17,46 @@ package org.apache.camel.component.plc4x; import java.util.Arrays; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.apache.plc4x.java.api.PlcConnection; import org.apache.plc4x.java.api.PlcDriver; import org.apache.plc4x.java.api.authentication.PlcAuthentication; import org.apache.plc4x.java.api.exceptions.PlcConnectionException; +import org.apache.plc4x.java.api.exceptions.PlcException; +import org.apache.plc4x.java.api.exceptions.PlcRuntimeException; import org.apache.plc4x.java.api.messages.PlcReadRequest; import org.apache.plc4x.java.api.messages.PlcReadResponse; import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest; -import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest; import org.apache.plc4x.java.api.messages.PlcWriteRequest; -import org.apache.plc4x.java.api.model.PlcTag; import org.apache.plc4x.java.api.types.PlcResponseCode; -import org.apache.plc4x.java.api.types.PlcValueType; -import org.apache.plc4x.java.spi.messages.DefaultPlcSubscriptionResponse; -import org.apache.plc4x.java.spi.messages.PlcSubscriber; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import static org.mockito.Mockito.*; /** * Test driver for the {@code mock} protocol. A read returns the values a test programmed with * {@link #programRead(String, Object...)} when every tag in the request is programmed, and Mockito deep stubs - * otherwise. The programming is static because every driver manager loads its own driver instance; tests call - * {@link #reset()}. + * otherwise. A programmed {@link #failedRead()} fails the whole read, a programmed + * {@link #responseCode(PlcResponseCode)} answers that tag with an error code. The programming is static because every + * driver manager loads its own driver instance; tests call {@link #reset()}. */ public class MockDriver implements PlcDriver { - public static final Logger LOGGER = LoggerFactory.getLogger(MockDriver.class); - private static final Map PROGRAMMED_TAGS = new ConcurrentHashMap<>(); + private static volatile CompletableFuture heldReads = CompletableFuture.completedFuture(null); + private static volatile CompletableFuture heldRead = new CompletableFuture<>(); private static final AtomicInteger REFUSED_CONNECTIONS = new AtomicInteger(); + private static final Object FAILED_READ = new Object(); private static volatile boolean refuseConnections; - ExecutorService executorService = Executors.newFixedThreadPool(10); - /** * Programs the values that successive reads of a tag address return; the last value repeats once they are used up. */ @@ -74,6 +67,44 @@ public static void programRead(String tagAddress, Object... values) { PROGRAMMED_TAGS.put(tagAddress, new ProgrammedTag(Arrays.asList(values))); } + /** + * A programmed value that makes the whole read in which it is due fail. + */ + public static Object failedRead() { + return FAILED_READ; + } + + /** + * A programmed value that makes the read in which it is due answer the tag with the given response code and no + * value. + */ + public static Object responseCode(PlcResponseCode responseCode) { + return new FailedTag(responseCode); + } + + /** + * Holds the following reads until {@link #releaseReads()}; a held read ignores interrupts, as a read that a PLC has + * already answered does. + */ + public static void holdReads() { + heldRead = new CompletableFuture<>(); + heldReads = new CompletableFuture<>(); + } + + /** + * Waits until a read is held by {@link #holdReads()}. + */ + public static void awaitHeldRead() throws Exception { + heldRead.get(10, TimeUnit.SECONDS); + } + + /** + * Lets the reads held by {@link #holdReads()} complete. + */ + public static void releaseReads() { + heldReads.complete(null); + } + /** * Makes every following connection attempt fail. */ @@ -81,6 +112,13 @@ public static void refuseConnections() { refuseConnections = true; } + /** + * Makes the following connection attempts succeed again. + */ + public static void acceptConnections() { + refuseConnections = false; + } + /** * @return the number of connection attempts refused since the last {@link #reset()} */ @@ -92,6 +130,7 @@ public static int refusedConnections() { * Forgets all programmed tag values and accepts connections again. */ public static void reset() { + releaseReads(); PROGRAMMED_TAGS.clear(); refuseConnections = false; REFUSED_CONNECTIONS.set(0); @@ -123,36 +162,6 @@ public PlcConnection getConnection(String url) throws PlcConnectionException { .thenReturn(mock(PlcSubscriptionRequest.Builder.class, RETURNS_DEEP_STUBS)); when(plcConnectionMock.unsubscriptionRequestBuilder()) .thenReturn(mock(PlcUnsubscriptionRequest.Builder.class, RETURNS_DEEP_STUBS)); - - // Mock a typical subscriber. - PlcSubscriber plcSubscriber = mock(PlcSubscriber.class, RETURNS_DEEP_STUBS); - when(plcSubscriber.subscribe(any(PlcSubscriptionRequest.class))).thenAnswer(invocation -> { - LOGGER.info("Received {}", invocation); - // TODO: Translate this so it actually does something ... - /*PlcSubscriptionRequest subscriptionRequest = invocation.getArgument(0); - List responseItems = - subscriptionRequest.getFieldNames().stream().map( - fieldName -> subscriptionRequest.getField(fieldName)).map(field -> { - Consumer consumer = subscriptionRequestItem.getConsumer(); - executorService.submit(() -> { - while (!Thread.currentThread().isInterrupted()) { - consumer.accept(new SubscriptionEventItem<>(null, Calendar.getInstance(), Collections.singletonList("HelloWorld"))); - try { - TimeUnit.MILLISECONDS.sleep(100); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException(e); - } - } - }); - return new SubscriptionResponseItem<>(subscriptionRequestItem, - mock(PlcSubscriptionHandle.class, RETURNS_DEEP_STUBS), PlcResponseCode.OK); - }).collect(Collectors.toList()); - PlcSubscriptionResponse response = new PlcSubscriptionResponse(subscriptionRequest, responseItems);*/ - PlcSubscriptionResponse response - = new DefaultPlcSubscriptionResponse(mock(PlcSubscriptionRequest.class), new HashMap<>()); - return CompletableFuture.completedFuture(response); - }); return plcConnectionMock; } @@ -161,18 +170,6 @@ public PlcConnection getConnection(String url, PlcAuthentication authentication) return getConnection(null); } - /** - * Every tag is a boolean tag. Only the scraper prepares tags, to validate its trigger. - */ - @Override - public PlcTag prepareTag(String tagAddress) { - return mock(PlcTag.class, invocation -> switch (invocation.getMethod().getName()) { - case "getAddressString" -> tagAddress; - case "getPlcValueType" -> PlcValueType.BOOL; - default -> RETURNS_DEFAULTS.answer(invocation); - }); - } - private static PlcReadRequest.Builder readRequestBuilder() { PlcReadRequest.Builder unprogrammed = mock(PlcReadRequest.Builder.class, RETURNS_DEEP_STUBS); Map tags = new LinkedHashMap<>(); @@ -190,21 +187,39 @@ private static PlcReadRequest.Builder readRequestBuilder() { private static PlcReadRequest programmedRequest(Map tags) { return mock(PlcReadRequest.class, invocation -> switch (invocation.getMethod().getName()) { - case "execute" -> CompletableFuture.completedFuture(read(tags)); + case "execute" -> read(tags); case "getTagNames" -> new LinkedHashSet<>(tags.keySet()); default -> RETURNS_DEFAULTS.answer(invocation); }); } - private static PlcReadResponse read(Map tags) { + private static CompletableFuture read(Map tags) { + heldRead.complete(null); + // join() does not return early on an interrupt + heldReads.join(); Map values = new LinkedHashMap<>(); tags.forEach((name, address) -> values.put(name, PROGRAMMED_TAGS.get(address).next())); - return mock(PlcReadResponse.class, invocation -> switch (invocation.getMethod().getName()) { - case "getTagNames" -> List.copyOf(values.keySet()); - case "getObject" -> values.get(invocation. getArgument(0)); - case "getResponseCode" -> PlcResponseCode.OK; - default -> RETURNS_DEFAULTS.answer(invocation); - }); + if (values.containsValue(FAILED_READ)) { + return CompletableFuture.failedFuture(new PlcException("Read failed in the mock driver")); + } + return CompletableFuture.completedFuture(mock(PlcReadResponse.class, + invocation -> switch (invocation.getMethod().getName()) { + case "getTagNames" -> List.copyOf(values.keySet()); + case "getObject" -> value(values.get(invocation. getArgument(0))); + case "getResponseCode" -> values.get(invocation. getArgument(0)) instanceof FailedTag failed + ? failed.responseCode() : PlcResponseCode.OK; + default -> RETURNS_DEFAULTS.answer(invocation); + })); + } + + private static Object value(Object programmed) { + if (programmed instanceof FailedTag failed) { + throw new PlcRuntimeException("Tag read failed with response code " + failed.responseCode()); + } + return programmed; + } + + private record FailedTag(PlcResponseCode responseCode) { } private static final class ProgrammedTag { diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java index eedc51deb7f2f..efc7de93bafe3 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java @@ -21,20 +21,22 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import org.apache.camel.Exchange; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.spi.ExceptionHandler; +import org.apache.camel.spi.Registry; import org.apache.camel.test.junit6.CamelTestSupport; -import org.apache.logging.log4j.Level; -import org.apache.logging.log4j.core.LogEvent; -import org.apache.logging.log4j.core.LoggerContext; -import org.apache.logging.log4j.core.appender.AbstractAppender; -import org.apache.logging.log4j.core.config.Property; +import org.apache.plc4x.java.api.exceptions.PlcConnectionException; +import org.apache.plc4x.java.api.types.PlcResponseCode; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -46,7 +48,14 @@ public class Plc4XConsumerRouteTest extends CamelTestSupport { private static final String PLC = "plc4x:mock:10.10.10.1/1/1?autoReconnect=true"; private static final String TRIGGERED = PLC + "&trigger=TRIG&period=100&tag.temperature=TEMP&tag.pressure=PRES"; private static final String UNTRIGGERED = PLC + "&tag.temperature=TEMP&tag.pressure=PRES"; - private static final String REFUSED = PLC + "&trigger=TRIG&period=50&tag.temperature=TEMP&tag.pressure=PRES"; + private static final String REPORTED = PLC + "&trigger=TRIG&period=50&tag.temperature=TEMP&tag.pressure=PRES" + + "&exceptionHandler=#pollFailures"; + private static final String WITHOUT_AUTO_RECONNECT + = "plc4x:mock:10.10.10.1/1/1?trigger=TRIG&period=50&tag.temperature=TEMP"; + private static final String RESERVED_TAG_NAME = PLC + "&trigger=TRIG&period=50&tag.TRIGGER_VAR=TEMP"; + private static final String WITHOUT_PERIOD = PLC + "&trigger=TRIG&tag.temperature=TEMP"; + + private final PollFailures pollFailures = new PollFailures(); @BeforeEach void resetMockDriver() { @@ -137,29 +146,121 @@ public void readsTheTagsOnceWithoutATrigger() throws Exception { } @Test - public void aRefusedConnectionNeitherFailsTheRouteStartNorFloodsTheLog() throws Exception { + public void deliversTheDataTagsOfTheReadThatSawTheEdge() throws Exception { + MockDriver.programRead("TRIG", false, true, false, true); + MockDriver.programRead("TEMP", 21, 22, 23, 24); + MockDriver.programRead("PRES", 1011, 1012, 1013, 1014); + MockEndpoint result = getMockEndpoint("mock:triggered"); + result.expectedMessageCount(2); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("triggered"); + + result.assertIsSatisfied(10000); + assertEquals(List.of(Map.of("temperature", 22, "pressure", 1012), Map.of("temperature", 24, "pressure", 1014)), + result.getExchanges().stream().map(exchange -> exchange.getIn().getBody()).toList()); + } + + @Test + public void reportsEachFailedPollAndResumesWhenThePlcReturns() throws Exception { MockDriver.programRead("TRIG", false, true); MockDriver.programRead("TEMP", 21); MockDriver.programRead("PRES", 1013); MockDriver.refuseConnections(); - MockEndpoint result = getMockEndpoint("mock:refused"); - result.expectedMessageCount(0); + MockEndpoint result = getMockEndpoint("mock:reported"); + result.expectedMessageCount(1); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("reported"); + assertTrue(context.getRouteController().getRouteStatus("reported").isStarted()); + await().atMost(10, TimeUnit.SECONDS).until(() -> pollFailures.count() >= 3); + assertEquals(0, result.getReceivedCounter(), "nothing is delivered while the PLC refuses connections"); + + MockDriver.acceptConnections(); + + result.assertIsSatisfied(10000); + assertTrue(pollFailures.causes().stream().allMatch(PlcConnectionException.class::isInstance), + () -> "Unexpected causes: " + pollFailures.causes()); + } + + @Test + public void aFailedPollForgetsTheTriggerSoAStillTrueTriggerEmitsAgain() throws Exception { + MockDriver.programRead("TRIG", true, true, MockDriver.failedRead(), true, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:reported"); + result.expectedMessageCount(2); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("reported"); + + result.assertIsSatisfied(10000); + assertEquals(1, pollFailures.count()); + } + + @Test + public void aTagReadWithAnErrorCodeFailsThePoll() throws Exception { + MockDriver.programRead("TRIG", false, true, true); + MockDriver.programRead("TEMP", 21, MockDriver.responseCode(PlcResponseCode.NOT_FOUND), 23); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:reported"); + result.expectedMessageCount(1); result.setAssertPeriod(1000); - WarningCapture warnings = new WarningCapture(); - - try { - warnings.attach(); - context.getRouteController().startRoute("refused"); - assertTrue(context.getRouteController().getRouteStatus("refused").isStarted()); - // the connection set up on start and at least one connection for reading the trigger were refused - await().atMost(10, TimeUnit.SECONDS).until(() -> MockDriver.refusedConnections() >= 2); - result.assertIsSatisfied(); - } finally { - warnings.detach(); - } - // more than 40 periods of 50 ms have elapsed; a warning per period would be a flood - assertTrue(warnings.messages().size() <= 5, () -> "Too many warnings: " + warnings.messages()); + context.getRouteController().startRoute("reported"); + + result.assertIsSatisfied(10000); + assertEquals(Map.of("temperature", 23, "pressure", 1013), result.getExchanges().get(0).getIn().getBody()); + assertEquals(1, pollFailures.count()); + assertTrue(pollFailures.causes().get(0).getMessage().contains("NOT_FOUND"), + () -> "Unexpected cause: " + pollFailures.causes().get(0)); + } + + @Test + public void aReadThatCompletesWhileTheRouteStopsIsNotDelivered() throws Exception { + MockDriver.programRead("TRIG", true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockDriver.holdReads(); + MockEndpoint result = getMockEndpoint("mock:reported"); + result.expectedMessageCount(0); + + context.getRouteController().startRoute("reported"); + MockDriver.awaitHeldRead(); + context.getRouteController().stopRoute("reported"); + MockDriver.releaseReads(); + + // with no message expected, the mock endpoint waits the whole time before it is satisfied + result.assertIsSatisfied(1000); + assertEquals(0, pollFailures.count(), () -> "Unexpected failures: " + pollFailures.causes()); + } + + @Test + public void aRefusedConnectionFailsTheRouteStartWithoutAutoReconnect() { + MockDriver.refuseConnections(); + + Exception e = assertThrows(Exception.class, + () -> context.getRouteController().startRoute("withoutAutoReconnect")); + + assertNotNull(causeOfType(e, PlcConnectionException.class), () -> "No PlcConnectionException in " + e); + } + + @Test + public void aTagNamedLikeTheTriggerFailsTheRouteStart() { + Exception e = assertThrows(Exception.class, () -> context.getRouteController().startRoute("reservedTagName")); + + Throwable cause = causeOfType(e, IllegalArgumentException.class); + assertNotNull(cause, () -> "No IllegalArgumentException in " + e); + assertTrue(cause.getMessage().contains(Constants.TRIGGER), cause::getMessage); + } + + @Test + public void aTriggerWithoutAPeriodFailsTheRouteStart() { + Exception e = assertThrows(Exception.class, () -> context.getRouteController().startRoute("withoutPeriod")); + + Throwable cause = causeOfType(e, IllegalArgumentException.class); + assertNotNull(cause, () -> "No IllegalArgumentException in " + e); + assertTrue(cause.getMessage().contains("period"), cause::getMessage); } @Override @@ -169,47 +270,57 @@ protected RouteBuilder createRouteBuilder() { public void configure() { from(TRIGGERED).routeId("triggered").autoStartup(false).to("mock:triggered"); from(UNTRIGGERED).routeId("untriggered").autoStartup(false).to("mock:untriggered"); - from(REFUSED).routeId("refused").autoStartup(false).to("mock:refused"); + from(REPORTED).routeId("reported").autoStartup(false).to("mock:reported"); + from(WITHOUT_AUTO_RECONNECT).routeId("withoutAutoReconnect").autoStartup(false) + .to("mock:withoutAutoReconnect"); + from(RESERVED_TAG_NAME).routeId("reservedTagName").autoStartup(false).to("mock:reservedTagName"); + from(WITHOUT_PERIOD).routeId("withoutPeriod").autoStartup(false).to("mock:withoutPeriod"); } }; } + @Override + protected void bindToRegistry(Registry registry) { + registry.bind("pollFailures", pollFailures); + } + + private static Throwable causeOfType(Throwable exception, Class type) { + Throwable cause = exception; + while (cause != null && !type.isInstance(cause)) { + cause = cause.getCause(); + } + return cause; + } + /** - * Captures WARN and ERROR events of the component and PLC4X loggers. + * Collects the failures the consumer reports to its exception handler. */ - private static final class WarningCapture extends AbstractAppender { + private static final class PollFailures implements ExceptionHandler { - private final List messages = new CopyOnWriteArrayList<>(); + private final List causes = new CopyOnWriteArrayList<>(); - WarningCapture() { - super("plc4x-warnings", null, null, true, Property.EMPTY_ARRAY); + @Override + public void handleException(Throwable exception) { + causes.add(exception); } @Override - public void append(LogEvent event) { - String logger = event.getLoggerName(); - if (event.getLevel().isMoreSpecificThan(Level.WARN) - && (logger.startsWith("org.apache.camel.component.plc4x") || logger.startsWith("org.apache.plc4x"))) { - messages.add(event.getLevel() + " " + logger + " " + event.getMessage().getFormattedMessage()); - } + public void handleException(String message, Throwable exception) { + causes.add(exception); } - void attach() { - start(); - LoggerContext loggerContext = LoggerContext.getContext(false); - loggerContext.getConfiguration().getRootLogger().addAppender(this, Level.WARN, null); - loggerContext.updateLoggers(); + @Override + public void handleException(String message, Exchange exchange, Throwable exception) { + causes.add(exception); } - void detach() { - LoggerContext loggerContext = LoggerContext.getContext(false); - loggerContext.getConfiguration().getRootLogger().removeAppender(getName()); - loggerContext.updateLoggers(); - stop(); + int count() { + return causes.size(); } - List messages() { - return messages; + List causes() { + return causes; } } + } diff --git a/parent/pom.xml b/parent/pom.xml index 853bcdb2b7665..98ba4df630507 100644 --- a/parent/pom.xml +++ b/parent/pom.xml @@ -448,7 +448,7 @@ 6.2.2 4.7.7 3.1.0 - 0.13.1 + 1.0.0 4.14.0 1.2.0 2.1.1 From 4ace4cfa37ced507216c5ca9cf5f84a958ebfc93 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sat, 24 Oct 2026 00:54:16 +0300 Subject: [PATCH 03/10] CAMEL-25617: camel-plc4x - Hand the driver only its own connection string parameters The endpoint passed its whole URI, Camel options included, to the PLC4X driver manager as the connection string. PLC4X 1.0.0 drivers report every parameter they do not know as a warning, so each connection logged one warning per Camel option (trigger, period, tag.*, autoReconnect, exceptionHandler, ...). On start the endpoint now removes its own options from the connection string, taking the option names and multi-value prefixes from the component's generated EndpointUriFactory, and keeps every other parameter as it is in the endpoint URI, percent-encoding included. The endpoint keeps reporting its configured URI; the connection string for the driver is kept separately. Co-Authored-By: Claude Fable 5.1 --- .../camel/component/plc4x/Plc4XEndpoint.java | 31 +++++++++++++++++-- .../camel/component/plc4x/MockDriver.java | 13 +++++++- .../plc4x/Plc4XConsumerRouteTest.java | 22 +++++++++++++ 3 files changed, 62 insertions(+), 4 deletions(-) diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java index 112404b8cb5ec..c3d6ce829b742 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java @@ -16,7 +16,9 @@ */ package org.apache.camel.component.plc4x; +import java.net.URISyntaxException; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; import org.apache.camel.Category; @@ -29,12 +31,14 @@ import org.apache.camel.spi.BodyType.Carrier; import org.apache.camel.spi.BodyTypes; import org.apache.camel.spi.EndpointServiceLocation; +import org.apache.camel.spi.EndpointUriFactory; import org.apache.camel.spi.Metadata; import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.spi.UriPath; import org.apache.camel.support.DefaultEndpoint; import org.apache.camel.util.StringHelper; +import org.apache.camel.util.URISupport; import org.apache.plc4x.java.DefaultPlcDriverManager; import org.apache.plc4x.java.api.PlcConnection; import org.apache.plc4x.java.api.exceptions.PlcConnectionException; @@ -84,6 +88,8 @@ public class Plc4XEndpoint extends DefaultEndpoint implements EndpointServiceLoc private boolean autoReconnect; private String uri; + // the connection string handed to the driver: the uri without the options of this endpoint + private String connectionString; public Plc4XEndpoint(String endpointUri, Component component) { super(endpointUri, component); @@ -135,7 +141,7 @@ public boolean isAutoReconnect() { */ public void setupConnection() throws PlcConnectionException { try { - connection = plcDriverManager.getConnection(this.uri); + connection = plcDriverManager.getConnection(connectionString); if (!isConnected()) { reconnectIfNeeded(); } @@ -163,7 +169,7 @@ public void reconnectIfNeeded() throws PlcConnectionException { if (isConnected()) { LOGGER.trace("No reconnect needed, already connected"); } else if (autoReconnect && connection == null) { - connection = plcDriverManager.getConnection(uri); + connection = plcDriverManager.getConnection(connectionString); LOGGER.debug("Successfully reconnected"); } else if (autoReconnect && !isConnected()) { if (connection != null) { @@ -172,7 +178,7 @@ public void reconnectIfNeeded() throws PlcConnectionException { // If reconnection fails without Exception, reset connection if (!isConnected()) { LOGGER.debug("No connection established after connect, resetting connection"); - connection = plcDriverManager.getConnection(uri); + connection = plcDriverManager.getConnection(connectionString); } LOGGER.debug("Successfully reconnected"); } else { @@ -296,10 +302,29 @@ public void setTags(Map tags) { @Override protected void doStart() throws Exception { super.doStart(); + connectionString = withoutEndpointOptions(uri); ClassLoader cl = getCamelContext().getApplicationContextClassLoader(); this.plcDriverManager = new DefaultPlcDriverManager(cl); } + /** + * Removes the options of this endpoint from a connection string, so the PLC4X driver only gets the parameters it + * knows. + */ + private String withoutEndpointOptions(String endpointUri) throws URISyntaxException { + EndpointUriFactory factory = getCamelContext().getCamelContextExtension().getEndpointUriFactory("plc4x"); + if (factory == null) { + throw new IllegalStateException("Cannot find the EndpointUriFactory of the plc4x component"); + } + Map parameters + = new LinkedHashMap<>(URISupport.parseQuery(URISupport.extractQuery(endpointUri), true)); + parameters.keySet().removeIf(name -> factory.propertyNames().contains(name) + || factory.multiValuePrefixes().values().stream().anyMatch(name::startsWith)); + String query = URISupport.createQueryString(parameters, false); + String path = URISupport.stripQuery(endpointUri); + return query.isEmpty() ? path : path + "?" + query; + } + @Override public void doStop() throws Exception { super.doStop(); diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java index 6c0eed1c917d7..44246f8c2fa75 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java @@ -23,6 +23,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -54,6 +55,7 @@ public class MockDriver implements PlcDriver { private static volatile CompletableFuture heldReads = CompletableFuture.completedFuture(null); private static volatile CompletableFuture heldRead = new CompletableFuture<>(); private static final AtomicInteger REFUSED_CONNECTIONS = new AtomicInteger(); + private static final List CONNECTION_STRINGS = new CopyOnWriteArrayList<>(); private static final Object FAILED_READ = new Object(); private static volatile boolean refuseConnections; @@ -126,6 +128,13 @@ public static int refusedConnections() { return REFUSED_CONNECTIONS.get(); } + /** + * @return the connection strings the driver was asked to connect with since the last {@link #reset()} + */ + public static List connectionStrings() { + return CONNECTION_STRINGS; + } + /** * Forgets all programmed tag values and accepts connections again. */ @@ -134,6 +143,7 @@ public static void reset() { PROGRAMMED_TAGS.clear(); refuseConnections = false; REFUSED_CONNECTIONS.set(0); + CONNECTION_STRINGS.clear(); } @Override @@ -148,6 +158,7 @@ public String getProtocolName() { @Override public PlcConnection getConnection(String url) throws PlcConnectionException { + CONNECTION_STRINGS.add(url); if (refuseConnections) { REFUSED_CONNECTIONS.incrementAndGet(); throw new PlcConnectionException("Connection refused by the mock driver"); @@ -167,7 +178,7 @@ public PlcConnection getConnection(String url) throws PlcConnectionException { @Override public PlcConnection getConnection(String url, PlcAuthentication authentication) throws PlcConnectionException { - return getConnection(null); + return getConnection(url); } private static PlcReadRequest.Builder readRequestBuilder() { diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java index efc7de93bafe3..14b61b717292a 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java @@ -18,6 +18,7 @@ import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; @@ -54,6 +55,9 @@ public class Plc4XConsumerRouteTest extends CamelTestSupport { = "plc4x:mock:10.10.10.1/1/1?trigger=TRIG&period=50&tag.temperature=TEMP"; private static final String RESERVED_TAG_NAME = PLC + "&trigger=TRIG&period=50&tag.TRIGGER_VAR=TEMP"; private static final String WITHOUT_PERIOD = PLC + "&trigger=TRIG&tag.temperature=TEMP"; + private static final String DRIVER_PARAMETERS = PLC + "&trigger=TRIG&period=100&tag.temperature=TEMP" + + "&exceptionHandler=#pollFailures&exchangePattern=InOnly" + + "&remote-rack=1&request-timeout-ms=500&percent-encoded=%25DB1"; private final PollFailures pollFailures = new PollFailures(); @@ -263,6 +267,23 @@ public void aTriggerWithoutAPeriodFailsTheRouteStart() { assertTrue(cause.getMessage().contains("period"), cause::getMessage); } + @Test + public void handsTheDriverOnlyItsOwnConnectionParameters() throws Exception { + MockDriver.programRead("TRIG", true); + MockDriver.programRead("TEMP", 21); + MockEndpoint result = getMockEndpoint("mock:driverParameters"); + result.expectedMessageCount(1); + + context.getRouteController().startRoute("driverParameters"); + + result.assertIsSatisfied(10000); + assertEquals(Set.of("mock:10.10.10.1/1/1?percent-encoded=%25DB1&remote-rack=1&request-timeout-ms=500"), + Set.copyOf(MockDriver.connectionStrings())); + String uri = context.getEndpoint(DRIVER_PARAMETERS, Plc4XEndpoint.class).getUri(); + assertTrue(uri.contains("trigger=TRIG") && uri.contains("autoReconnect=true"), + () -> "The endpoint no longer reports its configured uri: " + uri); + } + @Override protected RouteBuilder createRouteBuilder() { return new RouteBuilder() { @@ -275,6 +296,7 @@ public void configure() { .to("mock:withoutAutoReconnect"); from(RESERVED_TAG_NAME).routeId("reservedTagName").autoStartup(false).to("mock:reservedTagName"); from(WITHOUT_PERIOD).routeId("withoutPeriod").autoStartup(false).to("mock:withoutPeriod"); + from(DRIVER_PARAMETERS).routeId("driverParameters").autoStartup(false).to("mock:driverParameters"); } }; } From c4ab4957e04105d7e8484635b9967d1c4cec555b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sat, 24 Oct 2026 00:38:47 +0300 Subject: [PATCH 04/10] CAMEL-25617: camel-plc4x - Document the PLC4X 1.0.0 upgrade and the polled trigger Describe triggered reading in the component documentation: the trigger tag is polled every period on the endpoint's connection, the tags are read in the same request and delivered once per rising edge, and a failed poll goes to the consumer's exception handler while polling continues. It states that recovery after a lost connection needs autoReconnect=true, how the start behaves when the PLC cannot be reached, and the reserved trigger name. Add a camel-plc4x section to the 4.24 upgrade guide for PLC4X 1.0.0, its connection-string changes and the changed triggered consumer, and update the catalog copy of the component documentation. Co-Authored-By: Claude Fable 5.1 --- .../camel/catalog/docs/plc4x-component.adoc | 33 +++++++++++++++++++ .../src/main/docs/plc4x-component.adoc | 33 +++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 33 +++++++++++++++++++ 3 files changed, 99 insertions(+) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc index 294a7b6998822..af51b26ec8841 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc @@ -91,6 +91,39 @@ no PLC. With a real PLC, the driver in the URI and the addresses of the tags fol message: "Tags read: ${body}" ---- +With the `trigger` option, the consumer reads the tags each time a boolean trigger tag rises. It polls the trigger tag +every `period` milliseconds on the connection of the endpoint, starting when the route starts; `period` is required +with `trigger` and must be positive. On a rising edge of the trigger, from `false` (or from no value read yet) to +`true`, the consumer delivers one exchange whose body is a `Map` of the tags, read in the same request +as the trigger. The trigger tag is not part of the body; it is read under the reserved name `TRIGGER_VAR`, which no tag +may use. While the trigger stays `true`, no further exchange is delivered. + +A poll that fails, for example because a tag cannot be read or the PLC cannot be reached, is reported to the exception +handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling continues. +The trigger value read before the failure is forgotten, so a trigger that is `true` at the first successful poll after +a failure counts as a rising edge. Automatic recovery after a lost connection needs `autoReconnect=true`: a poll then +replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on the first lost +connection and does not recover. Likewise, a connection that cannot be set up when the route starts is retried by the +polls with `autoReconnect=true`, and fails the start of the route without it. + +The following route reads a register of a Modbus TCP device each time coil 1 goes from `false` to `true`, checking the +coil every 100 milliseconds: + +[source,yaml] +---- +- route: + from: + uri: plc4x:modbus-tcp://192.168.0.10:502 + parameters: + trigger: "coil:1" + period: 100 + autoReconnect: true + tag.counter: "holding-register:2:INT" + steps: + - log: + message: "Counter read on trigger: ${body}" +---- + === Polling Consumer The polling consumer supports consecutive reading. The input and output are the same as for the regular consumer. diff --git a/components/camel-plc4x/src/main/docs/plc4x-component.adoc b/components/camel-plc4x/src/main/docs/plc4x-component.adoc index 294a7b6998822..af51b26ec8841 100644 --- a/components/camel-plc4x/src/main/docs/plc4x-component.adoc +++ b/components/camel-plc4x/src/main/docs/plc4x-component.adoc @@ -91,6 +91,39 @@ no PLC. With a real PLC, the driver in the URI and the addresses of the tags fol message: "Tags read: ${body}" ---- +With the `trigger` option, the consumer reads the tags each time a boolean trigger tag rises. It polls the trigger tag +every `period` milliseconds on the connection of the endpoint, starting when the route starts; `period` is required +with `trigger` and must be positive. On a rising edge of the trigger, from `false` (or from no value read yet) to +`true`, the consumer delivers one exchange whose body is a `Map` of the tags, read in the same request +as the trigger. The trigger tag is not part of the body; it is read under the reserved name `TRIGGER_VAR`, which no tag +may use. While the trigger stays `true`, no further exchange is delivered. + +A poll that fails, for example because a tag cannot be read or the PLC cannot be reached, is reported to the exception +handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling continues. +The trigger value read before the failure is forgotten, so a trigger that is `true` at the first successful poll after +a failure counts as a rising edge. Automatic recovery after a lost connection needs `autoReconnect=true`: a poll then +replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on the first lost +connection and does not recover. Likewise, a connection that cannot be set up when the route starts is retried by the +polls with `autoReconnect=true`, and fails the start of the route without it. + +The following route reads a register of a Modbus TCP device each time coil 1 goes from `false` to `true`, checking the +coil every 100 milliseconds: + +[source,yaml] +---- +- route: + from: + uri: plc4x:modbus-tcp://192.168.0.10:502 + parameters: + trigger: "coil:1" + period: 100 + autoReconnect: true + tag.counter: "holding-register:2:INT" + steps: + - log: + message: "Counter read on trigger: ${body}" +---- + === Polling Consumer The polling consumer supports consecutive reading. The input and output are the same as for the regular consumer. diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 9af133837d120..d47621a1c80d0 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -3582,3 +3582,36 @@ Use `0.0.0.0` as the host to accept connections on all network interfaces. `AvroRpcHttpServerFactory` has a new method `create(SpecificResponder, String, int)` that receives the host. A custom factory should implement it. +=== camel-plc4x - PLC4X 1.0.0 and the triggered consumer + +`camel-plc4x` now uses PLC4X 1.0.0, which requires Java 21, as Camel 4.24 does. + +PLC4X 1.0.0 changes the connection strings of its drivers. Several parameters are renamed, for example +`request-timeout` to `request-timeout-ms`, and the S7 driver reads the rack and slot as `cotp.remote-rack` and +`cotp.remote-slot` instead of `remote-rack` and `remote-slot`. A parameter that a driver does not know, such as an old +name, is logged as a warning and does not apply. The OPC UA driver now uses the security policy `Basic256Sha256` by +default instead of `NONE`, `insecure-certificate-verification=true` becomes `tls.verify=false`, and a user name and +password are refused over a channel that is neither signed nor encrypted unless `allow-insecure-credentials` is set. +See the incompatible changes in the https://github.com/apache/plc4x/releases/tag/v1.0.0[PLC4X 1.0.0 release notes] +for the full list. + +A consumer with `trigger` now reads the trigger tag and the tags in one request on the connection of the endpoint, +every `period` milliseconds from the start of the route. Before, it sampled the trigger first after about one second, +and read the tags in a later, separate request on another connection. A failed read is now reported to the exception +handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling continues; +before, the consumer stopped triggering after the first failed read of the trigger. After a failed poll the remembered +trigger value is unknown, so a trigger that is still `true` when the PLC returns emits once. Since the consumer now uses +the connection of the endpoint, `autoReconnect` governs its reconnects. A connection that cannot be set up when the +route starts fails the start of the route, unless `autoReconnect=true`, in which case the polls retry it. As before, +`period` must be positive when `trigger` is set. See the xref:components::plc4x-component.adoc[PLC4X component] for how +triggered reading works. + +With `autoReconnect=true`, a lost connection is now closed and replaced with a new one from the driver instead of being +reconnected, because the PLC4X 1.0.0 S7 and EtherNet/IP drivers cannot connect a connection again once it is lost. This +applies to the producer and to the consumers alike. + +The component now removes its own options (`trigger`, `period`, `tag.*`, `autoReconnect` and the generic consumer +options such as `exceptionHandler`) from the connection string it hands to the driver. Before, the driver received them +as connection string parameters too. + +The constant `Constants.PLC_NAME`, which was only used with the PLC4X scraper, was removed. From b1fac44c463f4b6341f4484a6f2d86df516406ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sat, 24 Oct 2026 01:15:02 +0300 Subject: [PATCH 05/10] CAMEL-25617: camel-plc4x - Replace a lost connection instead of reconnecting it With autoReconnect the endpoint called connect() on a connection that reported itself not connected and only took a new connection when that call returned without one. The PLC4X 1.0.0 S7 and EtherNet/IP connections cannot connect again once their transport closed: connect() throws ("Transport is not open"), so after a PLC restart every poll of the triggered consumer, and every write of the producer, failed until the route was restarted. Run against a snap7 S7 server and libplctag's ab_server; the Modbus TCP connection happened to reconnect. The endpoint now closes the lost connection and asks the driver manager for a new one, which every driver supports. MockDriver connections now report that they are connected from connect() until the test drops them, so the route tests exercise the endpoint's one connection; a new test checks that the triggered consumer polls on a single connection. Co-Authored-By: Claude Fable 5.1 --- .../camel/component/plc4x/Plc4XEndpoint.java | 26 +++++++++------- .../camel/component/plc4x/MockDriver.java | 30 ++++++++++++++++-- .../plc4x/Plc4XConsumerRouteTest.java | 31 +++++++++++++++++++ 3 files changed, 74 insertions(+), 13 deletions(-) diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java index c3d6ce829b742..6457a6623f13a 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java @@ -168,25 +168,29 @@ public void setupConnection() throws PlcConnectionException { public void reconnectIfNeeded() throws PlcConnectionException { if (isConnected()) { LOGGER.trace("No reconnect needed, already connected"); - } else if (autoReconnect && connection == null) { + } else if (autoReconnect) { + // a lost PLC4X connection cannot always connect again (the S7 and EtherNet/IP drivers fail on a closed + // transport), so it is replaced by a new one + closeLostConnection(); connection = plcDriverManager.getConnection(connectionString); LOGGER.debug("Successfully reconnected"); - } else if (autoReconnect && !isConnected()) { - if (connection != null) { - connection.connect(); - } - // If reconnection fails without Exception, reset connection - if (!isConnected()) { - LOGGER.debug("No connection established after connect, resetting connection"); - connection = plcDriverManager.getConnection(connectionString); - } - LOGGER.debug("Successfully reconnected"); } else { LOGGER.warn("Connection lost and auto-reconnect is turned off, shutting down Plc4XEndpoint"); stop(); } } + private void closeLostConnection() { + if (connection != null) { + try { + connection.close(); + } catch (Exception e) { + LOGGER.debug("Closing the lost connection failed: {}", e.getMessage()); + } + connection = null; + } + } + /** * @return true if connection supports writing, else false */ diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java index 44246f8c2fa75..a27f979dc6f32 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java @@ -25,6 +25,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.apache.plc4x.java.api.PlcConnection; @@ -46,8 +47,10 @@ * Test driver for the {@code mock} protocol. A read returns the values a test programmed with * {@link #programRead(String, Object...)} when every tag in the request is programmed, and Mockito deep stubs * otherwise. A programmed {@link #failedRead()} fails the whole read, a programmed - * {@link #responseCode(PlcResponseCode)} answers that tag with an error code. The programming is static because every - * driver manager loads its own driver instance; tests call {@link #reset()}. + * {@link #responseCode(PlcResponseCode)} answers that tag with an error code. A connection reports that it is connected + * from {@code connect()} until {@link #dropConnections()} drops it; a dropped connection cannot connect again, as after + * a PLC restart. The programming is static because every driver manager loads its own driver instance; tests call + * {@link #reset()}. */ public class MockDriver implements PlcDriver { @@ -58,6 +61,8 @@ public class MockDriver implements PlcDriver { private static final List CONNECTION_STRINGS = new CopyOnWriteArrayList<>(); private static final Object FAILED_READ = new Object(); private static volatile boolean refuseConnections; + private static final AtomicInteger CREATED_CONNECTIONS = new AtomicInteger(); + private static volatile int droppedConnections; /** * Programs the values that successive reads of a tag address return; the last value repeats once they are used up. @@ -121,6 +126,14 @@ public static void acceptConnections() { refuseConnections = false; } + /** + * Drops the connections made so far, as a restarted PLC does: they report that they are not connected from now on + * and fail to connect again, while a new connection succeeds. + */ + public static void dropConnections() { + droppedConnections = CREATED_CONNECTIONS.get(); + } + /** * @return the number of connection attempts refused since the last {@link #reset()} */ @@ -143,6 +156,8 @@ public static void reset() { PROGRAMMED_TAGS.clear(); refuseConnections = false; REFUSED_CONNECTIONS.set(0); + CREATED_CONNECTIONS.set(0); + droppedConnections = 0; CONNECTION_STRINGS.clear(); } @@ -165,6 +180,17 @@ public PlcConnection getConnection(String url) throws PlcConnectionException { } // Mock a connection. PlcConnection plcConnectionMock = mock(PlcConnection.class, RETURNS_DEEP_STUBS); + int connectionNumber = CREATED_CONNECTIONS.getAndIncrement(); + AtomicBoolean connected = new AtomicBoolean(); + doAnswer(invocation -> { + if (connectionNumber < droppedConnections) { + throw new PlcConnectionException("Transport is not open"); + } + connected.set(true); + return null; + }).when(plcConnectionMock).connect(); + when(plcConnectionMock.isConnected()) + .thenAnswer(invocation -> connected.get() && connectionNumber >= droppedConnections); when(plcConnectionMock.getMetadata().isReadSupported()).thenReturn(true); when(plcConnectionMock.getMetadata().isWriteSupported()).thenReturn(true); when(plcConnectionMock.readRequestBuilder()).thenAnswer(invocation -> readRequestBuilder()); diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java index 14b61b717292a..cde895e1e79d0 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java @@ -187,6 +187,37 @@ public void reportsEachFailedPollAndResumesWhenThePlcReturns() throws Exception () -> "Unexpected causes: " + pollFailures.causes()); } + @Test + public void pollsOnTheOneConnectionOfTheEndpoint() throws Exception { + MockDriver.programRead("TRIG", false, false, false, false, false, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:reported"); + result.expectedMessageCount(1); + + context.getRouteController().startRoute("reported"); + + result.assertIsSatisfied(10000); + assertEquals(1, MockDriver.connectionStrings().size(), "six polls use the connection set up on start"); + } + + @Test + public void resumesOnANewConnectionWhenTheDroppedOneCannotConnectAgain() throws Exception { + MockDriver.programRead("TRIG", false, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:reported"); + result.expectedMessageCount(2); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("reported"); + await().atMost(10, TimeUnit.SECONDS).until(() -> result.getReceivedCounter() == 1); + MockDriver.dropConnections(); + MockDriver.programRead("TRIG", false, true); + + result.assertIsSatisfied(10000); + } + @Test public void aFailedPollForgetsTheTriggerSoAStillTrueTriggerEmitsAgain() throws Exception { MockDriver.programRead("TRIG", true, true, MockDriver.failedRead(), true, true); From 7629274c595d79c36ab2d44bcb961376b72bdd43 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sat, 24 Oct 2026 12:04:47 +0300 Subject: [PATCH 06/10] CAMEL-25617: camel-plc4x - Replace the shared connection under a lock The producer calls reconnectIfNeeded() for every exchange and the consumers for every read, so two callers could each close and replace the connection of the endpoint, leaking one connection or closing it under the other's request. The connection is now replaced, set up and closed while holding the endpoint's lock, and the field is volatile for the readers. stop() is called outside the lock, as a concurrent stop() holds the service lock while doStop() waits for the endpoint's lock. Co-authored-by: Claude Opus 5.5 --- .../camel/component/plc4x/Plc4XEndpoint.java | 48 ++++++++++++------- 1 file changed, 30 insertions(+), 18 deletions(-) diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java index 6457a6623f13a..2e9f174be743f 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java @@ -68,7 +68,7 @@ public class Plc4XEndpoint extends DefaultEndpoint implements EndpointServiceLoc private static final Logger LOGGER = LoggerFactory.getLogger(Plc4XEndpoint.class); protected DefaultPlcDriverManager plcDriverManager; - protected PlcConnection connection; + protected volatile PlcConnection connection; @UriPath @Metadata(required = true, description = "PLC4X connection string for the connection to the target") @@ -141,7 +141,9 @@ public boolean isAutoReconnect() { */ public void setupConnection() throws PlcConnectionException { try { - connection = plcDriverManager.getConnection(connectionString); + synchronized (this) { + connection = plcDriverManager.getConnection(connectionString); + } if (!isConnected()) { reconnectIfNeeded(); } @@ -166,18 +168,24 @@ public void setupConnection() throws PlcConnectionException { * @throws PlcConnectionException If reconnect failed and auto-reconnect is turned on */ public void reconnectIfNeeded() throws PlcConnectionException { - if (isConnected()) { - LOGGER.trace("No reconnect needed, already connected"); - } else if (autoReconnect) { - // a lost PLC4X connection cannot always connect again (the S7 and EtherNet/IP drivers fail on a closed - // transport), so it is replaced by a new one - closeLostConnection(); - connection = plcDriverManager.getConnection(connectionString); - LOGGER.debug("Successfully reconnected"); - } else { - LOGGER.warn("Connection lost and auto-reconnect is turned off, shutting down Plc4XEndpoint"); - stop(); + // the producer and the consumers of the endpoint share the connection: one of them replaces it at a time + synchronized (this) { + if (isConnected()) { + LOGGER.trace("No reconnect needed, already connected"); + return; + } + if (autoReconnect) { + // a lost PLC4X connection cannot always connect again (the S7 and EtherNet/IP drivers fail on a closed + // transport), so it is replaced by a new one + closeLostConnection(); + connection = plcDriverManager.getConnection(connectionString); + LOGGER.debug("Successfully reconnected"); + return; + } } + // stop() is called outside the lock, as a concurrent stop() holds the service lock while doStop() waits for it + LOGGER.warn("Connection lost and auto-reconnect is turned off, shutting down Plc4XEndpoint"); + stop(); } private void closeLostConnection() { @@ -195,14 +203,16 @@ private void closeLostConnection() { * @return true if connection supports writing, else false */ public boolean canWrite() { - return connection != null && connection.getMetadata().isWriteSupported(); + PlcConnection current = connection; + return current != null && current.getMetadata().isWriteSupported(); } /** * @return true if connection is established and still connected */ public boolean isConnected() { - return connection != null && connection.isConnected(); + PlcConnection current = connection; + return current != null && current.isConnected(); } @Override @@ -332,9 +342,11 @@ private String withoutEndpointOptions(String endpointUri) throws URISyntaxExcept @Override public void doStop() throws Exception { super.doStop(); - if (isConnected() && connection != null) { - connection.close(); - connection = null; + synchronized (this) { + if (isConnected()) { + connection.close(); + connection = null; + } } } From dc54abfa937ce4798765d1a2442e5fa0c3d2d554 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sat, 24 Oct 2026 12:04:47 +0300 Subject: [PATCH 07/10] CAMEL-25617: camel-plc4x - Fail a triggered poll on a non-boolean trigger or no answer A trigger that reads as a number or an array (PLC4X 1.0.0 answers an OPC UA or ADS array address without a selection with an array) was taken as false, so the consumer polled forever without delivering or logging anything. Such a poll now fails with the trigger address and the type that was read, as does a poll with no trigger value. The read of the trigger now waits at most as long as the producer waits for a write, and a read without an answer is cancelled and fails the poll. A failed poll forgets the trigger value and goes to the exception handler of the consumer. Co-authored-by: Claude Opus 5.5 --- .../camel/catalog/docs/plc4x-component.adoc | 6 ++++-- .../src/main/docs/plc4x-component.adoc | 6 ++++-- .../camel/component/plc4x/Plc4XEndpoint.java | 3 +++ .../camel/component/plc4x/Plc4XProducer.java | 2 +- .../plc4x/Plc4XTriggeredConsumer.java | 21 ++++++++++++++++--- .../plc4x/Plc4XConsumerRouteTest.java | 17 +++++++++++++++ 6 files changed, 47 insertions(+), 8 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc index af51b26ec8841..37f10570ac510 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc @@ -98,8 +98,10 @@ with `trigger` and must be positive. On a rising edge of the trigger, from `fals as the trigger. The trigger tag is not part of the body; it is read under the reserved name `TRIGGER_VAR`, which no tag may use. While the trigger stays `true`, no further exchange is delivered. -A poll that fails, for example because a tag cannot be read or the PLC cannot be reached, is reported to the exception -handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling continues. +A poll that fails, for example because a tag cannot be read, the PLC cannot be reached or does not answer within five +seconds, or the trigger reads as a value that is not a boolean (such as a number or an array), is reported to the +exception handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling +continues. The trigger value read before the failure is forgotten, so a trigger that is `true` at the first successful poll after a failure counts as a rising edge. Automatic recovery after a lost connection needs `autoReconnect=true`: a poll then replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on the first lost diff --git a/components/camel-plc4x/src/main/docs/plc4x-component.adoc b/components/camel-plc4x/src/main/docs/plc4x-component.adoc index af51b26ec8841..37f10570ac510 100644 --- a/components/camel-plc4x/src/main/docs/plc4x-component.adoc +++ b/components/camel-plc4x/src/main/docs/plc4x-component.adoc @@ -98,8 +98,10 @@ with `trigger` and must be positive. On a rising edge of the trigger, from `fals as the trigger. The trigger tag is not part of the body; it is read under the reserved name `TRIGGER_VAR`, which no tag may use. While the trigger stays `true`, no further exchange is delivered. -A poll that fails, for example because a tag cannot be read or the PLC cannot be reached, is reported to the exception -handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling continues. +A poll that fails, for example because a tag cannot be read, the PLC cannot be reached or does not answer within five +seconds, or the trigger reads as a value that is not a boolean (such as a number or an array), is reported to the +exception handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling +continues. The trigger value read before the failure is forgotten, so a trigger that is `true` at the first successful poll after a failure counts as a rising edge. Automatic recovery after a lost connection needs `autoReconnect=true`: a poll then replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on the first lost diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java index 2e9f174be743f..edf30b9819584 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java @@ -67,6 +67,9 @@ public class Plc4XEndpoint extends DefaultEndpoint implements EndpointServiceLocation { private static final Logger LOGGER = LoggerFactory.getLogger(Plc4XEndpoint.class); + // how long the producer and the triggered consumer wait for the PLC to answer a request + static final long REQUEST_TIMEOUT_MILLIS = 5000; + protected DefaultPlcDriverManager plcDriverManager; protected volatile PlcConnection connection; diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java index 6b82e831ead0e..13719137f77e1 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java @@ -83,7 +83,7 @@ public void process(Exchange exchange) throws Exception { int currentlyOpenRequests = openRequests.incrementAndGet(); try { log.debug("Currently open requests including {}:{}", exchange, currentlyOpenRequests); - Object plcWriteResponse = completableFuture.get(5000, TimeUnit.MILLISECONDS); + Object plcWriteResponse = completableFuture.get(Plc4XEndpoint.REQUEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); if (exchange.getPattern().isOutCapable()) { Message out = exchange.getMessage(); out.copyFrom(exchange.getIn()); diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java index 6852aecbb6672..5388f506ce5f0 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java @@ -17,8 +17,10 @@ package org.apache.camel.component.plc4x; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.apache.camel.Exchange; import org.apache.camel.Processor; @@ -85,7 +87,7 @@ protected int poll() throws Exception { // the consumer stopped while the read was in progress return 0; } - boolean triggered = Boolean.TRUE.equals(response.getObject(Constants.TRIGGER)); + boolean triggered = (Boolean) response.getObject(Constants.TRIGGER); boolean risingEdge = triggered && !Boolean.TRUE.equals(lastTrigger); lastTrigger = triggered; if (!risingEdge) { @@ -110,18 +112,25 @@ protected int poll() throws Exception { } /** - * Reads the trigger and the tags in one request; fails unless every tag was read. + * Reads the trigger and the tags in one request; fails unless every tag was read and the trigger is a boolean. */ private PlcReadResponse read() throws Exception { plc4XEndpoint.reconnectIfNeeded(); if (plc4XEndpoint.connection == null) { throw new PlcConnectionException("Cannot establish connection"); } + CompletableFuture future + = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER, trigger)).execute(); PlcReadResponse response; try { - response = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER, trigger)).execute().get(); + response = future.get(Plc4XEndpoint.REQUEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); } catch (ExecutionException e) { throw e.getCause() instanceof Exception cause ? cause : e; + } catch (TimeoutException e) { + future.cancel(true); + throw new TimeoutException( + "No response to the read of the trigger " + trigger + " within " + + Plc4XEndpoint.REQUEST_TIMEOUT_MILLIS + " millis"); } for (String tag : response.getTagNames()) { PlcResponseCode code = response.getResponseCode(tag); @@ -129,6 +138,12 @@ private PlcReadResponse read() throws Exception { throw new PlcException("Reading the tag " + tag + " failed with response code " + code); } } + Object value = response.getObject(Constants.TRIGGER); + if (!(value instanceof Boolean)) { + throw new PlcException( + "The trigger " + trigger + " must read as a Boolean, but read " + + (value == null ? "no value" : "a " + value.getClass().getName())); + } return response; } diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java index cde895e1e79d0..4ecf8dabe4094 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java @@ -251,6 +251,23 @@ public void aTagReadWithAnErrorCodeFailsThePoll() throws Exception { () -> "Unexpected cause: " + pollFailures.causes().get(0)); } + @Test + public void aTriggerThatIsNotABooleanFailsThePoll() throws Exception { + MockDriver.programRead("TRIG", 1, true); + MockDriver.programRead("TEMP", 21); + MockDriver.programRead("PRES", 1013); + MockEndpoint result = getMockEndpoint("mock:reported"); + result.expectedMessageCount(1); + result.setAssertPeriod(1000); + + context.getRouteController().startRoute("reported"); + + result.assertIsSatisfied(10000); + assertEquals(1, pollFailures.count(), () -> "Unexpected failures: " + pollFailures.causes()); + String message = pollFailures.causes().get(0).getMessage(); + assertTrue(message.contains("TRIG") && message.contains(Integer.class.getName()), message); + } + @Test public void aReadThatCompletesWhileTheRouteStopsIsNotDelivered() throws Exception { MockDriver.programRead("TRIG", true); From 674875603080b1e7c7d3b19403371e0d4473de2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sat, 24 Oct 2026 12:04:47 +0300 Subject: [PATCH 08/10] CAMEL-25617: camel-plc4x - Document the array notation, s7-light and a failed start The upgrade guide now names the PLC4X 1.0.0 array notation of tag addresses, the removal of the s7-light driver, that a triggered route that cannot connect on start also fails the start of the CamelContext unless it is supervised or not started automatically, and that a non-boolean trigger fails the poll. Co-authored-by: Claude Opus 5.5 --- .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index d47621a1c80d0..ff0a3a7a7a923 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -3595,6 +3595,15 @@ password are refused over a channel that is neither signed nor encrypted unless See the incompatible changes in the https://github.com/apache/plc4x/releases/tag/v1.0.0[PLC4X 1.0.0 release notes] for the full list. +PLC4X 1.0.0 writes the array selection of a tag address before the data type, and as a range instead of a count. An +address written as `holding-register:1:INT[4]` or `%DB42:28.0:BYTE[4]` for four elements is now written +`holding-register:1[0..3]:INT` or `%DB42:28.0[0..3]:BYTE`; `[4]` before the data type selects the single element at +index 4. Addresses in the old form no longer parse. This applies to the `tag.*` options, to `trigger` and to the +addresses in the body of the producer. + +The S7 driver of PLC4X 1.0.0 no longer includes the `s7-light` driver, so `plc4x:s7-light://` endpoints no longer find +a driver. Use `plc4x:s7://` instead. + A consumer with `trigger` now reads the trigger tag and the tags in one request on the connection of the endpoint, every `period` milliseconds from the start of the route. Before, it sampled the trigger first after about one second, and read the tags in a later, separate request on another connection. A failed read is now reported to the exception @@ -3602,9 +3611,11 @@ handler of the consumer (or to the error handler of the route with `bridgeErrorH before, the consumer stopped triggering after the first failed read of the trigger. After a failed poll the remembered trigger value is unknown, so a trigger that is still `true` when the PLC returns emits once. Since the consumer now uses the connection of the endpoint, `autoReconnect` governs its reconnects. A connection that cannot be set up when the -route starts fails the start of the route, unless `autoReconnect=true`, in which case the polls retry it. As before, -`period` must be positive when `trigger` is set. See the xref:components::plc4x-component.adoc[PLC4X component] for how -triggered reading works. +route starts fails the start of the route, unless `autoReconnect=true`, in which case the polls retry it. A route that +starts with the CamelContext then fails the start of the CamelContext too, unless the route is supervised by the +xref:route-controller.adoc[supervising route controller] or has `autoStartup=false`. As before, `period` must be +positive when `trigger` is set. A trigger that reads as a value that is not a boolean, such as a number or an array, +fails the poll. See the xref:components::plc4x-component.adoc[PLC4X component] for how triggered reading works. With `autoReconnect=true`, a lost connection is now closed and replaced with a new one from the driver instead of being reconnected, because the PLC4X 1.0.0 S7 and EtherNet/IP drivers cannot connect a connection again once it is lost. This From afa62bc19ec639f4760cdba95fa275c8e64166c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Mon, 26 Oct 2026 13:25:03 +0300 Subject: [PATCH 09/10] CAMEL-25617: camel-plc4x - Keep a live connection and close a lost one on stop Every start of a consumer or producer set up a new connection and overwrote the field without closing the old one. An endpoint is not stopped with its route, so each route restart, and a producer and a consumer on the same endpoint, left an open PLC connection behind. A start now keeps a connection that is still connected and closes and replaces only a lost one. Stopping the endpoint closed only a connected connection, so a dropped one was never closed or cleared; it now closes and clears either, and a failing close no longer fails the stop. Building a read request now fails with "Cannot establish connection" when there is no connection, instead of a NullPointerException when the connection is closed between the check of a consumer and the read. Co-authored-by: Claude Opus 5.5 --- .../camel/component/plc4x/Plc4XConsumer.java | 3 -- .../camel/component/plc4x/Plc4XEndpoint.java | 33 +++++++++++-------- .../component/plc4x/Plc4XPollingConsumer.java | 4 --- .../plc4x/Plc4XTriggeredConsumer.java | 4 --- .../plc4x/Plc4XConsumerRouteTest.java | 1 + .../component/plc4x/Plc4XEndpointTest.java | 18 ++++++++-- .../component/plc4x/Plc4XFailureTest.java | 4 --- 7 files changed, 36 insertions(+), 31 deletions(-) diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java index e3865b5e84287..bb4d8dbe76861 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java @@ -72,9 +72,6 @@ private void startUnTriggered() { PlcReadRequest request; try { plc4XEndpoint.reconnectIfNeeded(); - if (plc4XEndpoint.connection == null) { - throw new PlcConnectionException("Cannot establish connection"); - } request = plc4XEndpoint.buildPlcReadRequest(); } catch (Exception e) { if (LOGGER.isTraceEnabled()) { diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java index edf30b9819584..ad24c3f70a15e 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java @@ -145,7 +145,11 @@ public boolean isAutoReconnect() { public void setupConnection() throws PlcConnectionException { try { synchronized (this) { - connection = plcDriverManager.getConnection(connectionString); + // a restarted route or another consumer or producer of the endpoint keeps a live connection + if (!isConnected()) { + closeConnection(); + connection = plcDriverManager.getConnection(connectionString); + } } if (!isConnected()) { reconnectIfNeeded(); @@ -180,7 +184,7 @@ public void reconnectIfNeeded() throws PlcConnectionException { if (autoReconnect) { // a lost PLC4X connection cannot always connect again (the S7 and EtherNet/IP drivers fail on a closed // transport), so it is replaced by a new one - closeLostConnection(); + closeConnection(); connection = plcDriverManager.getConnection(connectionString); LOGGER.debug("Successfully reconnected"); return; @@ -191,12 +195,12 @@ public void reconnectIfNeeded() throws PlcConnectionException { stop(); } - private void closeLostConnection() { + private void closeConnection() { if (connection != null) { try { connection.close(); } catch (Exception e) { - LOGGER.debug("Closing the lost connection failed: {}", e.getMessage()); + LOGGER.debug("Closing the connection failed: {}", e.getMessage()); } connection = null; } @@ -239,18 +243,24 @@ public PollingConsumer createPollingConsumer() { /** * Build a {@link PlcReadRequest} using the tags specified in the endpoint. */ - public PlcReadRequest buildPlcReadRequest() { + public PlcReadRequest buildPlcReadRequest() throws PlcConnectionException { return buildPlcReadRequest(Map.of()); } /** * Build a {@link PlcReadRequest} using the given tags and the tags specified in the endpoint. * - * @param additionalTags tag addresses by name to read in addition to the tags of the endpoint - * @return {@link PlcReadRequest} + * @param additionalTags tag addresses by name to read in addition to the tags of the endpoint + * @return {@link PlcReadRequest} + * @throws PlcConnectionException if there is no connection */ - PlcReadRequest buildPlcReadRequest(Map additionalTags) { - PlcReadRequest.Builder builder = connection.readRequestBuilder(); + PlcReadRequest buildPlcReadRequest(Map additionalTags) throws PlcConnectionException { + // read the field once: a concurrent stop or reconnect can clear it at any time + PlcConnection current = connection; + if (current == null) { + throw new PlcConnectionException("Cannot establish connection"); + } + PlcReadRequest.Builder builder = current.readRequestBuilder(); additionalTags.forEach(builder::addTagAddress); if (tags != null) { for (Map.Entry tag : tags.entrySet()) { @@ -346,10 +356,7 @@ private String withoutEndpointOptions(String endpointUri) throws URISyntaxExcept public void doStop() throws Exception { super.doStop(); synchronized (this) { - if (isConnected()) { - connection.close(); - connection = null; - } + closeConnection(); } } diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java index dd063bfa8a1f7..b35c2d306e6ea 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java @@ -84,10 +84,6 @@ protected Exchange doReceive(long timeout) { Exchange exchange = plc4XEndpoint.createExchange(); try { plc4XEndpoint.reconnectIfNeeded(); - if (plc4XEndpoint.connection == null) { - throw new PlcConnectionException("Cannot establish connection"); - } - PlcReadRequest request = plc4XEndpoint.buildPlcReadRequest(); CompletableFuture future = request.execute().whenComplete((plcReadResponse, throwable) -> { diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java index 5388f506ce5f0..a742ac2a878b2 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java @@ -25,7 +25,6 @@ import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.support.ScheduledPollConsumer; -import org.apache.plc4x.java.api.exceptions.PlcConnectionException; import org.apache.plc4x.java.api.exceptions.PlcException; import org.apache.plc4x.java.api.messages.PlcReadResponse; import org.apache.plc4x.java.api.types.PlcResponseCode; @@ -116,9 +115,6 @@ protected int poll() throws Exception { */ private PlcReadResponse read() throws Exception { plc4XEndpoint.reconnectIfNeeded(); - if (plc4XEndpoint.connection == null) { - throw new PlcConnectionException("Cannot establish connection"); - } CompletableFuture future = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER, trigger)).execute(); PlcReadResponse response; diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java index 4ecf8dabe4094..6a94252023fda 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java @@ -133,6 +133,7 @@ public void stopsReadingWhenTheRouteStopsAndResumesWithoutDuplicatesAfterARestar context.getRouteController().startRoute("triggered"); result.assertIsSatisfied(10000); + assertEquals(1, MockDriver.connectionStrings().size(), "the restarted route uses the connection that is still open"); } @Test diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java index 24cdd40c11d43..7703fb6384e11 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java @@ -20,12 +20,15 @@ import org.apache.camel.Processor; import org.apache.camel.impl.DefaultCamelContext; import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.exceptions.PlcConnectionException; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.core.Is.is; import static org.hamcrest.core.IsNull.notNullValue; +import static org.hamcrest.core.IsNull.nullValue; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.*; class Plc4XEndpointTest { @@ -56,12 +59,21 @@ public void isSingleton() { } @Test - void doStopDisconnectedConnection() throws Exception { + void doStopClosesALostConnection() throws Exception { PlcConnection plcConnectionMock = mock(PlcConnection.class); + // isConnected() returns false on the mock, and close() fails as on a dead transport + doThrow(new PlcConnectionException("Transport is not open")).when(plcConnectionMock).close(); sut.connection = plcConnectionMock; - // isConnected() returns false by default on the mock, so doStop() should skip close() + sut.doStop(); - verify(plcConnectionMock, never()).close(); + + verify(plcConnectionMock).close(); + assertThat(sut.connection, nullValue()); + } + + @Test + void buildPlcReadRequestWithoutAConnection() { + assertThrows(PlcConnectionException.class, () -> sut.buildPlcReadRequest()); } } diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java index 54ac03fe2253a..e1fd611d8f821 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java @@ -24,7 +24,6 @@ import org.apache.camel.Exchange; import org.apache.camel.impl.DefaultCamelContext; import org.apache.camel.support.DefaultExchange; -import org.apache.plc4x.java.api.PlcConnection; import org.apache.plc4x.java.api.exceptions.PlcConnectionException; import org.apache.plc4x.java.api.exceptions.PlcRuntimeException; import org.apache.plc4x.java.api.messages.PlcReadRequest; @@ -111,7 +110,6 @@ void testPollingConsumerFailsWhenReadFails() throws Exception { PlcRuntimeException cause = new PlcRuntimeException("read failed"); PlcReadRequest request = mock(PlcReadRequest.class); when(request.execute()).thenAnswer(invocation -> CompletableFuture.failedFuture(cause)); - endpoint.connection = mock(PlcConnection.class); when(endpoint.buildPlcReadRequest()).thenReturn(request); Exchange exchange = new Plc4XPollingConsumer(endpoint).receive(1000); @@ -124,7 +122,6 @@ void testPollingConsumerFailsWhenReadFails() throws Exception { void testPollingConsumerReturnsNullWithoutResponse() throws Exception { PlcReadRequest request = mock(PlcReadRequest.class); when(request.execute()).thenAnswer(invocation -> new CompletableFuture<>()); - endpoint.connection = mock(PlcConnection.class); when(endpoint.buildPlcReadRequest()).thenReturn(request); // receiveNoWait and receive(timeout) return null when nothing is received in time @@ -134,7 +131,6 @@ void testPollingConsumerReturnsNullWithoutResponse() throws Exception { @Test void testPollingConsumerReadsTags() throws Exception { - endpoint.connection = mock(PlcConnection.class); PlcReadRequest request = mock(PlcReadRequest.class, RETURNS_DEEP_STUBS); when(endpoint.buildPlcReadRequest()).thenReturn(request); From a214ef9ebc1ae1fdbd441c5ab74ecc4c3303be29 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Mon, 26 Oct 2026 14:41:55 +0300 Subject: [PATCH 10/10] CAMEL-25617: camel-plc4x - Fail a write without a connection instead of a NullPointerException Building a write request read the connection field without a check, so a write after the endpoint was stopped, or while its connection was being closed, failed the exchange with a NullPointerException. It now reads the field once and fails with "Cannot establish connection" when there is no connection, as building a read request does. Co-authored-by: Claude Opus 5.5 --- .../camel/component/plc4x/Plc4XEndpoint.java | 14 ++++++++++---- .../camel/component/plc4x/Plc4XEndpointTest.java | 7 +++++++ 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java index ad24c3f70a15e..c541fb8998018 100644 --- a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java +++ b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java @@ -290,11 +290,17 @@ static Map tagValues(PlcReadResponse response) { * Build a {@link PlcWriteRequest}. *

* - * @param tags tags to add to write request - * @return {@link PlcWriteRequest} + * @param tags tags to add to write request + * @return {@link PlcWriteRequest} + * @throws PlcConnectionException if there is no connection */ - public PlcWriteRequest buildPlcWriteRequest(Map> tags) { - PlcWriteRequest.Builder builder = connection.writeRequestBuilder(); + public PlcWriteRequest buildPlcWriteRequest(Map> tags) throws PlcConnectionException { + // read the field once: a concurrent stop or reconnect can clear it at any time + PlcConnection current = connection; + if (current == null) { + throw new PlcConnectionException("Cannot establish connection"); + } + PlcWriteRequest.Builder builder = current.writeRequestBuilder(); for (Map.Entry> entry : tags.entrySet()) { //Tags are stored like this --> Map> for writing diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java index 7703fb6384e11..28df776b2c83f 100644 --- a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java @@ -16,6 +16,8 @@ */ package org.apache.camel.component.plc4x; +import java.util.Map; + import org.apache.camel.Component; import org.apache.camel.Processor; import org.apache.camel.impl.DefaultCamelContext; @@ -76,4 +78,9 @@ void buildPlcReadRequestWithoutAConnection() { assertThrows(PlcConnectionException.class, () -> sut.buildPlcReadRequest()); } + @Test + void buildPlcWriteRequestWithoutAConnection() { + assertThrows(PlcConnectionException.class, () -> sut.buildPlcWriteRequest(Map.of())); + } + }