gnodet-bot commented on code in PR #28387:
URL: https://github.com/apache/camel/pull/28387#discussion_r4374411619
##########
components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java:
##########
@@ -133,7 +144,9 @@ public boolean isAutoReconnect() {
*/
public void setupConnection() throws PlcConnectionException {
try {
- connection = plcDriverManager.getConnection(this.uri);
+ synchronized (this) {
Review Comment:
⚠️ **Connection leak on re-setup** (follow-up, non-blocking):
`setupConnection()` replaces `connection` without first calling
`closeLostConnection()`. If it is called while a prior connection handle exists
(e.g. after a transport failure before `reconnectIfNeeded` had a chance to
run), the old `PlcConnection` is orphaned and never closed.
`reconnectIfNeeded()` handles this correctly — it calls
`closeLostConnection()` inside the lock before reassigning. `setupConnection()`
should do the same:
```suggestion
synchronized (this) {
closeLostConnection();
connection =
plcDriverManager.getConnection(connectionString);
}
```
##########
components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.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;
+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<String, String> 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) response.getObject(Constants.TRIGGER);
+ boolean risingEdge = triggered && !Boolean.TRUE.equals(lastTrigger);
+ lastTrigger = triggered;
+ if (!risingEdge) {
+ return 0;
+ }
+
+ Map<String, Object> 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 and the trigger is a boolean.
+ */
+ private PlcReadResponse read() throws Exception {
+ plc4XEndpoint.reconnectIfNeeded();
+ if (plc4XEndpoint.connection == null) {
+ throw new PlcConnectionException("Cannot establish connection");
+ }
Review Comment:
⚠️ **TOCTOU: `plc4XEndpoint.connection` read outside any lock** (from prior
review thread `PRRC_kwDOAAMl7c8AAAABBEZK2A`): `reconnectIfNeeded()` updates
`connection` under `synchronized(plc4XEndpoint)`, but the null-check on line
121 and the subsequent `buildPlcReadRequest()` call (which reads `connection`
again on line 644) both happen without holding the lock.
A concurrent stop or reconnect can null or replace `connection` between the
check and the use, causing a `NullPointerException` on `buildPlcReadRequest()`.
Suggested fix — snapshot the volatile field once under the lock:
```java
PlcConnection conn;
synchronized (plc4XEndpoint) {
conn = plc4XEndpoint.connection;
}
if (conn == null) {
throw new PlcConnectionException("Cannot establish connection");
}
CompletableFuture<? extends PlcReadResponse> future
= plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER,
trigger)).execute();
```
(Or expose a synchronized accessor on `Plc4XEndpoint` rather than accessing
the field directly.)
##########
components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java:
##########
@@ -271,16 +319,37 @@ public void setTags(Map<String, String> 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<String, Object> 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();
- if (isConnected() && connection != null) {
- connection.close();
- connection = null;
+ synchronized (this) {
+ if (isConnected()) {
Review Comment:
⚠️ **Dropped connection not cleaned up in `doStop()`** (follow-up,
non-blocking): the `isConnected()` guard means a connection whose transport has
dropped (field is non-null but `isConnected()` returns false) is never closed
or nulled here. The PLC driver may hold OS resources (sockets, file
descriptors) that will leak until GC.
Consider replacing with a null check, which is the correct criterion for "do
we have a handle to close":
```suggestion
if (connection != null) {
```
Alternatively, delegate to `closeLostConnection()` which already handles
this correctly.
##########
components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java:
##########
@@ -271,16 +319,37 @@ public void setTags(Map<String, String> 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<String, Object> 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();
- if (isConnected() && connection != null) {
- connection.close();
- connection = null;
+ synchronized (this) {
+ if (isConnected()) {
Review Comment:
⚠️ **Dropped connection not cleaned up in `doStop()`** (follow-up,
non-blocking): the `isConnected()` guard means a connection whose transport has
dropped (field is non-null but `isConnected()` returns false) is never closed
or nulled here. The PLC driver may hold OS resources (sockets, file
descriptors) that will leak until GC.
Consider replacing with a null check, which is the correct criterion for "do
we have a handle to close":
```suggestion
if (connection != null) {
```
Alternatively, delegate to `closeLostConnection()` which already handles
this correctly.
##########
components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java:
##########
@@ -133,7 +144,9 @@ public boolean isAutoReconnect() {
*/
public void setupConnection() throws PlcConnectionException {
try {
- connection = plcDriverManager.getConnection(this.uri);
+ synchronized (this) {
Review Comment:
⚠️ **Connection leak on re-setup** (follow-up, non-blocking):
`setupConnection()` replaces `connection` without first calling
`closeLostConnection()`. If it is called while a prior connection handle exists
(e.g. after a transport failure before `reconnectIfNeeded` had a chance to
run), the old `PlcConnection` is orphaned and never closed.
`reconnectIfNeeded()` handles this correctly — it calls
`closeLostConnection()` inside the lock before reassigning. `setupConnection()`
should do the same:
```suggestion
synchronized (this) {
closeLostConnection();
connection =
plcDriverManager.getConnection(connectionString);
}
```
##########
components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.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;
+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<String, String> 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) response.getObject(Constants.TRIGGER);
+ boolean risingEdge = triggered && !Boolean.TRUE.equals(lastTrigger);
+ lastTrigger = triggered;
+ if (!risingEdge) {
+ return 0;
+ }
+
+ Map<String, Object> 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 and the trigger is a boolean.
+ */
+ private PlcReadResponse read() throws Exception {
+ plc4XEndpoint.reconnectIfNeeded();
+ if (plc4XEndpoint.connection == null) {
+ throw new PlcConnectionException("Cannot establish connection");
+ }
Review Comment:
⚠️ **TOCTOU: `plc4XEndpoint.connection` read outside any lock** (from prior
review thread `PRRC_kwDOAAMl7c8AAAABBEZK2A`): `reconnectIfNeeded()` updates
`connection` under `synchronized(plc4XEndpoint)`, but the null-check on line
121 and the subsequent `buildPlcReadRequest()` call (which reads `connection`
again on line 644) both happen without holding the lock.
A concurrent stop or reconnect can null or replace `connection` between the
check and the use, causing a `NullPointerException` on `buildPlcReadRequest()`.
Suggested fix — snapshot the volatile field once under the lock:
```java
PlcConnection conn;
synchronized (plc4XEndpoint) {
conn = plc4XEndpoint.connection;
}
if (conn == null) {
throw new PlcConnectionException("Cannot establish connection");
}
CompletableFuture<? extends PlcReadResponse> future
= plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER,
trigger)).execute();
```
(Or expose a synchronized accessor on `Plc4XEndpoint` rather than accessing
the field directly.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
