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 37f10570ac510..c273c34409d3b 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 @@ -101,12 +101,14 @@ may use. While the trigger stays `true`, no further exchange is delivered. 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. +continues. The default exception handler logs the first failed poll of a run at WARN with the stack trace and the +following ones at DEBUG without it; the first successful poll after a run of failed polls is logged at INFO. 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. +replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on each lost +connection, and the consumer fails that poll and stops polling until the route is restarted. 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: diff --git a/components/camel-plc4x/src/main/docs/plc4x-component.adoc b/components/camel-plc4x/src/main/docs/plc4x-component.adoc index 37f10570ac510..c273c34409d3b 100644 --- a/components/camel-plc4x/src/main/docs/plc4x-component.adoc +++ b/components/camel-plc4x/src/main/docs/plc4x-component.adoc @@ -101,12 +101,14 @@ may use. While the trigger stays `true`, no further exchange is delivered. 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. +continues. The default exception handler logs the first failed poll of a run at WARN with the stack trace and the +following ones at DEBUG without it; the first successful poll after a run of failed polls is logged at INFO. 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. +replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on each lost +connection, and the consumer fails that poll and stops polling until the route is restarted. 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: 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 c541fb8998018..2fd507d342a75 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 @@ -169,7 +169,8 @@ public void setupConnection() throws PlcConnectionException { } /** - * Reconnects if needed. If connection is lost and auto-reconnect is turned off, endpoint will be shutdown. + * Reconnects if needed. If connection is lost and auto-reconnect is turned off, endpoint will be shutdown, once for + * each lost connection. *

* * @throws PlcConnectionException If reconnect failed and auto-reconnect is turned on @@ -189,6 +190,12 @@ public void reconnectIfNeeded() throws PlcConnectionException { LOGGER.debug("Successfully reconnected"); return; } + if (connection == null) { + // no connection left to lose, as when an earlier call shut the lost one down or none could be set up + return; + } + // closed under the lock, so that only one caller shuts the endpoint down for the lost connection + closeConnection(); } // 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"); @@ -222,6 +229,13 @@ public boolean isConnected() { return current != null && current.isConnected(); } + /** + * @return false if the lost connection was shut down, or none was set up + */ + boolean hasConnection() { + return connection != null; + } + @Override public Producer createProducer() { return new Plc4XProducer(this); 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 a742ac2a878b2..09af4edcdb5d5 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 @@ -22,18 +22,25 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import org.apache.camel.CamelContext; import org.apache.camel.Exchange; import org.apache.camel.Processor; +import org.apache.camel.support.LoggingExceptionHandler; 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; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * 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 static final Logger LOGGER = LoggerFactory.getLogger(Plc4XTriggeredConsumer.class); + private final Plc4XEndpoint plc4XEndpoint; private final String trigger; private final int period; @@ -49,6 +56,8 @@ public Plc4XTriggeredConsumer(Plc4XEndpoint endpoint, Processor processor) { setInitialDelay(0); setDelay(period); setTimeUnit(TimeUnit.MILLISECONDS); + // the endpoint replaces it when an exceptionHandler or bridgeErrorHandler is configured + setExceptionHandler(new FailedPollExceptionHandler(endpoint.getCamelContext())); } @Override @@ -86,6 +95,10 @@ protected int poll() throws Exception { // the consumer stopped while the read was in progress return 0; } + if (getErrorCounter() > 0) { + // the error counter is reset after this poll, so it still counts the failed polls before it + LOGGER.info("Polling endpoint: {} succeeded again after {} failed polls", getEndpoint(), getErrorCounter()); + } boolean triggered = (Boolean) response.getObject(Constants.TRIGGER); boolean risingEdge = triggered && !Boolean.TRUE.equals(lastTrigger); lastTrigger = triggered; @@ -111,10 +124,22 @@ protected int poll() throws Exception { } /** - * Reads the trigger and the tags in one request; fails unless every tag was read and the trigger is a boolean. + * Reads the trigger and the tags in one request; fails unless every tag was read and the trigger is a boolean. A + * connection that is lost without autoReconnect also stops the polling. */ private PlcReadResponse read() throws Exception { plc4XEndpoint.reconnectIfNeeded(); + // a connection that is lost after reconnectIfNeeded() fails the read instead, and the next poll shuts it down + if (!plc4XEndpoint.isAutoReconnect() && !plc4XEndpoint.hasConnection()) { + // the endpoint shut the lost connection down, so no later poll could read: polling stops until the route is + // restarted, and this poll fails, so the health check of the consumer stays down + getScheduler().unscheduleTask(); + // unscheduling cancels the task this poll runs in, which interrupts this thread: clear it, so that the + // failure is reported as for any other poll + Thread.interrupted(); + throw new PlcConnectionException( + "Connection lost and auto-reconnect is turned off, polling stops until the route is restarted"); + } CompletableFuture future = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER, trigger)).execute(); PlcReadResponse response; @@ -147,4 +172,32 @@ private PlcReadResponse read() throws Exception { public String toString() { return "Plc4XTriggeredConsumer[" + plc4XEndpoint + "]"; } + + /** + * Logs like the default exception handler of a consumer, but a failed poll only at WARN with the stack trace when + * it starts a run of failed polls or stops the polling, and the following failed polls of the run at DEBUG without + * it, so that a PLC that cannot be reached is not logged on every poll. The failed poll that stops the polling says + * so, instead of announcing a next poll. + */ + private final class FailedPollExceptionHandler extends LoggingExceptionHandler { + FailedPollExceptionHandler(CamelContext camelContext) { + super(camelContext, Plc4XTriggeredConsumer.class); + } + + @Override + public void handleException(String message, Throwable exception) { + // a failed poll is reported without an exchange and before it is counted, so the first one of a run finds an + // error counter of 0, and the one that stops the polling finds its task unscheduled + if (!isSchedulerStarted()) { + super.handleException( + "Failed polling endpoint: " + getEndpoint() + ". Polling stops until the route is restarted", + exception); + } else if (getErrorCounter() == 0) { + super.handleException(message, exception); + } else { + // the first failed poll of the run is logged with the stack trace already + LOGGER.debug("{}. Caused by: {}", message, exception.getMessage()); + } + } + } } diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/LogCapture.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/LogCapture.java new file mode 100644 index 0000000000000..d195ec71c720b --- /dev/null +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/LogCapture.java @@ -0,0 +1,73 @@ +/* + * 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.concurrent.CopyOnWriteArrayList; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; + +/** + * Captures the log events of one logger, down to DEBUG, while it is open. Only one capture is open at a time: opening + * another one resets the level of the loggers. + */ +final class LogCapture extends AbstractAppender implements AutoCloseable { + + private final List events = new CopyOnWriteArrayList<>(); + private final Logger logger; + private final Level previousLevel; + + LogCapture(Class loggerClass) { + super("LogCapture-" + loggerClass.getSimpleName(), null, null, true, Property.EMPTY_ARRAY); + this.logger = (Logger) LogManager.getLogger(loggerClass); + this.previousLevel = logger.getLevel(); + start(); + logger.addAppender(this); + logger.setLevel(Level.DEBUG); + } + + @Override + public void append(LogEvent event) { + events.add(event.toImmutable()); + } + + /** + * @return the events logged so far, in the order they were logged + */ + List events() { + return List.copyOf(events); + } + + /** + * @return the events logged so far whose message contains the given text + */ + List eventsContaining(String text) { + return events.stream().filter(e -> e.getMessage().getFormattedMessage().contains(text)).toList(); + } + + @Override + public void close() { + logger.removeAppender(this); + logger.setLevel(previousLevel); + stop(); + } +} 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 6a94252023fda..6ad5fff4ecd99 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 @@ -19,13 +19,10 @@ import java.util.List; import java.util.Map; import java.util.Set; -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.plc4x.java.api.exceptions.PlcConnectionException; @@ -363,35 +360,4 @@ private static Throwable causeOfType(Throwable exception, Class causes = new CopyOnWriteArrayList<>(); - - @Override - public void handleException(Throwable exception) { - causes.add(exception); - } - - @Override - public void handleException(String message, Throwable exception) { - causes.add(exception); - } - - @Override - public void handleException(String message, Exchange exchange, Throwable exception) { - causes.add(exception); - } - - int count() { - return causes.size(); - } - - List causes() { - return causes; - } - } - } diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java new file mode 100644 index 0000000000000..4e897e9762a3f --- /dev/null +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java @@ -0,0 +1,217 @@ +/* + * 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.TimeUnit; + +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +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.plc4x.java.api.exceptions.PlcConnectionException; +import org.apache.plc4x.java.api.exceptions.PlcException; +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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * How a route reports a PLC that keeps failing its requests: the triggered consumer logs the first failed poll of a run + * at WARN and the end of the run at INFO, instead of a WARN with a stack trace on every poll, while + * {@code bridgeErrorHandler} still gets every failure. Without {@code autoReconnect}, a lost connection shuts the + * endpoint down once, and the triggered consumer stops polling until the route is restarted. + */ +public class Plc4XRepeatedFailureRouteTest extends CamelTestSupport { + + private static final String PLC = "plc4x:mock:10.10.10.1/1/1?trigger=TRIG&period=50&tag.temperature=TEMP"; + private static final String BRIDGED = PLC + "&bridgeErrorHandler=true"; + private static final String REPORTED = PLC + "&exceptionHandler=#pollFailures"; + private static final String WRITE = "plc4x:mock:10.10.10.2/1/1"; + private static final String SHUTDOWN = "Connection lost and auto-reconnect is turned off"; + private static final String POLLING_STOPS = "polling stops until the route is restarted"; + + private final PollFailures pollFailures = new PollFailures(); + + @BeforeEach + void resetMockDriver() { + MockDriver.reset(); + } + + @AfterAll + static void leaveMockDriverUnprogrammed() { + MockDriver.reset(); + } + + @Test + public void logsARunOfFailedPollsOnceAtWarnAndItsEndOnceAtInfo() throws Exception { + Object failed = MockDriver.failedRead(); + // two runs of failed polls, each followed by successful polls, and a rising edge after them + MockDriver.programRead("TRIG", failed, failed, failed, false, false, false, failed, failed, false, false, true); + MockDriver.programRead("TEMP", 21); + MockEndpoint result = getMockEndpoint("mock:logged"); + result.expectedMessageCount(1); + + try (LogCapture log = new LogCapture(Plc4XTriggeredConsumer.class)) { + context.getRouteController().startRoute("logged"); + result.assertIsSatisfied(10000); + + List events = log.events(); + assertEquals(List.of(Level.WARN, Level.DEBUG, Level.DEBUG, Level.INFO, Level.WARN, Level.DEBUG, Level.INFO), + events.stream().map(LogEvent::getLevel).toList(), () -> "Unexpected log: " + messages(events)); + assertEquals(5, log.eventsContaining("Failed polling endpoint").size(), + () -> "Unexpected log: " + messages(events)); + assertNotNull(events.get(0).getThrown(), "the first failed poll of a run is logged with its stack trace"); + assertNotNull(events.get(4).getThrown(), "the first failed poll of a run is logged with its stack trace"); + assertNull(events.get(1).getThrown(), "a following failed poll of a run is logged without its stack trace"); + // the error counter of the consumer still counts each failed poll of a run + String firstEnd = events.get(3).getMessage().getFormattedMessage(); + assertTrue(firstEnd.endsWith("after 3 failed polls"), firstEnd); + String secondEnd = events.get(6).getMessage().getFormattedMessage(); + assertTrue(secondEnd.endsWith("after 2 failed polls"), secondEnd); + } + } + + @Test + public void bridgeErrorHandlerRoutesEveryFailedPollToTheErrorHandlerOfTheRoute() throws Exception { + Object failed = MockDriver.failedRead(); + MockDriver.programRead("TRIG", failed, failed, failed, false); + MockDriver.programRead("TEMP", 21); + MockEndpoint failures = getMockEndpoint("mock:failures"); + failures.expectedMessageCount(3); + failures.setAssertPeriod(500); + + context.getRouteController().startRoute("bridged"); + + failures.assertIsSatisfied(10000); + assertTrue(failures.getExchanges().stream() + .allMatch(exchange -> exchange.getProperty(Exchange.EXCEPTION_CAUGHT) instanceof PlcException), + () -> "Unexpected failures: " + failures.getExchanges()); + } + + @Test + public void aLostConnectionWithoutAutoReconnectStopsPollingUntilTheRouteIsRestarted() throws Exception { + MockDriver.programRead("TRIG", false); + MockDriver.programRead("TEMP", 21); + MockEndpoint result = getMockEndpoint("mock:reported"); + + try (LogCapture log = new LogCapture(Plc4XEndpoint.class)) { + context.getRouteController().startRoute("reported"); + Plc4XTriggeredConsumer consumer = (Plc4XTriggeredConsumer) context.getRoute("reported").getConsumer(); + await().atMost(10, TimeUnit.SECONDS).until(() -> consumer.getSuccessCounter() >= 2); + + MockDriver.dropConnections(); + + // the poll that finds the connection lost fails, and no poll follows it + await().atMost(10, TimeUnit.SECONDS).until(() -> !pollFailures.failures().isEmpty()); + await().during(500, TimeUnit.MILLISECONDS).atMost(1, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1, pollFailures.failures().size(), "the failed polls")); + assertEquals(1, log.eventsContaining(SHUTDOWN).size(), "the endpoint is shut down once"); + assertEquals(1, consumer.getErrorCounter(), "the failed poll keeps the health check of the consumer down"); + PollFailures.Failure failure = pollFailures.failures().get(0); + assertInstanceOf(PlcConnectionException.class, failure.cause()); + assertFalse(failure.interrupted(), "the failure is reported on a thread that is not interrupted"); + + // the restarted route sets up a new connection and polls again + MockDriver.programRead("TRIG", false, true); + result.expectedMessageCount(1); + context.getRouteController().stopRoute("reported"); + context.getRouteController().startRoute("reported"); + result.assertIsSatisfied(10000); + + // and a loss of the new connection shuts the endpoint down again + MockDriver.dropConnections(); + await().atMost(10, TimeUnit.SECONDS).until(() -> log.eventsContaining(SHUTDOWN).size() == 2); + await().atMost(10, TimeUnit.SECONDS).until(() -> pollFailures.failures().size() == 2); + } + } + + @Test + public void logsTheFailedPollThatStopsPollingAtWarnAfterAFailedRead() throws Exception { + MockDriver.programRead("TRIG", false, MockDriver.failedRead()); + MockDriver.programRead("TEMP", 21); + + try (LogCapture log = new LogCapture(Plc4XTriggeredConsumer.class)) { + context.getRouteController().startRoute("logged"); + Plc4XTriggeredConsumer consumer = (Plc4XTriggeredConsumer) context.getRoute("logged").getConsumer(); + // a read fails as the PLC goes away, and a later poll finds the connection lost within the same run + await().atMost(10, TimeUnit.SECONDS).until(() -> consumer.getErrorCounter() >= 1); + + MockDriver.dropConnections(); + + await().atMost(10, TimeUnit.SECONDS).until(() -> !log.eventsContaining(POLLING_STOPS).isEmpty()); + assertEquals(List.of(Level.WARN), + log.eventsContaining(POLLING_STOPS).stream().map(LogEvent::getLevel).toList(), + () -> "Unexpected log: " + messages(log.events())); + // the log line of the poll says that no next poll follows it + String stopped = log.eventsContaining(POLLING_STOPS).get(0).getMessage().getFormattedMessage(); + assertTrue(stopped.startsWith( + "Failed polling endpoint: " + consumer.getEndpoint() + ". Polling stops until the route is restarted"), + stopped); + } + } + + @Test + public void aProducerReportsALostConnectionWithoutAutoReconnectOnce() throws Exception { + Map> tags = Map.of("temperature", Map.of("%DB1:DINT", 21)); + template.sendBody(WRITE, tags); + + try (LogCapture log = new LogCapture(Plc4XEndpoint.class)) { + MockDriver.dropConnections(); + + for (int i = 0; i < 3; i++) { + Exchange exchange = template.send(WRITE, e -> e.getIn().setBody(tags)); + assertInstanceOf(PlcConnectionException.class, exchange.getException(), "each write fails"); + } + assertEquals(1, log.eventsContaining(SHUTDOWN).size(), "the endpoint is shut down once"); + } + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from(PLC).routeId("logged").autoStartup(false).to("mock:logged"); + from(BRIDGED).routeId("bridged").autoStartup(false) + .onException(Exception.class).handled(true).to("mock:failures").end() + .to("mock:bridged"); + from(REPORTED).routeId("reported").autoStartup(false).to("mock:reported"); + } + }; + } + + @Override + protected void bindToRegistry(Registry registry) { + registry.bind("pollFailures", pollFailures); + } + + private static List messages(List events) { + return events.stream().map(event -> event.getLevel() + " " + event.getMessage().getFormattedMessage()).toList(); + } +} diff --git a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/PollFailures.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/PollFailures.java new file mode 100644 index 0000000000000..d6abf39f764bf --- /dev/null +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/PollFailures.java @@ -0,0 +1,62 @@ +/* + * 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.concurrent.CopyOnWriteArrayList; + +import org.apache.camel.Exchange; +import org.apache.camel.spi.ExceptionHandler; + +/** + * Collects the failures a consumer reports to its exception handler, and whether the thread that reported each one was + * interrupted. + */ +final class PollFailures implements ExceptionHandler { + + private final List failures = new CopyOnWriteArrayList<>(); + + @Override + public void handleException(Throwable exception) { + handleException(null, null, exception); + } + + @Override + public void handleException(String message, Throwable exception) { + handleException(message, null, exception); + } + + @Override + public void handleException(String message, Exchange exchange, Throwable exception) { + failures.add(new Failure(exception, Thread.currentThread().isInterrupted())); + } + + List failures() { + return failures; + } + + int count() { + return failures.size(); + } + + List causes() { + return failures.stream().map(Failure::cause).toList(); + } + + record Failure(Throwable cause, boolean interrupted) { + } +} 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 0be69c592ec19..43be76d680975 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 @@ -3628,7 +3628,8 @@ and read the tags in a later, separate request on another connection. A failed r 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 +the connection of the endpoint, `autoReconnect` governs its reconnects: without it, a lost connection stops +the endpoint and the polling until the route is restarted. 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. 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